src/txqueue.c

   1 /*
   2  *  aim_txqueue.c
   3  *
   4  * Herein lies all the mangement routines for the transmit (Tx) queue.
   5  *
   6  */
   7
   8 #define FAIM_INTERNAL
   9 #include <aim.h>
  10
  11 #ifndef _WIN32
  12 #include <sys/socket.h>
  13 #endif
  14
  15 /*
  16  * Allocate a new tx frame.
  17  *
  18  * This is more for looks than anything else.
  19  *
  20  * Right now, that is.  If/when we implement a pool of transmit
  21  * frames, this will become the request-an-unused-frame part.
  22  *
  23  * framing = AIM_FRAMETYPE_OFT/FLAP
  24  * chan = channel for FLAP, hdrtype for OFT
  25  *
  26  */
  27 faim_internal aim_frame_t *aim_tx_new(aim_session_t *sess, aim_conn_t *conn, fu8_t framing, fu8_t chan, int datalen)
  28 {
  29         aim_frame_t *fr;
  30
  31         if (!conn) {
  32                 faimdprintf(sess, 0, "aim_tx_new: ERROR: no connection specified\n");
  33                 return NULL;
  34         }
  35
  36         /* For sanity... */
  37         if ((conn->type == AIM_CONN_TYPE_RENDEZVOUS) ||
  38                         (conn->type == AIM_CONN_TYPE_RENDEZVOUS_OUT)) {
  39                 if (framing != AIM_FRAMETYPE_OFT) {
  40                         faimdprintf(sess, 0, "aim_tx_new: attempted to allocate inappropriate frame type for rendezvous connection\n");
  41                         return NULL;
  42                 }
  43         } else {
  44                 if (framing != AIM_FRAMETYPE_FLAP) {
  45                         faimdprintf(sess, 0, "aim_tx_new: attempted to allocate inappropriate frame type for FLAP connection\n");
  46                         return NULL;
  47                 }
  48         }
  49
  50         if (!(fr = (aim_frame_t *)malloc(sizeof(aim_frame_t))))
  51                 return NULL;
  52         memset(fr, 0, sizeof(aim_frame_t));
  53
  54         fr->conn = conn;
  55
  56         fr->hdrtype = framing;
  57
  58         if (fr->hdrtype == AIM_FRAMETYPE_FLAP) {
  59
  60                 fr->hdr.flap.type = chan;
  61
  62         } else if (fr->hdrtype == AIM_FRAMETYPE_OFT) {
  63
  64                 fr->hdr.oft.type = chan;
  65                 fr->hdr.oft.hdr2len = 0; /* this will get setup by caller */
  66
  67         } else
  68                 faimdprintf(sess, 0, "tx_new: unknown framing\n");
  69
  70         if (datalen > 0) {
  71                 fu8_t *data;
  72
  73                 if (!(data = (unsigned char *)malloc(datalen))) {
  74                         aim_frame_destroy(fr);
  75                         return NULL;
  76                 }
  77
  78                 aim_bstream_init(&fr->data, data, datalen);
  79         }
  80
  81         return fr;
  82 }
  83
  84 /*
  85  * aim_tx_enqeue__queuebased()
  86  *
  87  * The overall purpose here is to enqueue the passed in command struct
  88  * into the outgoing (tx) queue.  Basically...
  89  *   1) Make a scope-irrelevent copy of the struct
  90  *   3) Mark as not-sent-yet
  91  *   4) Enqueue the struct into the list
  92  *   6) Return
  93  *
  94  * Note that this is only used when doing queue-based transmitting;
  95  * that is, when sess->tx_enqueue is set to &aim_tx_enqueue__queuebased.
  96  *
  97  */
  98 static int aim_tx_enqueue__queuebased(aim_session_t *sess, aim_frame_t *fr)
  99 {
 100
 101         if (!fr->conn) {
 102                 faimdprintf(sess, 1, "aim_tx_enqueue: WARNING: enqueueing packet with no connecetion\n");
 103                 fr->conn = aim_getconn_type(sess, AIM_CONN_TYPE_BOS);
 104         }
 105
 106         if (fr->hdrtype == AIM_FRAMETYPE_FLAP) {
 107                 /* assign seqnum -- XXX should really not assign until hardxmit */
 108                 fr->hdr.flap.seqnum = aim_get_next_txseqnum(fr->conn);
 109         }
 110
 111         fr->handled = 0; /* not sent yet */
 112
 113         /* see overhead note in aim_rxqueue counterpart */
 114         if (!sess->queue_outgoing)
 115                 sess->queue_outgoing = fr;
 116         else {
 117                 aim_frame_t *cur;
 118
 119                 for (cur = sess->queue_outgoing; cur->next; cur = cur->next)
 120                         ;
 121                 cur->next = fr;
 122         }
 123
 124         return 0;
 125 }
 126
 127 /*
 128  * aim_tx_enqueue__immediate()
 129  *
 130  * Parallel to aim_tx_enqueue__queuebased, however, this bypasses
 131  * the whole queue mess when you want immediate writes to happen.
 132  *
 133  * Basically the same as its __queuebased couterpart, however
 134  * instead of doing a list append, it just calls aim_tx_sendframe()
 135  * right here.
 136  *
 137  */
 138 static int aim_tx_enqueue__immediate(aim_session_t *sess, aim_frame_t *fr)
 139 {
 140
 141         if (!fr->conn) {
 142                 faimdprintf(sess, 1, "aim_tx_enqueue: ERROR: packet has no connection\n");
 143                 aim_frame_destroy(fr);
 144                 return 0;
 145         }
 146
 147         if (fr->hdrtype == AIM_FRAMETYPE_FLAP)
 148                 fr->hdr.flap.seqnum = aim_get_next_txseqnum(fr->conn);
 149
 150         fr->handled = 0; /* not sent yet */
 151
 152         aim_tx_sendframe(sess, fr);
 153
 154         aim_frame_destroy(fr);
 155
 156         return 0;
 157 }
 158
 159 faim_export int aim_tx_setenqueue(aim_session_t *sess, int what, int (*func)(aim_session_t *, aim_frame_t *))
 160 {
 161
 162         if (what == AIM_TX_QUEUED)
 163                 sess->tx_enqueue = &aim_tx_enqueue__queuebased;
 164         else if (what == AIM_TX_IMMEDIATE)
 165                 sess->tx_enqueue = &aim_tx_enqueue__immediate;
 166         else if (what == AIM_TX_USER) {
 167                 if (!func)
 168                         return -EINVAL;
 169                 sess->tx_enqueue = func;
 170         } else
 171                 return -EINVAL; /* unknown action */
 172
 173         return 0;
 174 }
 175
 176 faim_internal int aim_tx_enqueue(aim_session_t *sess, aim_frame_t *fr)
 177 {
 178
 179         /*
 180          * If we want to send a connection thats inprogress, we have to force
 181          * them to use the queue based version. Otherwise, use whatever they
 182          * want.
 183          */
 184         if (fr && fr->conn &&
 185                         (fr->conn->status & AIM_CONN_STATUS_INPROGRESS)) {
 186                 return aim_tx_enqueue__queuebased(sess, fr);
 187         }
 188
 189         return (*sess->tx_enqueue)(sess, fr);
 190 }
 191
 192 /*
 193  *  aim_get_next_txseqnum()
 194  *
 195  *   This increments the tx command count, and returns the seqnum
 196  *   that should be stamped on the next FLAP packet sent.  This is
 197  *   normally called during the final step of packet preparation
 198  *   before enqueuement (in aim_tx_enqueue()).
 199  *
 200  */
 201 faim_internal flap_seqnum_t aim_get_next_txseqnum(aim_conn_t *conn)
 202 {
 203         flap_seqnum_t ret;
 204
 205         ret = ++conn->seqnum;
 206
 207         return ret;
 208 }
 209
 210 static int aim_send(int fd, const void *buf, size_t count)
 211 {
 212         int left, cur;
 213
 214         for (cur = 0, left = count; left; ) {
 215                 int ret;
 216
 217                 ret = send(fd, ((unsigned char *)buf)+cur, left, 0);
 218                 if (ret == -1)
 219                         return -1;
 220                 else if (ret == 0)
 221                         return cur;
 222
 223                 cur += ret;
 224                 left -= ret;
 225         }
 226
 227         return cur;
 228 }
 229
 230 static int aim_bstream_send(aim_bstream_t *bs, aim_conn_t *conn, size_t count)
 231 {
 232         int wrote = 0;
 233
 234         if (!bs || !conn || (count < 0))
 235                 return -EINVAL;
 236
 237         if (count > aim_bstream_empty(bs))
 238                 count = aim_bstream_empty(bs); /* truncate to remaining space */
 239
 240         if (count)
 241                 wrote = aim_send(conn->fd, bs->data + bs->offset, count);
 242
 243         if (((aim_session_t *)conn->sessv)->debug >= 2) {
 244                 int i;
 245                 aim_session_t *sess = (aim_session_t *)conn->sessv;
 246
 247                 faimdprintf(sess, 2, "\nOutgoing data: (%d bytes)", wrote);
 248                 for (i = 0; i < wrote; i++) {
 249                         if (!(i % 8))
 250                                 faimdprintf(sess, 2, "\n\t");
 251                         faimdprintf(sess, 2, "0x%02x ", *(bs->data + bs->offset + i));
 252                 }
 253                 faimdprintf(sess, 2, "\n");
 254         }
 255
 256
 257         bs->offset += wrote;
 258
 259         return wrote;
 260 }
 261
 262 static int sendframe_flap(aim_session_t *sess, aim_frame_t *fr)
 263 {
 264         aim_bstream_t obs;
 265         fu8_t *obs_raw;
 266         int payloadlen, err = 0, obslen;
 267
 268         payloadlen = aim_bstream_curpos(&fr->data);
 269
 270         if (!(obs_raw = malloc(6 + payloadlen)))
 271                 return -ENOMEM;
 272
 273         aim_bstream_init(&obs, obs_raw, 6 + payloadlen);
 274
 275         /* FLAP header */
 276         aimbs_put8(&obs, 0x2a);
 277         aimbs_put8(&obs, fr->hdr.flap.type);
 278         aimbs_put16(&obs, fr->hdr.flap.seqnum);
 279         aimbs_put16(&obs, payloadlen);
 280
 281         /* payload */
 282         aim_bstream_rewind(&fr->data);
 283         aimbs_putbs(&obs, &fr->data, payloadlen);
 284
 285         obslen = aim_bstream_curpos(&obs);
 286         aim_bstream_rewind(&obs);
 287
 288         if (aim_bstream_send(&obs, fr->conn, obslen) != obslen)
 289                 err = -errno;
 290
 291         free(obs_raw); /* XXX aim_bstream_free */
 292
 293         fr->handled = 1;
 294         fr->conn->lastactivity = time(NULL);
 295
 296         return err;
 297 }
 298
 299 static int sendframe_oft(aim_session_t *sess, aim_frame_t *fr)
 300 {
 301         aim_bstream_t hbs;
 302         fu8_t *hbs_raw;
 303         int hbslen;
 304         int err = 0;
 305
 306         hbslen = 8 + fr->hdr.oft.hdr2len;
 307
 308         if (!(hbs_raw = malloc(hbslen)))
 309                 return -1;
 310
 311         aim_bstream_init(&hbs, hbs_raw, hbslen);
 312
 313         aimbs_putraw(&hbs, fr->hdr.oft.magic, 4);
 314         aimbs_put16(&hbs, fr->hdr.oft.hdr2len + 8);
 315         aimbs_put16(&hbs, fr->hdr.oft.type);
 316         aimbs_putraw(&hbs, fr->hdr.oft.hdr2, fr->hdr.oft.hdr2len);
 317
 318         aim_bstream_rewind(&hbs);
 319
 320         if (aim_bstream_send(&hbs, fr->conn, hbslen) != hbslen) {
 321
 322                 err = -errno;
 323
 324         } else if (aim_bstream_curpos(&fr->data)) {
 325                 int len;
 326
 327                 len = aim_bstream_curpos(&fr->data);
 328                 aim_bstream_rewind(&fr->data);
 329
 330                 if (aim_bstream_send(&fr->data, fr->conn, len) != len)
 331                         err = -errno;
 332         }
 333
 334         free(hbs_raw); /* XXX aim_bstream_free */
 335
 336         fr->handled = 1;
 337         fr->conn->lastactivity = time(NULL);
 338
 339
 340         return err;
 341 }
 342
 343 faim_internal int aim_tx_sendframe(aim_session_t *sess, aim_frame_t *fr)
 344 {
 345         if (fr->hdrtype == AIM_FRAMETYPE_FLAP)
 346                 return sendframe_flap(sess, fr);
 347         else if (fr->hdrtype == AIM_FRAMETYPE_OFT)
 348                 return sendframe_oft(sess, fr);
 349         return -1;
 350 }
 351
 352 faim_export int aim_tx_flushqueue(aim_session_t *sess)
 353 {
 354         aim_frame_t *cur;
 355
 356         for (cur = sess->queue_outgoing; cur; cur = cur->next) {
 357
 358                 if (cur->handled)
 359                         continue; /* already been sent */
 360
 361                 if (cur->conn && (cur->conn->status & AIM_CONN_STATUS_INPROGRESS))
 362                         continue;
 363
 364                 /*
 365                  * And now for the meager attempt to force transmit
 366                  * latency and avoid missed messages.
 367                  */
 368                 if ((cur->conn->lastactivity + cur->conn->forcedlatency) >= time(NULL)) {
 369                         /*
 370                          * XXX should be a break! we dont want to block the
 371                          * upper layers
 372                          *
 373                          * XXX or better, just do this right.
 374                          *
 375                          */
 376                         sleep((cur->conn->lastactivity + cur->conn->forcedlatency) - time(NULL));
 377                 }
 378
 379                 /* XXX this should call the custom "queuing" function!! */
 380                 aim_tx_sendframe(sess, cur);
 381         }
 382
 383         /* purge sent commands from queue */
 384         aim_tx_purgequeue(sess);
 385
 386         return 0;
 387 }
 388
 389 /*
 390  *  aim_tx_purgequeue()
 391  *
 392  *  This is responsable for removing sent commands from the transmit
 393  *  queue. This is not a required operation, but it of course helps
 394  *  reduce memory footprint at run time!
 395  *
 396  */
 397 faim_export void aim_tx_purgequeue(aim_session_t *sess)
 398 {
 399         aim_frame_t *cur, **prev;
 400
 401         for (prev = &sess->queue_outgoing; (cur = *prev); ) {
 402
 403                 if (cur->handled) {
 404                         *prev = cur->next;
 405
 406                         aim_frame_destroy(cur);
 407
 408                 } else
 409                         prev = &cur->next;
 410         }
 411
 412         return;
 413 }
 414
 415 /**
 416  * aim_tx_cleanqueue - get rid of packets waiting for tx on a dying conn
 417  * @sess: session
 418  * @conn: connection that's dying
 419  *
 420  * for now this simply marks all packets as sent and lets them
 421  * disappear without warning.
 422  *
 423  */
 424 faim_export void aim_tx_cleanqueue(aim_session_t *sess, aim_conn_t *conn)
 425 {
 426         aim_frame_t *cur;
 427
 428         for (cur = sess->queue_outgoing; cur; cur = cur->next) {
 429                 if (cur->conn == conn)
 430                         cur->handled = 1;
 431         }
 432
 433         return;
 434 }
 435
 436