SMTP-Client: alocate the Statusmessage big enough so we don't reallocate them later on.
[citadel.git] / citadel / modules / smtp / serv_smtpqueue.c
index 5ae9796bdd0f703c6cad3eb33c7b7b5783488a0d..2f7c3a5905cbe2a16596f17baeaf161ed26b65e2 100644 (file)
  * The VRFY and EXPN commands have been removed from this implementation
  * because nobody uses these commands anymore, except for spammers.
  *
- * Copyright (c) 1998-2009 by the citadel.org team
+ * Copyright (c) 1998-2012 by the citadel.org team
  *
- *  This program is free software; you can redistribute it and/or modify
- *  it under the terms of the GNU General Public License as published by
- *  the Free Software Foundation; either version 3 of the License, or
- *  (at your option) any later version.
+ *  This program is open source software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License version 3.
+ *  
+ *  
  *
  *  This program is distributed in the hope that it will be useful,
  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  *  GNU General Public License for more details.
  *
- *  You should have received a copy of the GNU General Public License
- *  along with this program; if not, write to the Free Software
- *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+ *  
+ *  
+ *  
  */
 
 #include "sysdep.h"
 #include "ctdl_module.h"
 
 #include "smtpqueue.h"
+#include "smtp_clienthandlers.h"
 #include "event_client.h"
 
+
+struct CitContext smtp_queue_CC;
+pthread_mutex_t ActiveQItemsLock;
+HashList *ActiveQItems  = NULL;
 HashList *QItemHandlers = NULL;
+int max_sessions_for_outbound_smtp = 500; /* how many sessions might be active till we stop adding more smtp jobs */
+int ndelay_count = 50; /* every n queued messages we will sleep... */
+int delay_msec = 5000; /* this many seconds. */
+
+static const long MaxRetry = SMTP_RETRY_INTERVAL * 2 * 2 * 2 * 2 * 2 * 2 * 2 * 2 * 2 * 2 * 2 * 2 * 2 * 2;
+int MsgCount            = 0;
+int run_queue_now       = 0;   /* Set to 1 to ignore SMTP send retry times */
+
+void RegisterQItemHandler(const char *Key, long Len, QItemHandler H)
+{
+       QItemHandlerStruct *HS = (QItemHandlerStruct*)malloc(sizeof(QItemHandlerStruct));
+       HS->H = H;
+       Put(QItemHandlers, Key, Len, HS, NULL);
+}
 
-citthread_mutex_t ActiveQItemsLock;
-HashList *ActiveQItems = NULL;
 
-int MsgCount = 0;
-int run_queue_now = 0; /* Set to 1 to ignore SMTP send retry times */
 
-void smtp_try(OneQueItem *MyQItem, 
-             MailQEntry *MyQEntry, 
-             StrBuf *MsgText, 
-             int KeepMsgText, /* KeepMsgText allows us to use MsgText as ours. */
-             int MsgCount, 
-             ParsedURL *RelayUrls);
+void smtp_try_one_queue_entry(OneQueItem *MyQItem,
+                             MailQEntry *MyQEntry,
+                             StrBuf *MsgText,
+/* KeepMsgText allows us to use MsgText as ours. */
+                             int KeepMsgText,
+                             int MsgCount,
+                             ParsedURL *RelayUrls);
 
 
 void smtp_evq_cleanup(void)
 {
-       citthread_mutex_lock(&ActiveQItemsLock);
+
+       pthread_mutex_lock(&ActiveQItemsLock);
        DeleteHash(&QItemHandlers);
        DeleteHash(&ActiveQItems);
-       citthread_mutex_unlock(&ActiveQItemsLock);
-       citthread_mutex_destroy(&ActiveQItemsLock);
+       pthread_mutex_unlock(&ActiveQItemsLock);
+       pthread_setspecific(MyConKey, (void *)&smtp_queue_CC);
+/*     citthread_mutex_destroy(&ActiveQItemsLock); TODO */
 }
 
 int DecreaseQReference(OneQueItem *MyQItem)
 {
        int IDestructQueItem;
 
-       citthread_mutex_lock(&ActiveQItemsLock);
+       pthread_mutex_lock(&ActiveQItemsLock);
        MyQItem->ActiveDeliveries--;
        IDestructQueItem = MyQItem->ActiveDeliveries == 0;
-       citthread_mutex_unlock(&ActiveQItemsLock);
+       pthread_mutex_unlock(&ActiveQItemsLock);
        return IDestructQueItem;
 }
 
+void DecreaseShutdownDeliveries(OneQueItem *MyQItem)
+{
+       pthread_mutex_lock(&ActiveQItemsLock);
+       MyQItem->NotYetShutdownDeliveries--;
+       pthread_mutex_unlock(&ActiveQItemsLock);
+}
+
+int GetShutdownDeliveries(OneQueItem *MyQItem)
+{
+       int DestructNow;
+
+       pthread_mutex_lock(&ActiveQItemsLock);
+       DestructNow = MyQItem->ActiveDeliveries == 0;
+       pthread_mutex_unlock(&ActiveQItemsLock);
+       return DestructNow;
+}
 void RemoveQItem(OneQueItem *MyQItem)
 {
+       long len;
+       const char* Key;
+       void *VData;
        HashPos  *It;
 
-       It = GetNewHashPos(MyQItem->MailQEntries, 0);
-       citthread_mutex_lock(&ActiveQItemsLock);
-       {
-               GetHashPosFromKey(ActiveQItems, IKEY(MyQItem->MessageID), It);
+       pthread_mutex_lock(&ActiveQItemsLock);
+       It = GetNewHashPos(ActiveQItems, 0);
+       if (GetHashPosFromKey(ActiveQItems, LKEY(MyQItem->MessageID), It))
                DeleteEntryFromHash(ActiveQItems, It);
+       else
+       {
+               SMTPC_syslog(LOG_WARNING,
+                            "unable to find QItem with ID[%ld]",
+                            MyQItem->MessageID);
+               while (GetNextHashPos(ActiveQItems, It, &len, &Key, &VData))
+                       SMTPC_syslog(LOG_WARNING,
+                                    "have_: ID[%ld]",
+                                    ((OneQueItem *)VData)->MessageID);
        }
-       citthread_mutex_unlock(&ActiveQItemsLock);
+       pthread_mutex_unlock(&ActiveQItemsLock);
        DeleteHashPos(&It);
 }
 
-void FreeURL(ParsedURL** Url)
-{
-       if (*Url != NULL) {
-               FreeStrBuf(&(*Url)->URL);
-               if ((*Url)->Next != NULL)
-                       FreeURL(&(*Url)->Next);
-               free(*Url);
-               *Url = NULL;
-       }
-}
 
 void FreeMailQEntry(void *qv)
 {
        MailQEntry *Q = qv;
+/*
+       SMTPC_syslog(LOG_DEBUG, "---------------%s--------------", __FUNCTION__);
+       cit_backtrace();
+*/
        FreeStrBuf(&Q->Recipient);
        FreeStrBuf(&Q->StatusMessage);
+
+       memset(Q, 0, sizeof(MailQEntry));
        free(Q);
 }
 void FreeQueItem(OneQueItem **Item)
 {
+/*
+       SMTPC_syslog(LOG_DEBUG, "---------------%s--------------", __FUNCTION__);
+       cit_backtrace();
+*/
        DeleteHash(&(*Item)->MailQEntries);
        FreeStrBuf(&(*Item)->EnvelopeFrom);
        FreeStrBuf(&(*Item)->BounceTo);
+       FreeStrBuf(&(*Item)->SenderRoom);
        FreeURL(&(*Item)->URL);
+       memset(*Item, 0, sizeof(OneQueItem));
        free(*Item);
        Item = NULL;
 }
@@ -170,35 +217,58 @@ void HFreeQueItem(void *Item)
        FreeQueItem((OneQueItem**)&Item);
 }
 
