]> andersk Git - libfaim.git/blobdiff - aim_txqueue.c
- Sun Feb 11 01:07:36 UTC 2001
[libfaim.git] / aim_txqueue.c
index 50bf425f10e8a9020ce8516be9d99e24e8670966..23ef74f81617b9713eabcff154b3e8ce19745a1d 100644 (file)
@@ -5,10 +5,62 @@
  *
  */
 
+#define FAIM_INTERNAL
 #include <faim/aim.h>
 
+#ifndef _WIN32
+#include <sys/socket.h>
+#endif
+
+/*
+ * Allocate a new tx frame.
+ *
+ * This is more for looks than anything else.
+ *
+ * Right now, that is.  If/when we implement a pool of transmit
+ * frames, this will become the request-an-unused-frame part.
+ *
+ * framing = AIM_FRAMETYPE_OFT/OSCAR
+ * chan = channel for OSCAR, hdrtype for OFT
+ *
+ */
+faim_internal struct command_tx_struct *aim_tx_new(unsigned char framing, int chan, struct aim_conn_t *conn, int datalen)
+{
+  struct command_tx_struct *newtx;
+
+  if (!conn) {
+    printf("aim_tx_new: ERROR: no connection specified\n");
+    return NULL;
+  }
+
+  newtx = (struct command_tx_struct *)malloc(sizeof(struct command_tx_struct));
+  if (!newtx)
+    return NULL;
+  memset(newtx, 0, sizeof(struct command_tx_struct));
+
+  newtx->conn = conn; 
+
+  if(datalen) {
+    newtx->data = (unsigned char *)malloc(datalen);
+    newtx->commandlen = datalen;
+  } else
+    newtx->data = NULL;
+
+  newtx->hdrtype = framing;
+  if (newtx->hdrtype == AIM_FRAMETYPE_OSCAR) {
+    newtx->hdr.oscar.type = chan;
+  } else if (newtx->hdrtype == AIM_FRAMETYPE_OFT) {
+    newtx->hdr.oft.type = chan;
+    newtx->hdr.oft.hdr2len = 0; /* this will get setup by caller */
+  } else { 
+    printf("tx_new: unknown framing\n");
+  }
+
+  return newtx;
+}
+
 /*
- * aim_tx_enqeue()
+ * aim_tx_enqeue__queuebased()
  *
  * The overall purpose here is to enqueue the passed in command struct
  * into the outgoing (tx) queue.  Basically...
  *   5) Unlock the struct once it's linked in
  *   6) Return
  *
+ * Note that this is only used when doing queue-based transmitting;
+ * that is, when sess->tx_enqueue is set to &aim_tx_enqueue__queuebased.
+ *
  */