-/* inspect recipients with a status of: 
- * - 0 (no delivery yet attempted) 
+/* inspect recipients with a status of:
+ * - 0 (no delivery yet attempted)
  * - 3/4 (transient errors
  *        were experienced and it's time to try again)
  */
+int CheckQEntryActive(MailQEntry *ThisItem)
+{
+       if ((ThisItem->Status == 0) ||
+           (ThisItem->Status == 3) ||
+           (ThisItem->Status == 4))
+       {
+               return 1;
+       }
+       else
+               return 0;
+}
+int CheckQEntryIsBounce(MailQEntry *ThisItem)
+{
+       if ((ThisItem->Status == 3) ||
+           (ThisItem->Status == 4) ||
+           (ThisItem->Status == 5))
+       {
+               return 1;
+       }
+       else
+               return 0;
+}      
+
 int CountActiveQueueEntries(OneQueItem *MyQItem)
 {
        HashPos  *It;
        long len;
+       long ActiveDeliveries;
        const char *Key;
        void *vQE;
 
-       MyQItem->ActiveDeliveries = 0;
+       ActiveDeliveries = 0;
        It = GetNewHashPos(MyQItem->MailQEntries, 0);
        while (GetNextHashPos(MyQItem->MailQEntries, It, &len, &Key, &vQE))
        {
                MailQEntry *ThisItem = vQE;
-               if ((ThisItem->Status == 0) || 
-                   (ThisItem->Status == 3) ||
-                   (ThisItem->Status == 4))
+
+               if (CheckQEntryActive(ThisItem))
                {
-                       MyQItem->ActiveDeliveries++;
+                       ActiveDeliveries++;
                        ThisItem->Active = 1;
                }
-               else 
+               else
                        ThisItem->Active = 0;
        }
        DeleteHashPos(&It);
-       return MyQItem->ActiveDeliveries;
+       return ActiveDeliveries;
 }
 
 OneQueItem *DeserializeQueueItem(StrBuf *RawQItem, long QueMsgID)
@@ -211,29 +281,10 @@ OneQueItem *DeserializeQueueItem(StrBuf *RawQItem, long QueMsgID)
 
        Item = (OneQueItem*)malloc(sizeof(OneQueItem));
        memset(Item, 0, sizeof(OneQueItem));
-       Item->LastAttempt.retry = SMTP_RETRY_INTERVAL;
+       Item->Retry = SMTP_RETRY_INTERVAL;
        Item->MessageID = -1;
        Item->QueMsgID = QueMsgID;
 
-       citthread_mutex_lock(&ActiveQItemsLock);
-       if (GetHash(ActiveQItems, 
-                   IKEY(Item->QueMsgID), 
-                   &v))
-       {
-               /* WHOOPS. somebody else is already working on this. */
-               citthread_mutex_unlock(&ActiveQItemsLock);
-               FreeQueItem(&Item);
-               return NULL;
-       }
-       else {
-               /* mark our claim on this. */
-               Put(ActiveQItems, 
-                   IKEY(Item->QueMsgID),
-                   Item,
-                   HFreeQueItem);
-               citthread_mutex_unlock(&ActiveQItemsLock);
-       }
-
        Token = NewStrBuf();
        Line = NewStrBufPlain(NULL, 128);
        while (pLine != StrBufNOTNULL) {
@@ -245,13 +296,36 @@ OneQueItem *DeserializeQueueItem(StrBuf *RawQItem, long QueMsgID)
                StrBufExtract_NextToken(Token, Line, &pItemPart, '|');
                if (GetHash(QItemHandlers, SKEY(Token), &vHandler))
                {
-                       QItemHandler H;
-                       H = (QItemHandler) vHandler;
-                       H(Item, Line, &pItemPart);
+                       QItemHandlerStruct *HS;
+                       HS = (QItemHandlerStruct*) vHandler;
+                       HS->H(Item, Line, &pItemPart);
                }
        }
        FreeStrBuf(&Line);
        FreeStrBuf(&Token);
+
+       if (Item->Retry >= MaxRetry)
+               Item->FailNow = 1;
+
+       pthread_mutex_lock(&ActiveQItemsLock);
+       if (GetHash(ActiveQItems,
+                   LKEY(Item->MessageID),
+                   &v))
+       {
+               /* WHOOPS. somebody else is already working on this. */
+               pthread_mutex_unlock(&ActiveQItemsLock);
+               FreeQueItem(&Item);
+               return NULL;
+       }
+       else {
+               /* mark our claim on this. */
+               Put(ActiveQItems,
+                   LKEY(Item->MessageID),
+                   Item,
+                   HFreeQueItem);
+               pthread_mutex_unlock(&ActiveQItemsLock);
+       }
+
        return Item;
 }
 
@@ -265,8 +339,7 @@ StrBuf *SerializeQueueItem(OneQueItem *MyQItem)
 
        QMessage = NewStrBufPlain(NULL, SIZ);
        StrBufPrintf(QMessage, "Content-type: %s\n", SPOOLMIME);
-
-//                "attempted|%ld\n"  "retry|%ld\n",, (long)time(NULL), (long)retry );
+//     "attempted|%ld\n"  "retry|%ld\n",, (long)time(NULL), (long)retry );
        StrBufAppendBufPlain(QMessage, HKEY("\nmsgid|"), 0);
        StrBufAppendPrintf(QMessage, "%ld", MyQItem->MessageID);
 
@@ -283,25 +356,24 @@ StrBuf *SerializeQueueItem(OneQueItem *MyQItem)
                StrBufAppendBuf(QMessage, MyQItem->EnvelopeFrom, 0);
        }
 
+       if (StrLength(MyQItem->SenderRoom) > 0) {
+               StrBufAppendBufPlain(QMessage, HKEY("\nsource_room|"), 0);
+               StrBufAppendBuf(QMessage, MyQItem->SenderRoom, 0);
+       }
+
+       StrBufAppendBufPlain(QMessage, HKEY("\nretry|"), 0);
+       StrBufAppendPrintf(QMessage, "%ld",
+                          MyQItem->Retry);
+
+       StrBufAppendBufPlain(QMessage, HKEY("\nattempted|"), 0);
+       StrBufAppendPrintf(QMessage, "%ld",
+                          time(NULL) /*ctdl_ev_now()*/ + MyQItem->Retry);
+
        It = GetNewHashPos(MyQItem->MailQEntries, 0);
        while (GetNextHashPos(MyQItem->MailQEntries, It, &len, &Key, &vQE))
        {
                MailQEntry *ThisItem = vQE;
-               int i;
-
-               if (!ThisItem->Active)
-                       continue; /* skip already sent ones from the spoolfile. */
-
-               for (i=0; i < ThisItem->nAttempts; i++) {
-                       /* TODO: most probably there is just one retry/attempted per message! */
-                       StrBufAppendBufPlain(QMessage, HKEY("\nretry|"), 0);
-                       StrBufAppendPrintf(QMessage, "%ld", 
-                                          ThisItem->Attempts[i].retry);
 
-                       StrBufAppendBufPlain(QMessage, HKEY("\nattempted|"), 0);
-                       StrBufAppendPrintf(QMessage, "%ld", 
-                                          ThisItem->Attempts[i].when);
-               }
                StrBufAppendBufPlain(QMessage, HKEY("\nremote|"), 0);
                StrBufAppendBuf(QMessage, ThisItem->Recipient, 0);
                StrBufAppendBufPlain(QMessage, HKEY("|"), 0);
@@ -310,7 +382,7 @@ StrBuf *SerializeQueueItem(OneQueItem *MyQItem)
                StrBufAppendBuf(QMessage, ThisItem->StatusMessage, 0);
        }
        DeleteHashPos(&It);
-       StrBufAppendBufPlain(QMessage, HKEY("\n"), 0);  
+       StrBufAppendBufPlain(QMessage, HKEY("\n"), 0);
        return QMessage;
 }
 
@@ -325,14 +397,18 @@ void NewMailQEntry(OneQueItem *Item)
 
        if (Item->MailQEntries == NULL)
                Item->MailQEntries = NewHash(1, Flathash);
-       Item->Current->StatusMessage = NewStrBuf();
+       /* alocate big buffer so we won't get problems reallocating later. */
+       Item->Current->StatusMessage = NewStrBufPlain(NULL, SIZ);
        Item->Current->n = GetCount(Item->MailQEntries);
-       Put(Item->MailQEntries, IKEY(Item->Current->n), Item->Current, FreeMailQEntry);
+       Put(Item->MailQEntries,
+           IKEY(Item->Current->n),
+           Item->Current,
+           FreeMailQEntry);
 }
 
 void QItem_Handle_MsgID(OneQueItem *Item, StrBuf *Line, const char **Pos)
 {
-       Item->MessageID = StrBufExtractNext_int(Line, Pos, '|');
+       Item->MessageID = StrBufExtractNext_long(Line, Pos, '|');
 }
 
 void QItem_Handle_EnvelopeFrom(OneQueItem *Item, StrBuf *Line, const char **Pos)
@@ -349,12 +425,19 @@ void QItem_Handle_BounceTo(OneQueItem *Item, StrBuf *Line, const char **Pos)
        StrBufExtract_NextToken(Item->BounceTo, Line, Pos, '|');
 }
 