-
-int aim_tx_enqueue(struct aim_session_t *sess,
-                  struct command_tx_struct *newpacket)
+static int aim_tx_enqueue__queuebased(struct aim_session_t *sess, struct command_tx_struct *newpacket)
 {
-  struct command_tx_struct *workingPtr = NULL;
-  struct command_tx_struct *newpacket_copy = NULL;
+  struct command_tx_struct *cur;
 
-  if (newpacket->conn == NULL)
-    {
-      printf("aim_tx_enqueue: WARNING: enqueueing packet with no connecetion,  defaulting to BOS\n");
+  if (newpacket->conn == NULL) {
+      faimdprintf(1, "aim_tx_enqueue: WARNING: enqueueing packet with no connecetion\n");
       newpacket->conn = aim_getconn_type(sess, AIM_CONN_TYPE_BOS);
-    }
+  }
  
-  newpacket_copy = (struct command_tx_struct *) malloc (sizeof(struct command_tx_struct));
-  memcpy(newpacket_copy, newpacket, sizeof(struct command_tx_struct));
-
-  /* assign seqnum */
-  newpacket_copy->seqnum = aim_get_next_txseqnum(newpacket_copy->conn);
+  if (newpacket->hdrtype == AIM_FRAMETYPE_OSCAR) {
+    /* assign seqnum */
+    newpacket->hdr.oscar.seqnum = aim_get_next_txseqnum(newpacket->conn);
+  }
   /* set some more fields */
-  newpacket_copy->lock = 1; /* lock */
-  newpacket_copy->sent = 0; /* not sent yet */
-  newpacket_copy->next = NULL; /* always last */
+  newpacket->lock = 1; /* lock */
+  newpacket->sent = 0; /* not sent yet */
+  newpacket->next = NULL; /* always last */
 
-  if (sess->queue_outgoing == NULL)
-    {
-      sess->queue_outgoing = newpacket_copy;
-    }
-  else
-    {
-      workingPtr = sess->queue_outgoing;
-      while (workingPtr->next != NULL)
-       workingPtr = workingPtr->next;
-      workingPtr->next = newpacket_copy;
-    }
+  /* see overhead note in aim_rxqueue counterpart */
+  if (sess->queue_outgoing == NULL) {
+    sess->queue_outgoing = newpacket;
+  } else {
+    for (cur = sess->queue_outgoing;
+        cur->next;
+        cur = cur->next)
+      ;
+    cur->next = newpacket;
+  }
 
-  newpacket_copy->lock = 0; /* unlock so it can be sent */
+  newpacket->lock = 0; /* unlock so it can be sent */
 
-#if debug > 2
-  printf("calling aim_tx_printqueue()\n");
-  aim_tx_printqueue();
-  printf("back from aim_tx_printqueue()\n");
+#if debug == 2
+  faimdprintf(2, "calling aim_tx_printqueue()\n");
+  aim_tx_printqueue(sess);
+  faimdprintf(2, "back from aim_tx_printqueue()\n");
 #endif
 
-  /* we'll force a flush for now -- this behavior probably will change */
-#if debug > 1
-  printf("calling aim_tx_flushqueue()\n");
-#endif
-  aim_tx_flushqueue(sess);
-#if debug > 1
-  printf("back from aim_tx_flushqueue()\n");
-#endif
+  return 0;
+}
+
+/*
+ * aim_tx_enqueue__immediate()
+ *
+ * Parallel to aim_tx_enqueue__queuebased, however, this bypasses
+ * the whole queue mess when you want immediate writes to happen.
+ *
+ * Basically the same as its __queuebased couterpart, however
+ * instead of doing a list append, it just calls aim_tx_sendframe()
+ * right here. 
+ * 
+ */
+static int aim_tx_enqueue__immediate(struct aim_session_t *sess, struct command_tx_struct *newpacket)
+{
+  if (newpacket->conn == NULL) {
+    faimdprintf(1, "aim_tx_enqueue: ERROR: packet has no connection\n");
+    if (newpacket->data)
+      free(newpacket->data);
+    free(newpacket);
+    return -1;
+  }
+
+  if (newpacket->hdrtype == AIM_FRAMETYPE_OSCAR)
+    newpacket->hdr.oscar.seqnum = aim_get_next_txseqnum(newpacket->conn);
+
+  newpacket->lock = 1; /* lock */
+  newpacket->sent = 0; /* not sent yet */
+
+  aim_tx_sendframe(sess, newpacket);
+
+  if (newpacket->data)
+    free(newpacket->data);
+  free(newpacket);
 
   return 0;
 }
 
+faim_export int aim_tx_setenqueue(struct aim_session_t *sess, int what,  int (*func)(struct aim_session_t *, struct command_tx_struct *))
+{
+  if (!sess)
+    return -1;
+
+  if (what == AIM_TX_QUEUED)
+    sess->tx_enqueue = &aim_tx_enqueue__queuebased;
+  else if (what == AIM_TX_IMMEDIATE) 
+    sess->tx_enqueue = &aim_tx_enqueue__immediate;
+  else if (what == AIM_TX_USER) {
+    if (!func)
+      return -1;
+    sess->tx_enqueue = func;
+  } else
+    return -1; /* unknown action */
+
+  return 0;
+}
+
+faim_internal int aim_tx_enqueue(struct aim_session_t *sess, struct command_tx_struct *command)
+{
+  /*
+   * If we want to send a connection thats inprogress, we have to force
+   * them to use the queue based version. Otherwise, use whatever they
+   * want.
+   */
+  if (command && command->conn && (command->conn->status & AIM_CONN_STATUS_INPROGRESS)) {
+    return aim_tx_enqueue__queuebased(sess, command);
+  }
+  return (*sess->tx_enqueue)(sess, command);
+}
+
 /* 
  *  aim_get_next_txseqnum()
  *
@@ -84,9 +192,14 @@ int aim_tx_enqueue(struct aim_session_t *sess,
  *   before enqueuement (in aim_tx_enqueue()).
  *
  */