+void QItem_Handle_SenderRoom(OneQueItem *Item, StrBuf *Line, const char **Pos)
+{
+       if (Item->SenderRoom == NULL)
+               Item->SenderRoom = NewStrBufPlain(NULL, StrLength(Line));
+       StrBufExtract_NextToken(Item->SenderRoom, Line, Pos, '|');
+}
+
 void QItem_Handle_Recipient(OneQueItem *Item, StrBuf *Line, const char **Pos)
 {
        if (Item->Current == NULL)
                NewMailQEntry(Item);
        if (Item->Current->Recipient == NULL)
-               Item->Current->Recipient =  NewStrBufPlain(NULL, StrLength(Line));
+               Item->Current->Recipient=NewStrBufPlain(NULL, StrLength(Line));
        StrBufExtract_NextToken(Item->Current->Recipient, Line, Pos, '|');
        Item->Current->Status = StrBufExtractNext_int(Line, Pos, '|');
        StrBufExtract_NextToken(Item->Current->StatusMessage, Line, Pos, '|');
@@ -364,15 +447,12 @@ void QItem_Handle_Recipient(OneQueItem *Item, StrBuf *Line, const char **Pos)
 
 void QItem_Handle_retry(OneQueItem *Item, StrBuf *Line, const char **Pos)
 {
-       if (Item->Current == NULL)
-               NewMailQEntry(Item);
-       if (Item->Current->Attempts[Item->Current->nAttempts].retry != 0)
-               Item->Current->nAttempts++;
-       if (Item->Current->nAttempts > MaxAttempts) {
-               Item->FailNow = 1;
-               return;
-       }
-       Item->Current->Attempts[Item->Current->nAttempts].retry = StrBufExtractNext_int(Line, Pos, '|');
+       Item->Retry =
+               StrBufExtractNext_int(Line, Pos, '|');
+       if (Item->Retry == 0)
+               Item->Retry = SMTP_RETRY_INTERVAL;
+       else
+               Item->Retry *= 2;
 }
 
 
@@ -384,23 +464,7 @@ void QItem_Handle_Submitted(OneQueItem *Item, StrBuf *Line, const char **Pos)
 
 void QItem_Handle_Attempted(OneQueItem *Item, StrBuf *Line, const char **Pos)
 {
-       if (Item->Current == NULL)
-               NewMailQEntry(Item);
-       if (Item->Current->Attempts[Item->Current->nAttempts].when != 0)
-               Item->Current->nAttempts++;
-       if (Item->Current->nAttempts > MaxAttempts) {
-               Item->FailNow = 1;
-               return;
-       }
-               
-       Item->Current->Attempts[Item->Current->nAttempts].when = StrBufExtractNext_int(Line, Pos, '|');
-       if (Item->Current->Attempts[Item->Current->nAttempts].when > Item->LastAttempt.when)
-       {
-               Item->LastAttempt.when = Item->Current->Attempts[Item->Current->nAttempts].when;
-               Item->LastAttempt.retry = Item->Current->Attempts[Item->Current->nAttempts].retry * 2;
-               if (Item->LastAttempt.retry > SMTP_RETRY_MAX)
-                       Item->LastAttempt.retry = SMTP_RETRY_MAX;
-       }
+       Item->ReattemptWhen = StrBufExtractNext_int(Line, Pos, '|');
 }
 
 
@@ -412,18 +476,22 @@ StrBuf *smtp_load_msg(OneQueItem *MyQItem, int n)
 {
        CitContext *CCC=CC;
        StrBuf *SendMsg;
-       
+
        CCC->redirect_buffer = NewStrBufPlain(NULL, SIZ);
-       CtdlOutputMsg(MyQItem->MessageID, MT_RFC822, HEADERS_ALL, 0, 1, NULL, (ESC_DOT|SUPPRESS_ENV_TO) );
+       CtdlOutputMsg(MyQItem->MessageID,
+                     MT_RFC822, HEADERS_ALL,
+                     0, 1, NULL,
+                     (ESC_DOT|SUPPRESS_ENV_TO) );
+
        SendMsg = CCC->redirect_buffer;
        CCC->redirect_buffer = NULL;
-       if ((StrLength(SendMsg) > 0) && 
+       if ((StrLength(SendMsg) > 0) &&
            ChrPtr(SendMsg)[StrLength(SendMsg) - 1] != '\n') {
-               CtdlLogPrintf(CTDL_WARNING, 
-                             "SMTP client[%ld]: Possible problem: message did not "
-                             "correctly terminate. (expecting 0x10, got 0x%02x)\n",
-                             MsgCount, //yes uncool, but best choice here... 
-                             ChrPtr(SendMsg)[StrLength(SendMsg) - 1] );
+               SMTPC_syslog(LOG_WARNING,
+                            "[%d] Possible problem: message did not "
+                            "correctly terminate. (expecting 0x10, got 0x%02x)\n",
+                            MsgCount, //yes uncool, but best choice here...
+                            ChrPtr(SendMsg)[StrLength(SendMsg) - 1] );
                StrBufAppendBufPlain(SendMsg, HKEY("\r\n"), 0);
        }
        return SendMsg;
@@ -436,45 +504,57 @@ StrBuf *smtp_load_msg(OneQueItem *MyQItem, int n)
  * instructions for "5" codes (permanent fatal errors) and produce/deliver
  * a "bounce" message (delivery status notification).
  */
-void smtpq_do_bounce(OneQueItem *MyQItem, StrBuf *OMsgTxt) 
+void smtpq_do_bounce(OneQueItem *MyQItem, StrBuf *OMsgTxt)
 {
        static int seq = 0;
 
        struct CtdlMessage *bmsg = NULL;
        StrBuf *boundary;
-       StrBuf *Msg = NULL; 
+       StrBuf *Msg = NULL;
        StrBuf *BounceMB;
        struct recptypes *valid;
-       
+       time_t now;
+
        HashPos *It;
        void *vQE;
        long len;
        const char *Key;
 
+       int first_attempt = 0;
        int successful_bounce = 0;
        int num_bounces = 0;
        int give_up = 0;
 
-       CtdlLogPrintf(CTDL_DEBUG, "smtp_do_bounce() called\n");
+       SMTPCM_syslog(LOG_DEBUG, "smtp_do_bounce() called\n");
+
+       if (MyQItem->SendBounceMail == 0)
+               return;
+
+       now = time (NULL); //ev_time();
+
+       if ( (now - MyQItem->Submitted) > SMTP_GIVE_UP ) {
+               give_up = 1;
+       }
 
-       if ( (ev_time() - MyQItem->Submitted) > SMTP_GIVE_UP ) {
-               give_up = 1;/// TODO: replace time by libevq timer get
+       if (MyQItem->Retry == SMTP_RETRY_INTERVAL) {
+               first_attempt = 1;
        }
 
        /*
         * Now go through the instructions checking for stuff.
         */
+       Msg = NewStrBufPlain(NULL, 1024);
        It = GetNewHashPos(MyQItem->MailQEntries, 0);
        while (GetNextHashPos(MyQItem->MailQEntries, It, &len, &Key, &vQE))
        {
                MailQEntry *ThisItem = vQE;
-               if ((ThisItem->Status == 5) || /* failed now? */
-                   ((give_up == 1) && (ThisItem->Status != 2))) /* giving up after failed attempts... */
+               if ((ThisItem->Active && (ThisItem->Status == 5)) || /* failed now? */
+                   ((give_up == 1) && (ThisItem->Status != 2)) ||
+                   ((first_attempt == 1) && (ThisItem->Status != 2)))
+                       /* giving up after failed attempts... */
                {
-                       if (num_bounces == 0)
-                               Msg = NewStrBufPlain(NULL, 1024);
                        ++num_bounces;
-       
+
                        StrBufAppendBuf(Msg, ThisItem->Recipient, 0);
                        StrBufAppendBufPlain(Msg, HKEY(": "), 0);
                        StrBufAppendBuf(Msg, ThisItem->StatusMessage, 0);
@@ -484,7 +564,7 @@ void smtpq_do_bounce(OneQueItem *MyQItem, StrBuf *OMsgTxt)
        DeleteHashPos(&It);
 
        /* Deliver the bounce if there's anything worth mentioning */
-       CtdlLogPrintf(CTDL_DEBUG, "num_bounces = %d\n", num_bounces);
+       SMTPC_syslog(LOG_DEBUG, "num_bounces = %d\n", num_bounces);
 
        if (num_bounces == 0) {
                FreeStrBuf(&Msg);
@@ -492,16 +572,21 @@ void smtpq_do_bounce(OneQueItem *MyQItem, StrBuf *OMsgTxt)
        }
 
        boundary = NewStrBufPlain(HKEY("=_Citadel_Multipart_"));
-       StrBufAppendPrintf(boundary, "%s_%04x%04x", config.c_fqdn, getpid(), ++seq);
+       StrBufAppendPrintf(boundary,
+                          "%s_%04x%04x",
+                          config.c_fqdn,
+                          getpid(),
+                          ++seq);
 
        /* Start building our bounce message; go shopping for memory first. */
-       BounceMB = NewStrBufPlain(NULL, 
-                                 1024 + /* mime stuff.... */
-                                 StrLength(Msg) +  /* the bounce information... */
-                                 StrLength(OMsgTxt)); /* the original message */
+       BounceMB = NewStrBufPlain(
+               NULL,
+               1024 + /* mime stuff.... */
+               StrLength(Msg) +  /* the bounce information... */
+               StrLength(OMsgTxt)); /* the original message */
        if (BounceMB == NULL) {
                FreeStrBuf(&boundary);
-               CtdlLogPrintf(CTDL_ERR, "Failed to alloc() bounce message.\n");
+               SMTPCM_syslog(LOG_ERR, "Failed to alloc() bounce message.\n");
 
                return;
        }
@@ -510,7 +595,7 @@ void smtpq_do_bounce(OneQueItem *MyQItem, StrBuf *OMsgTxt)
        if (bmsg == NULL) {
                FreeStrBuf(&boundary);
                FreeStrBuf(&BounceMB);
-               CtdlLogPrintf(CTDL_ERR, "Failed to alloc() bounce message.\n");
+               SMTPCM_syslog(LOG_ERR, "Failed to alloc() bounce message.\n");
 
                return;
        }
@@ -519,66 +604,85 @@ void smtpq_do_bounce(OneQueItem *MyQItem, StrBuf *OMsgTxt)
 
        StrBufAppendBufPlain(BounceMB, HKEY("Content-type: multipart/mixed; boundary=\""), 0);
        StrBufAppendBuf(BounceMB, boundary, 0);
-        StrBufAppendBufPlain(BounceMB, HKEY("\"\r\n"), 0);
+       StrBufAppendBufPlain(BounceMB, HKEY("\"\r\n"), 0);
        StrBufAppendBufPlain(BounceMB, HKEY("MIME-Version: 1.0\r\n"), 0);
        StrBufAppendBufPlain(BounceMB, HKEY("X-Mailer: " CITADEL "\r\n"), 0);
-        StrBufAppendBufPlain(BounceMB, HKEY("\r\nThis is a multipart message in MIME format.\r\n\r\n"), 0);
-        StrBufAppendBufPlain(BounceMB, HKEY("--"), 0);
-        StrBufAppendBuf(BounceMB, boundary, 0);
+       StrBufAppendBufPlain(BounceMB, HKEY("\r\nThis is a multipart message in MIME format.\r\n\r\n"), 0);
+       StrBufAppendBufPlain(BounceMB, HKEY("--"), 0);
+       StrBufAppendBuf(BounceMB, boundary, 0);
        StrBufAppendBufPlain(BounceMB, HKEY("\r\n"), 0);
-        StrBufAppendBufPlain(BounceMB, HKEY("Content-type: text/plain\r\n\r\n"), 0);
+       StrBufAppendBufPlain(BounceMB, HKEY("Content-type: text/plain\r\n\r\n"), 0);
 
-       if (give_up) 
+       if (give_up)
                StrBufAppendBufPlain(
-                       BounceMB, 
+                       BounceMB,
                        HKEY(
-                               "A message you sent could not be delivered to some or all of its recipients\n"
-                               "due to prolonged unavailability of its destination(s).\n"
+                               "A message you sent could not be delivered "
+                               "to some or all of its recipients\n"
+                               "due to prolonged unavailability "
+                               "of its destination(s).\n"
                                "Giving up on the following addresses:\n\n"
                                ), 0);
-       else 
+       else
                StrBufAppendBufPlain(
-                       BounceMB, 
+                       BounceMB,
                        HKEY(
-                               "A message you sent could not be delivered to some or all of its recipients.\n"
-                               "The following addresses were undeliverable:\n\n"
+                               "A message you sent could not be delivered "
+                               "to some or all of its recipients.\n"
+                               "The following addresses "
+                               "were undeliverable:\n\n"
                                ), 0);
 
        StrBufAppendBuf(BounceMB, Msg, 0);
        FreeStrBuf(&Msg);
 
+       if (StrLength(MyQItem->SenderRoom) > 0)
+       {
+               StrBufAppendBufPlain(
+                       BounceMB,
+                       HKEY("The message was originaly posted in: "), 0);
+               StrBufAppendBuf(BounceMB, MyQItem->SenderRoom, 0);
+               StrBufAppendBufPlain(
+                       BounceMB,
+                       HKEY("\n"), 0);
+       }
+
        /* Attach the original message */
-       StrBufAppendBufPlain(BounceMB, HKEY("--"), 0);
+       StrBufAppendBufPlain(BounceMB, HKEY("\r\n--"), 0);
        StrBufAppendBuf(BounceMB, boundary, 0);
        StrBufAppendBufPlain(BounceMB, HKEY("\r\n"), 0);
-       StrBufAppendBufPlain(BounceMB, HKEY("Content-type: message/rfc822\r\n"), 0);
-       StrBufAppendBufPlain(BounceMB, HKEY("Content-Transfer-Encoding: 7bit\r\n"), 0);
-       StrBufAppendBufPlain(BounceMB, HKEY("Content-Disposition: inline\r\n"), 0);
+       StrBufAppendBufPlain(BounceMB,
+                            HKEY("Content-type: message/rfc822\r\n"), 0);
+       StrBufAppendBufPlain(BounceMB,
+                            HKEY("Content-Transfer-Encoding: 7bit\r\n"), 0);
+       StrBufAppendBufPlain(BounceMB,
+                            HKEY("Content-Disposition: inline\r\n"), 0);
        StrBufAppendBufPlain(BounceMB, HKEY("\r\n"), 0);
        StrBufAppendBuf(BounceMB, OMsgTxt, 0);
 
        /* Close the multipart MIME scope */
-        StrBufAppendBufPlain(BounceMB, HKEY("--"), 0);
+       StrBufAppendBufPlain(BounceMB, HKEY("--"), 0);
        StrBufAppendBuf(BounceMB, boundary, 0);
        StrBufAppendBufPlain(BounceMB, HKEY("--\r\n"), 0);
 
+       bmsg->cm_magic = CTDLMESSAGE_MAGIC;
+       bmsg->cm_anon_type = MES_NORMAL;
+       bmsg->cm_format_type = FMT_RFC822;
 
-
-        bmsg->cm_magic = CTDLMESSAGE_MAGIC;
-        bmsg->cm_anon_type = MES_NORMAL;
-        bmsg->cm_format_type = FMT_RFC822;
-
-        bmsg->cm_fields['O'] = strdup(MAILROOM);
-        bmsg->cm_fields['A'] = strdup("Citadel");
-        bmsg->cm_fields['N'] = strdup(config.c_nodename);
-        bmsg->cm_fields['U'] = strdup("Delivery Status Notification (Failure)");
+       bmsg->cm_fields['O'] = strdup(MAILROOM);
+       bmsg->cm_fields['A'] = strdup("Citadel");
+       bmsg->cm_fields['N'] = strdup(config.c_nodename);
+       bmsg->cm_fields['U'] = strdup("Delivery Status Notification (Failure)");
        bmsg->cm_fields['M'] = SmashStrBuf(&BounceMB);
 
        /* First try the user who sent the message */
-       if (StrLength(MyQItem->BounceTo) == 0) 
-               CtdlLogPrintf(CTDL_ERR, "No bounce address specified\n");
-       else
-               CtdlLogPrintf(CTDL_DEBUG, "bounce to user? <%s>\n", ChrPtr(MyQItem->BounceTo));
+       if (StrLength(MyQItem->BounceTo) == 0) {
+               SMTPCM_syslog(LOG_ERR, "No bounce address specified\n");
+       }
+       else {
+               SMTPC_syslog(LOG_DEBUG, "bounce to user? <%s>\n",
+                      ChrPtr(MyQItem->BounceTo));
+       }
 
        /* Can we deliver the bounce to the original sender? */
        valid = validate_recipients(ChrPtr(MyQItem->BounceTo), NULL, 0);
@@ -596,98 +700,19 @@ void smtpq_do_bounce(OneQueItem *MyQItem, StrBuf *OMsgTxt)
        free_recipients(valid);
        FreeStrBuf(&boundary);
        CtdlFreeMessage(bmsg);
-       CtdlLogPrintf(CTDL_DEBUG, "Done processing bounces\n");
-}
-
-
-
-int ParseURL(ParsedURL **Url, StrBuf *UrlStr, short DefaultPort)
-{
-       const char *pch, *pEndHost, *pPort, *pCredEnd, *pUserEnd;
-       ParsedURL *url = (ParsedURL *)malloc(sizeof(ParsedURL));
-       memset(url, 0, sizeof(ParsedURL));
-
-       url->af = AF_INET;
-       url->Port =  DefaultPort;
-       /*
-        * http://username:passvoid@[ipv6]:port/url 
-        */
-       url->URL = NewStrBufDup(UrlStr);
-       url->Host = pch = ChrPtr(url->URL);
-       url->LocalPart = strchr(pch, '/');
-       if (url->LocalPart != NULL) {
-               if ((*(url->LocalPart + 1) == '/') && 
-                   (*(url->LocalPart + 2) == ':')) { /* TODO: find default port for this protocol... */
-                       url->Host = url->LocalPart + 3;
-                       url->LocalPart = strchr(url->Host, '/');
-               }
-       }
-       if (url->LocalPart == NULL) {
-               url->LocalPart = pch + StrLength(url->URL);
-       }
-
-       pCredEnd = strchr(pch, '@');
-       if (pCredEnd >= url->LocalPart)
-               pCredEnd = NULL;
-       if (pCredEnd != NULL)
-       {
-               url->User = url->Host;
-               url->Host = pCredEnd + 1;
-               pUserEnd = strchr(url->User, ':');
-               
-               if (pUserEnd > pCredEnd)
-                       pUserEnd = pCredEnd;
-               else {
-                       url->Pass = pUserEnd + 1;
-               }
-               StrBufPeek(url->URL, pUserEnd, 0, '\0');
-               StrBufPeek(url->URL, pCredEnd, 0, '\0');                
-       }
-       
-       pPort = NULL;
-       if (*url->Host == '[') {
-               url->Host ++;
-               pEndHost = strchr(url->Host, ']');
-               if (pEndHost == NULL) {
-                       FreeStrBuf(&url->URL);
-                       free(url);
-                       return 0; /* invalid syntax, no ipv6 */
-               }
-               StrBufPeek(url->URL, pEndHost, 0, '\0');
-               if (*(pEndHost + 1) == ':'){
-                       StrBufPeek(url->URL, pEndHost + 1, 0, '\0');
-                       pPort = pEndHost + 2;
-               }
-               url->af = AF_INET6;
-       }
-       else {
-               pPort = strchr(url->Host, ':');
-               if (pPort != NULL) {
-                       StrBufPeek(url->URL, pPort, 0, '\0');
-                       pPort ++;
-               }
-       }
-       if (pPort != NULL)
-               url->Port = atol(pPort);
-       url->IsIP = inet_pton(url->af, url->Host, &url->Addr);
-       *Url = url;
-       return 1;
+       SMTPCM_syslog(LOG_DEBUG, "Done processing bounces\n");
 }
-/*
-{
 
-       if (threadding)
-               n_smarthosts =  get_hosts(char *mxbuf, char *rectype);
-}
-*/
 /*
  * smtp_do_procmsg()
  *
  * Called by smtp_do_queue() to handle an individual message.
  */
 void smtp_do_procmsg(long msgnum, void *userdata) {
+       time_t now;
+       int mynumsessions = num_sessions;
        struct CtdlMessage *msg = NULL;
-       char *instr = NULL;     
+       char *instr = NULL;
        StrBuf *PlainQItem;
        OneQueItem *MyQItem;
        char *pch;
@@ -699,13 +724,21 @@ void smtp_do_procmsg(long msgnum, void *userdata) {
        ParsedURL *RelayUrls = NULL;
        int HaveBuffers = 0;
        StrBuf *Msg =NULL;
-       
-       CtdlLogPrintf(CTDL_DEBUG, "SMTP Queue: smtp_do_procmsg(%ld)\n", msgnum);
+
+       if (mynumsessions > max_sessions_for_outbound_smtp) {
+               SMTPC_syslog(LOG_DEBUG,
+                            "skipping because of num jobs %d > %d max_sessions_for_outbound_smtp",
+                            mynumsessions,
+                            max_sessions_for_outbound_smtp);
+       }
+
+       SMTPC_syslog(LOG_DEBUG, "smtp_do_procmsg(%ld)\n", msgnum);
        ///strcpy(envelope_from, "");
 
        msg = CtdlFetchMessage(msgnum, 1);
        if (msg == NULL) {
-               CtdlLogPrintf(CTDL_ERR, "SMTP Queue: tried %ld but no such message!\n", msgnum);
+               SMTPC_syslog(LOG_ERR, "tried %ld but no such message!\n",
+                      msgnum);
                return;
        }
 
@@ -725,40 +758,56 @@ void smtp_do_procmsg(long msgnum, void *userdata) {
        FreeStrBuf(&PlainQItem);
 
        if (MyQItem == NULL) {
-               CtdlLogPrintf(CTDL_ERR, "SMTP Queue: Msg No %ld: already in progress!\n", msgnum);              
+               SMTPC_syslog(LOG_ERR,
+                            "Msg No %ld: already in progress!\n",
+                            msgnum);
                return; /* s.b. else is already processing... */
        }
 
        /*
         * Postpone delivery if we've already tried recently.
-        * /
-       if (((time(NULL) - MyQItem->LastAttempt.when) < MyQItem->LastAttempt.retry) && (run_queue_now == 0)) {
-               CtdlLogPrintf(CTDL_DEBUG, "SMTP client: Retry time not yet reached.\n");
+        */
+       now = time(NULL);
+       if ((MyQItem->ReattemptWhen != 0) && 
+           (now < MyQItem->ReattemptWhen) &&
+           (run_queue_now == 0))
+       {
+               SMTPC_syslog(LOG_DEBUG, 
+                            "Retry time not yet reached. %ld seconds left.",
+                            MyQItem->ReattemptWhen - now);
 
                It = GetNewHashPos(MyQItem->MailQEntries, 0);
-               citthread_mutex_lock(&ActiveQItemsLock);
+               pthread_mutex_lock(&ActiveQItemsLock);
                {
-                       GetHashPosFromKey(ActiveQItems, IKEY(MyQItem->MessageID), It);
-                       DeleteEntryFromHash(ActiveQItems, It);
+                       if (GetHashPosFromKey(ActiveQItems,
+                                             LKEY(MyQItem->MessageID),
+                                             It))
+                       {
+                               DeleteEntryFromHash(ActiveQItems, It);
+                       }
                }
-               citthread_mutex_unlock(&ActiveQItemsLock);
+               pthread_mutex_unlock(&ActiveQItemsLock);
                ////FreeQueItem(&MyQItem); TODO: DeleteEntryFromHash frees this?
                DeleteHashPos(&It);
                return;
-       }// TODO: reenable me.*/
+       }
 
        /*
         * Bail out if there's no actual message associated with this
         */
        if (MyQItem->MessageID < 0L) {
-               CtdlLogPrintf(CTDL_ERR, "SMTP Queue: no 'msgid' directive found!\n");
+               SMTPCM_syslog(LOG_ERR, "no 'msgid' directive found!\n");
                It = GetNewHashPos(MyQItem->MailQEntries, 0);
-               citthread_mutex_lock(&ActiveQItemsLock);
+               pthread_mutex_lock(&ActiveQItemsLock);
                {
-                       GetHashPosFromKey(ActiveQItems, IKEY(MyQItem->MessageID), It);
-                       DeleteEntryFromHash(ActiveQItems, It);
+                       if (GetHashPosFromKey(ActiveQItems,
+                                             LKEY(MyQItem->MessageID),
+                                             It))
+                       {
+                               DeleteEntryFromHash(ActiveQItems, It);
+                       }
                }
-               citthread_mutex_unlock(&ActiveQItemsLock);
+               pthread_mutex_unlock(&ActiveQItemsLock);
                DeleteHashPos(&It);
                ////FreeQueItem(&MyQItem); TODO: DeleteEntryFromHash frees this?
                return;
@@ -774,12 +823,46 @@ void smtp_do_procmsg(long msgnum, void *userdata) {
                        const char *Pos = NULL;
                        All = NewStrBufPlain(mxbuf, -1);
                        One = NewStrBufPlain(NULL, StrLength(All) + 1);
-                       
-                       while ((Pos != StrBufNOTNULL) && ((Pos == NULL) || !IsEmptyStr(Pos))) {
+
+                       while ((Pos != StrBufNOTNULL) &&
+                              ((Pos == NULL) ||
+                               !IsEmptyStr(Pos)))
+                       {
+                               StrBufExtract_NextToken(One, All, &Pos, '|');
+                               if (!ParseURL(Url, One, 25)) {
+                                       SMTPC_syslog(LOG_DEBUG,
+                                                    "Failed to parse: %s\n",
+                                                    ChrPtr(One));
+                               }
+                               else {
+                                       ///if (!Url->IsIP)) // todo dupe me fork ipv6
+                                       Url = &(*Url)->Next;
+                               }
+                       }
+                       FreeStrBuf(&All);
+                       FreeStrBuf(&One);
+               }
+
+               Url = &MyQItem->FallBackHost;
+               nRelays = get_hosts(mxbuf, "fallbackhost");
+               if (nRelays > 0) {
+                       StrBuf *All;
+                       StrBuf *One;
+                       const char *Pos = NULL;
+                       All = NewStrBufPlain(mxbuf, -1);
+                       One = NewStrBufPlain(NULL, StrLength(All) + 1);
+
+                       while ((Pos != StrBufNOTNULL) &&
+                              ((Pos == NULL) ||
+                               !IsEmptyStr(Pos)))
+                       {
                                StrBufExtract_NextToken(One, All, &Pos, '|');
-                               if (!ParseURL(Url, One, 25))
-                                       CtdlLogPrintf(CTDL_DEBUG, "Failed to parse: %s\n", ChrPtr(One));
-                               else 
+                               if (!ParseURL(Url, One, 25)) {
+                                       SMTPC_syslog(LOG_DEBUG,
+                                                    "Failed to parse: %s\n",
+                                                    ChrPtr(One));
+                               }
+                               else
                                        Url = &(*Url)->Next;
                        }
                        FreeStrBuf(&All);
@@ -791,41 +874,115 @@ void smtp_do_procmsg(long msgnum, void *userdata) {
        while (GetNextHashPos(MyQItem->MailQEntries, It, &len, &Key, &vQE))
        {
                MailQEntry *ThisItem = vQE;
-               CtdlLogPrintf(CTDL_DEBUG, "SMTP Queue: Task: <%s> %d\n", ChrPtr(ThisItem->Recipient), ThisItem->Active);
+               SMTPC_syslog(LOG_DEBUG, "SMTP Queue: Task: <%s> %d\n",
+                            ChrPtr(ThisItem->Recipient),
+                            ThisItem->Active);
        }
        DeleteHashPos(&It);
 
-       CountActiveQueueEntries(MyQItem);
+       MyQItem->NotYetShutdownDeliveries = 
+       MyQItem->ActiveDeliveries = CountActiveQueueEntries(MyQItem);
+
+       /* failsafe against overload: 
+        * will we exceed the limit set? 
+        */
+       if ((MyQItem->ActiveDeliveries + mynumsessions > max_sessions_for_outbound_smtp) && 
+           /* if yes, did we reach more than half of the quota? */
+           ((mynumsessions * 2) > max_sessions_for_outbound_smtp) && 
+           /* if... would we ever fit into half of the quota?? */
+           (((MyQItem->ActiveDeliveries * 2)  < max_sessions_for_outbound_smtp)))
+       {
+               /* abort delivery for another time. */
+               SMTPC_syslog(LOG_DEBUG,
+                            "SMTP Queue: skipping because of num jobs %d + %ld > %d max_sessions_for_outbound_smtp",
+                            mynumsessions,
+                            MyQItem->ActiveDeliveries,
+                            max_sessions_for_outbound_smtp);
+
+               FreeQueItem(&MyQItem);
+
+               return;
+       }
+
+
        if (MyQItem->ActiveDeliveries > 0)
        {
+               int nActivated = 0;
                int n = MsgCount++;
+               int m = MyQItem->ActiveDeliveries;
                int i = 1;
                Msg = smtp_load_msg(MyQItem, n);
                It = GetNewHashPos(MyQItem->MailQEntries, 0);
-               while ((i <= MyQItem->ActiveDeliveries) && 
-                      (GetNextHashPos(MyQItem->MailQEntries, It, &len, &Key, &vQE)))
+               while ((i <= m) &&
+                      (GetNextHashPos(MyQItem->MailQEntries,
+                                      It, &len, &Key, &vQE)))
                {
                        MailQEntry *ThisItem = vQE;
-                       if (ThisItem->Active == 1) {
-                               int KeepBuffers = (i == MyQItem->ActiveDeliveries);
+
+                       if (ThisItem->Active == 1)
+                       {
+                               int KeepBuffers = (i == m);
+
+                               nActivated++;
+                               if (nActivated % ndelay_count == 0)
+                                       usleep(delay_msec);
+
                                if (i > 1) n = MsgCount++;
-                               CtdlLogPrintf(CTDL_DEBUG, "SMTP Queue: Trying <%s>\n", ChrPtr(ThisItem->Recipient));
-                               smtp_try(MyQItem, ThisItem, Msg, KeepBuffers, n, RelayUrls);
+                               syslog(LOG_DEBUG,
+                                      "SMTPC: Trying <%ld> <%s> %d / %d \n",
+                                      MyQItem->MessageID,
+                                      ChrPtr(ThisItem->Recipient),
+                                      i,
+                                      m);
+                               (*((int*) userdata)) ++;
+                               smtp_try_one_queue_entry(MyQItem,
+                                                        ThisItem,
+                                                        Msg,
+                                                        KeepBuffers,
+                                                        n,
+                                                        RelayUrls);
+
                                if (KeepBuffers) HaveBuffers = 1;
+
                                i++;
                        }
                }
                DeleteHashPos(&It);
        }
-       else 
+       else
        {
                It = GetNewHashPos(MyQItem->MailQEntries, 0);
-               citthread_mutex_lock(&ActiveQItemsLock);
+               pthread_mutex_lock(&ActiveQItemsLock);
                {
-                       GetHashPosFromKey(ActiveQItems, IKEY(MyQItem->MessageID), It);
-                       DeleteEntryFromHash(ActiveQItems, It);
+                       if (GetHashPosFromKey(ActiveQItems,
+                                             LKEY(MyQItem->MessageID),
+                                             It))
+                       {
+                               DeleteEntryFromHash(ActiveQItems, It);
+                       }
+                       else
+                       {
+                               long len;
+                               const char* Key;
+                               void *VData;
+
+                               SMTPC_syslog(LOG_WARNING,
+                                            "unable to find QItem with ID[%ld]",
+                                            MyQItem->MessageID);
+                               while (GetNextHashPos(ActiveQItems,
+                                                     It,
+                                                     &len,
+                                                     &Key,
+                                                     &VData))
+                               {
+                                       SMTPC_syslog(LOG_WARNING,
+                                                    "have: ID[%ld]",
+                                                    ((OneQueItem *)VData)->MessageID);
+                               }
+                       }
+
                }
-               citthread_mutex_unlock(&ActiveQItemsLock);
+               pthread_mutex_unlock(&ActiveQItemsLock);
                DeleteHashPos(&It);
                ////FreeQueItem(&MyQItem); TODO: DeleteEntryFromHash frees this?
 
@@ -842,35 +999,39 @@ void smtp_do_procmsg(long msgnum, void *userdata) {
 
 /*
  * smtp_queue_thread()
- * 
+ *
  * Run through the queue sending out messages.
  */
-void *smtp_queue_thread(void *arg) {
+void smtp_do_queue(void) {
+       static int is_running = 0;
        int num_processed = 0;
-       struct CitContext smtp_queue_CC;
-
-       CtdlThreadSleep(10);
+       int num_activated = 0;
 
-       CtdlFillSystemContext(&smtp_queue_CC, "SMTP Send");
-       citthread_setspecific(MyConKey, (void *)&smtp_queue_CC);
-       CtdlLogPrintf(CTDL_DEBUG, "smtp_queue_thread() initializing\n");
+       if (is_running)
+               return; /* Concurrency check - only one can run */
+       is_running = 1;
 
-       while (!CtdlThreadCheckStop()) {
-               
-               CtdlLogPrintf(CTDL_INFO, "SMTP client: processing outbound queue\n");
+       pthread_setspecific(MyConKey, (void *)&smtp_queue_CC);
+       SMTPCM_syslog(LOG_INFO, "processing outbound queue");
 
-               if (CtdlGetRoom(&CC->room, SMTP_SPOOLOUT_ROOM) != 0) {
-                       CtdlLogPrintf(CTDL_ERR, "Cannot find room <%s>\n", SMTP_SPOOLOUT_ROOM);
-               }
-               else {
-                       num_processed = CtdlForEachMessage(MSGS_ALL, 0L, NULL, SPOOLMIME, NULL, smtp_do_procmsg, NULL);
-               }
-               CtdlLogPrintf(CTDL_INFO, "SMTP client: queue run completed; %d messages processed\n", num_processed);
-               CtdlThreadSleep(60);
+       if (CtdlGetRoom(&CC->room, SMTP_SPOOLOUT_ROOM) != 0) {
+               SMTPC_syslog(LOG_ERR, "Cannot find room <%s>", SMTP_SPOOLOUT_ROOM);
+       }
+       else {
+               num_processed = CtdlForEachMessage(MSGS_ALL,
+                                                  0L,
+                                                  NULL,
+                                                  SPOOLMIME,
+                                                  NULL,
+                                                  smtp_do_procmsg,
+                                                  &num_activated);
        }
+       SMTPC_syslog(LOG_INFO,
+                    "queue run completed; %d messages processed %d activated",
+                    num_processed, num_activated);
 
-       CtdlClearSystemContext();
-       return(NULL);
+       run_queue_now = 0;
+       is_running = 0;
 }
 
 
@@ -885,7 +1046,7 @@ void smtp_init_spoolout(void) {
         * Create the room.  This will silently fail if the room already
         * exists, and that's perfectly ok, because we want it to exist.
         */
-       CtdlCreateRoom(SMTP_SPOOLOUT_ROOM, 3, "", 0, 1, 0, VIEW_MAILBOX);
+       CtdlCreateRoom(SMTP_SPOOLOUT_ROOM, 3, "", 0, 1, 0, VIEW_QUEUE);
 
        /*
         * Make sure it's set to be a "system room" so it doesn't show up
@@ -941,35 +1102,50 @@ void cmd_smtp(char *argbuf) {
 }
 
 
-
-
 CTDL_MODULE_INIT(smtp_queu)
 {
-#ifdef EXPERIMENTAL_SMTP_EVENT_CLIENT
+       char *pstr;
+
        if (!threading)
        {
-               ActiveQItems = NewHash(1, Flathash);
-               citthread_mutex_init(&ActiveQItemsLock, NULL);
+               pstr = getenv("CITSERVER_n_session_max");
+               if ((pstr != NULL) && (*pstr != '\0'))
+                       max_sessions_for_outbound_smtp = atol(pstr); /* how many sessions might be active till we stop adding more smtp jobs */
+
+               pstr = getenv("CITSERVER_smtp_n_delay_count");
+               if ((pstr != NULL) && (*pstr != '\0'))
+                       ndelay_count = atol(pstr); /* every n queued messages we will sleep... */
+
+               pstr = getenv("CITSERVER_smtp_delay");
+               if ((pstr != NULL) && (*pstr != '\0'))
+                       delay_msec = atol(pstr) * 1000; /* this many seconds. */
+
+
+
+
+               CtdlFillSystemContext(&smtp_queue_CC, "SMTP_Send");
+               ActiveQItems = NewHash(1, lFlathash);
+               pthread_mutex_init(&ActiveQItemsLock, NULL);
 
                QItemHandlers = NewHash(0, NULL);
 
-               Put(QItemHandlers, HKEY("msgid"), QItem_Handle_MsgID, reference_free_handler);
-               Put(QItemHandlers, HKEY("envelope_from"), QItem_Handle_EnvelopeFrom, reference_free_handler);
-               Put(QItemHandlers, HKEY("retry"), QItem_Handle_retry, reference_free_handler);
-               Put(QItemHandlers, HKEY("attempted"), QItem_Handle_Attempted, reference_free_handler);
-               Put(QItemHandlers, HKEY("remote"), QItem_Handle_Recipient, reference_free_handler);
-               Put(QItemHandlers, HKEY("bounceto"), QItem_Handle_BounceTo, reference_free_handler);
-               Put(QItemHandlers, HKEY("submitted"), QItem_Handle_Submitted, reference_free_handler);
-////TODO: flush qitemhandlers on exit
+               RegisterQItemHandler(HKEY("msgid"),             QItem_Handle_MsgID);
+               RegisterQItemHandler(HKEY("envelope_from"),     QItem_Handle_EnvelopeFrom);
+               RegisterQItemHandler(HKEY("retry"),             QItem_Handle_retry);
+               RegisterQItemHandler(HKEY("attempted"),         QItem_Handle_Attempted);
+               RegisterQItemHandler(HKEY("remote"),            QItem_Handle_Recipient);
+               RegisterQItemHandler(HKEY("bounceto"),          QItem_Handle_BounceTo);
+               RegisterQItemHandler(HKEY("source_room"),       QItem_Handle_SenderRoom);
+               RegisterQItemHandler(HKEY("submitted"),         QItem_Handle_Submitted);
+
                smtp_init_spoolout();
 
-               CtdlRegisterCleanupHook(smtp_evq_cleanup);
-               CtdlThreadCreate("SMTPEvent Send", CTDLTHREAD_BIGSTACK, smtp_queue_thread, NULL);
+               CtdlRegisterEVCleanupHook(smtp_evq_cleanup);
 
                CtdlRegisterProtoHook(cmd_smtp, "SMTP", "SMTP utility commands");
+               CtdlRegisterSessionHook(smtp_do_queue, EVT_TIMER);
        }
-#endif
-       
+
        /* return our Subversion id for the Log */
        return "smtpeventclient";
 }