-u_int aim_get_next_txseqnum(struct aim_conn_t *conn)
+faim_internal unsigned int aim_get_next_txseqnum(struct aim_conn_t *conn)
 {
-  return ( ++conn->seqnum );
+  u_int ret;
+  
+  faim_mutex_lock(&conn->seqnum_lock);
+  ret = ++conn->seqnum;
+  faim_mutex_unlock(&conn->seqnum_lock);
+  return ret;
 }
 
 /*
@@ -97,29 +210,28 @@ u_int aim_get_next_txseqnum(struct aim_conn_t *conn)
  *  if the queue isn't working quite right.
  *
  */
-#if debug > 2
-int aim_tx_printqueue(void)
+#if debug == 2
+faim_internal int aim_tx_printqueue(struct aim_session_t *sess)
 {
-  struct command_tx_struct *workingPtr = NULL;
+  struct command_tx_struct *cur;
 
-  workingPtr = sess->queue_outgoing;
-#if debug > 2
-  printf("\ncurrent aim_queue_outgoing...\n");
-  printf("\ttype seqnum  len  lock sent\n");  
-#endif
-  if (workingPtr == NULL)
-    printf("aim_tx_flushqueue(): queue empty");
-  else
-    {
-      while (workingPtr != NULL)
-       {
-         printf("\t  %2x   %4x %4x   %1d    %1d\n", workingPtr->type, workingPtr->seqnum, workingPtr->commandlen, workingPtr->lock, workingPtr->sent);
-         
-         workingPtr = workingPtr->next;
-       }
-    }
+  faimdprintf(2, "\ncurrent aim_queue_outgoing...\n");
+  faimdprintf(2, "\ttype seqnum  len  lock sent\n");  
+
+  if (sess->queue_outgoing == NULL)
+    faimdprintf(2, "aim_tx_flushqueue(): queue empty");
+  else {
+      for (cur = sess->queue_outgoing; cur; cur = cur->next) {
+         faimdprintf(2, "\t  %2x  %2x   %4x %4x   %1d    %1d\n", 
+                     cur->hdrtype,
+                     (cur->hdrtype==AIM_FRAMETYPE_OFT)?cur->hdr.oft.type:cur->hdr.oscar.type, 
+                     (cur->hdrtype==AIM_FRAMETYPE_OSCAR)?cur->hdr.oscar.seqnum:0, 
+                     cur->commandlen, cur->lock, 
+                     cur->sent);
+      }
+  }
 
-  printf("\n(done printing queue)\n");
+  faimdprintf(2, "\n(done printing queue)\n");
   
   return 0;
 }
@@ -149,99 +261,149 @@ int aim_tx_printqueue(void)
  *    9) Step to next struct in list and go back to 1.
  *
  */
-int aim_tx_flushqueue(struct aim_session_t *sess)
+faim_internal int aim_tx_sendframe(struct aim_session_t *sess, struct command_tx_struct *cur)
 {
-  struct command_tx_struct *workingPtr = NULL;
-  u_char *curPacket = NULL;
-#if debug > 1
-  int i = 0;
-#endif
+  int buflen = 0;
+  unsigned char *curPacket;
+
+  if (!cur)
+    return -1; /* fatal */
+
+  cur->lock = 1; /* lock the struct */
+
+  if (cur->hdrtype == AIM_FRAMETYPE_OSCAR)
+    buflen = cur->commandlen + 6;
+  else if (cur->hdrtype == AIM_FRAMETYPE_OFT)
+    buflen = cur->hdr.oft.hdr2len + 8;
+  else {
+    cur->lock = 0;
+    return -1;
+  }
+
+  /* allocate full-packet buffer */
+  if (!(curPacket = (unsigned char *) malloc(buflen))) {
+    cur->lock = 0;
+    return -1;
+  }
+      
+  if (cur->hdrtype == AIM_FRAMETYPE_OSCAR) {
+    /* command byte */
+    curPacket[0] = 0x2a;
+      
+    /* type/family byte */
+    curPacket[1] = cur->hdr.oscar.type;
+      
+    /* bytes 3+4: word: FLAP sequence number */
+    aimutil_put16(curPacket+2, cur->hdr.oscar.seqnum);
+
+    /* bytes 5+6: word: SNAC len */
+    aimutil_put16(curPacket+4, cur->commandlen);
+      
+    /* bytes 7 and on: raw: SNAC data */  /* XXX: ye gods! get rid of this! */
+    memcpy(&(curPacket[6]), cur->data, cur->commandlen);
+
+  } else if (cur->hdrtype == AIM_FRAMETYPE_OFT) {
+    int z = 0;
+
+    z += aimutil_put8(curPacket+z, cur->hdr.oft.magic[0]);
+    z += aimutil_put8(curPacket+z, cur->hdr.oft.magic[1]);
+    z += aimutil_put8(curPacket+z, cur->hdr.oft.magic[2]);
+    z += aimutil_put8(curPacket+z, cur->hdr.oft.magic[3]);
+
+    z += aimutil_put16(curPacket+z, cur->hdr.oft.hdr2len + 8);
+    z += aimutil_put16(curPacket+z, cur->hdr.oft.type);
+
+    memcpy(curPacket+z, cur->hdr.oft.hdr2, cur->hdr.oft.hdr2len);
+  }
+
+  /* 
+   * For OSCAR, a full image of the raw packet data now in curPacket.
+   * For OFT, an image of just the bloated header is in curPacket, 
+   * since OFT allows us to do the data in a different write (yay!).
+   */
+  faim_mutex_lock(&cur->conn->active);
+  if (send(cur->conn->fd, curPacket, buflen, 0) != buflen) {
+    faim_mutex_unlock(&cur->conn->active);
+    cur->sent = 1;
+    aim_conn_close(cur->conn);
+    return 0; /* bail out */
+  }
+
+  if ((cur->hdrtype == AIM_FRAMETYPE_OFT) && cur->commandlen) {
+    int curposi;
+    for(curposi = 0; curposi < cur->commandlen; curposi++)
+      printf("%02x ", cur->data[curposi]);
+
+    if (send(cur->conn->fd, cur->data, cur->commandlen, 0) != (int)cur->commandlen) {
+      /* 
+       * Theres nothing we can do about this since we've already sent the 
+       * header!  The connection is unstable.
+       */
+      faim_mutex_unlock(&cur->conn->active);
+      cur->sent = 1;
+      aim_conn_close(cur->conn);
+      return 0; /* bail out */
+    }
+
+  }
+
+  cur->sent = 1; /* mark the struct as sent */
+  cur->conn->lastactivity = time(NULL);
+
+  faim_mutex_unlock(&cur->conn->active);
 
-  workingPtr = sess->queue_outgoing;
-#if debug > 1
-  printf("beginning txflush...\n");
-#endif
-  while (workingPtr != NULL)
-    {
-      /* only process if its unlocked and unsent */
-      if ( (workingPtr->lock == 0) &&
-          (workingPtr->sent == 0) )
-       {
-
-         /*
-          * And now for the meager attempt to force transmit
-          * latency and avoid missed messages.
-          */
-         if ((workingPtr->conn->lastactivity + workingPtr->conn->forcedlatency) 
-             >= time(NULL))
-           {
-             /* FIXME FIXME -- should be a break! we dont want to block the upper layers */
-             sleep((workingPtr->conn->lastactivity + workingPtr->conn->forcedlatency) - time(NULL));
-           }
-
-         workingPtr->lock = 1; /* lock the struct */
-         
-         /* allocate full-packet buffer */
-         curPacket = (char *) malloc(workingPtr->commandlen + 6);
-         
-         /* command byte */
-         curPacket[0] = 0x2a;
-
-         /* type/family byte */
-         curPacket[1] = workingPtr->type;
-
-         /* bytes 3+4: word: FLAP sequence number */
-         aimutil_put16(curPacket+2, workingPtr->seqnum);
-
-         /* bytes 5+6: word: SNAC len */
-         aimutil_put16(curPacket+4, workingPtr->commandlen);
-
-         /* bytes 7 and on: raw: SNAC data */
-         memcpy(&(curPacket[6]), workingPtr->data, workingPtr->commandlen);
-         
-         /* full image of raw packet data now in curPacket */
-         if ( (u_int)write(workingPtr->conn->fd, curPacket, (workingPtr->commandlen + 6)) != (workingPtr->commandlen + 6))
-           {
-             printf("\nWARNING: Error in sending packet 0x%4x -- will try again next time\n\n", workingPtr->seqnum);
-             workingPtr->sent = 0; /* mark it unsent */
-             return -1;  /* bail out */
-           }
-         else
-           {
-#if debug > 2
-             printf("\nSENT 0x%4x\n\n", workingPtr->seqnum);
-#endif
-             workingPtr->sent = 1; /* mark the struct as sent */
-             workingPtr->conn->lastactivity = time(NULL);
-           }
 #if debug > 2
-         printf("\nPacket:");
-         for (i = 0; i < (workingPtr->commandlen + 6); i++)
-           {
-             if ((i % 8) == 0)
-               printf("\n\t");
-             if (curPacket[i] >= ' ' && curPacket[i]<127)
-                printf("%c=%02x ",curPacket[i], curPacket[i]);
-             else
-                printf("0x%2x ", curPacket[i]);
-           }
-         printf("\n");
+  faimdprintf(2, "\nPacket:");
+  for (i = 0; i < (cur->commandlen + 6); i++) {
+    if ((i % 8) == 0) {
+      faimdprintf(2, "\n\t");
+    }
+    if (curPacket[i] >= ' ' && curPacket[i]<127) {
+      faimdprintf(2, "%c=%02x ", curPacket[i], curPacket[i]);
+    } else {
+      faimdprintf(2, "0x%2x ", curPacket[i]);
+    }
+  }
+  faimdprintf(2, "\n");
 #endif
-         workingPtr->lock = 0; /* unlock the struct */
-         free(curPacket); /* free up full-packet buffer */
-       }
-      workingPtr = workingPtr->next;
+  cur->lock = 0; /* unlock the struct */
+  free(curPacket); /* free up full-packet buffer */
+
+  return 1; /* success */
+}
+
+faim_export int aim_tx_flushqueue(struct aim_session_t *sess)
+{
+  struct command_tx_struct *cur;
+   
+  if (sess->queue_outgoing == NULL)
+    return 0;
+
+  faimdprintf(2, "beginning txflush...\n");
+  for (cur = sess->queue_outgoing; cur; cur = cur->next) {
+    /* only process if its unlocked and unsent */
+    if (!cur->lock && !cur->sent) {
+
+      if (cur->conn && (cur->conn->status & AIM_CONN_STATUS_INPROGRESS))
+       continue;
+
+      /*
+       * And now for the meager attempt to force transmit
+       * latency and avoid missed messages.
+       */
+      if ((cur->conn->lastactivity + cur->conn->forcedlatency) >= time(NULL)) {
+       /* FIXME FIXME -- should be a break! we dont want to block the upper layers */
+       sleep((cur->conn->lastactivity + cur->conn->forcedlatency) - time(NULL));
+      }
+
+      /* XXX XXX XXX this should call the custom "queuing" function!! */
+      if (aim_tx_sendframe(sess, cur) == -1)
+       break;
     }
+  }
 
   /* purge sent commands from queue */
-  /*   this may not always occur explicitly--i may put this on a timer later */
-#if debug > 1
-  printf("calling aim_tx_purgequeue()\n");
-#endif
   aim_tx_purgequeue(sess);
-#if debug > 1
-  printf("back from aim_tx_purgequeu() [you must be a lucky one]\n");
-#endif
 
   return 0;
 }
@@ -254,68 +416,45 @@ int aim_tx_flushqueue(struct aim_session_t *sess)
  *  reduce memory footprint at run time!  
  *
  */
-int aim_tx_purgequeue(struct aim_session_t *sess)
+faim_export void aim_tx_purgequeue(struct aim_session_t *sess)
 {
-  struct command_tx_struct *workingPtr = NULL;
-  struct command_tx_struct *workingPtr2 = NULL;
-#if debug > 1
-  printf("purgequeue(): starting purge\n");
-#endif
-  /* Empty queue: nothing to do */
+  struct command_tx_struct *cur = NULL;
+  struct command_tx_struct *tmp;
+
   if (sess->queue_outgoing == NULL)
-    {
-#if debug > 1
-      printf("purgequeue(): purge done (len=0)\n");
-#endif
-      return 0;
-    }
-  /* One Node queue: free node and return */
-  else if (sess->queue_outgoing->next == NULL)
-    {
-#if debug > 1
-      printf("purgequeue(): entered case len=1\n");
-#endif
-      /* only free if sent AND unlocked -- dont assume sent structs are done */
-      if ( (sess->queue_outgoing->lock == 0) &&
-          (sess->queue_outgoing->sent == 1) )
-       {
-#if debug > 1
-         printf("purgequeue(): purging seqnum 0x%04x\n", aim_queue_outgoing->seqnum);
-#endif
-         workingPtr2 = sess->queue_outgoing;
-         sess->queue_outgoing = NULL;
-         free(workingPtr2->data);
-         free(workingPtr2);
-       }
-#if debug > 1
-      printf("purgequeue(): purge done (len=1)\n");
-#endif
-      return 0;
-    }
-  else
-    {
-#if debug > 1
-      printf("purgequeue(): entering case len>1\n");
-#endif
-      while(workingPtr->next != NULL)
-       {
-         if ( (workingPtr->next->lock == 0) &&
-              (workingPtr->next->sent == 1) )
-           {
-#if debug > 1
-             printf("purgequeue(): purging seqnum 0x%04x\n", workingPtr->next->seqnum);
-#endif
-             workingPtr2 = workingPtr->next;
-             workingPtr->next = workingPtr2->next;
-             free(workingPtr2->data);
-             free(workingPtr2);
-           }
-       }
-#if debug > 1
-      printf("purgequeue(): purge done (len>1)\n");
-#endif
-      return 0;
+    return;
+  
+  if (sess->queue_outgoing->next == NULL) {
+    if (!sess->queue_outgoing->lock && sess->queue_outgoing->sent) {
+      tmp = sess->queue_outgoing;
+      sess->queue_outgoing = NULL;
+      if (tmp->hdrtype == AIM_FRAMETYPE_OFT)
+       free(tmp->hdr.oft.hdr2);
+      free(tmp->data);
+      free(tmp);
     }
+    return;
+  }
+
+  for(cur = sess->queue_outgoing; cur->next != NULL; ) {
+    if (!cur->next->lock && cur->next->sent) {
+      tmp = cur->next;
+      cur->next = tmp->next;
+      if (tmp->hdrtype == AIM_FRAMETYPE_OFT)
+       free(tmp->hdr.oft.hdr2);
+      free(tmp->data);
+      free(tmp);
+    }  
+    cur = cur->next;
 
-  /* no reach */
+    /* 
+     * Be careful here.  Because of the way we just
+     * manipulated the pointer, cur may be NULL and 
+     * the for() will segfault doing the check unless
+     * we find this case first.
+     */
+    if (cur == NULL)   
+      break;
+  }
+  return;
 }
This page took 0.051748 seconds and 4 git commands to generate.