citadel/modules/smtp/serv_smtpqueue.c

   1 /*
   2  * This module is an SMTP and ESMTP implementation for the Citadel system.
   3  * It is compliant with all of the following:
   4  *
   5  * RFC  821 - Simple Mail Transfer Protocol
   6  * RFC  876 - Survey of SMTP Implementations
   7  * RFC 1047 - Duplicate messages and SMTP
   8  * RFC 1652 - 8 bit MIME
   9  * RFC 1869 - Extended Simple Mail Transfer Protocol
  10  * RFC 1870 - SMTP Service Extension for Message Size Declaration
  11  * RFC 2033 - Local Mail Transfer Protocol
  12  * RFC 2197 - SMTP Service Extension for Command Pipelining
  13  * RFC 2476 - Message Submission
  14  * RFC 2487 - SMTP Service Extension for Secure SMTP over TLS
  15  * RFC 2554 - SMTP Service Extension for Authentication
  16  * RFC 2821 - Simple Mail Transfer Protocol
  17  * RFC 2822 - Internet Message Format
  18  * RFC 2920 - SMTP Service Extension for Command Pipelining
  19  *
  20  * The VRFY and EXPN commands have been removed from this implementation
  21  * because nobody uses these commands anymore, except for spammers.
  22  *
  23  * Copyright (c) 1998-2009 by the citadel.org team
  24  *
  25  *  This program is free software; you can redistribute it and/or modify
  26  *  it under the terms of the GNU General Public License as published by
  27  *  the Free Software Foundation; either version 3 of the License, or
  28  *  (at your option) any later version.
  29  *
  30  *  This program is distributed in the hope that it will be useful,
  31  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
  32  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  33  *  GNU General Public License for more details.
  34  *
  35  *  You should have received a copy of the GNU General Public License
  36  *  along with this program; if not, write to the Free Software
  37  *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  38  */
  39
  40 #include "sysdep.h"
  41 #include <stdlib.h>
  42 #include <unistd.h>
  43 #include <stdio.h>
  44 #include <termios.h>
  45 #include <fcntl.h>
  46 #include <signal.h>
  47 #include <pwd.h>
  48 #include <errno.h>
  49 #include <sys/types.h>
  50 #include <syslog.h>
  51
  52 #if TIME_WITH_SYS_TIME
  53 # include <sys/time.h>
  54 # include <time.h>
  55 #else
  56 # if HAVE_SYS_TIME_H
  57 #  include <sys/time.h>
  58 # else
  59 #  include <time.h>
  60 # endif
  61 #endif
  62 #include <sys/wait.h>
  63 #include <ctype.h>
  64 #include <string.h>
  65 #include <limits.h>
  66 #include <sys/socket.h>
  67 #include <netinet/in.h>
  68 #include <arpa/inet.h>
  69 #include <libcitadel.h>
  70 #include "citadel.h"
  71 #include "server.h"
  72 #include "citserver.h"
  73 #include "support.h"
  74 #include "config.h"
  75 #include "control.h"
  76 #include "user_ops.h"
  77 #include "database.h"
  78 #include "msgbase.h"
  79 #include "internet_addressing.h"
  80 #include "genstamp.h"
  81 #include "domain.h"
  82 #include "clientsocket.h"
  83 #include "locate_host.h"
  84 #include "citadel_dirs.h"
  85
  86 #include "ctdl_module.h"
  87
  88 #include "smtp_util.h"
  89 #include "smtpqueue.h"
  90 #include "event_client.h"
  91
  92 HashList *QItemHandlers = NULL;
  93
  94 citthread_mutex_t ActiveQItemsLock;
  95 HashList *ActiveQItems = NULL;
  96
  97 int MsgCount = 0;
  98 int run_queue_now = 0;  /* Set to 1 to ignore SMTP send retry times */
  99
 100 void smtp_try(OneQueItem *MyQItem,
 101               MailQEntry *MyQEntry,
 102               StrBuf *MsgText,
 103               int KeepMsgText, /* KeepMsgText allows us to use MsgText as ours. */
 104               int MsgCount);
 105
 106
 107 void smtp_evq_cleanup(void)
 108 {
 109         citthread_mutex_lock(&ActiveQItemsLock);
 110         DeleteHash(&QItemHandlers);
 111         DeleteHash(&ActiveQItems);
 112         citthread_mutex_unlock(&ActiveQItemsLock);
 113         citthread_mutex_destroy(&ActiveQItemsLock);
 114 }
 115
 116 int DecreaseQReference(OneQueItem *MyQItem)
 117 {
 118         int IDestructQueItem;
 119
 120         citthread_mutex_lock(&ActiveQItemsLock);
 121         MyQItem->ActiveDeliveries--;
 122         IDestructQueItem = MyQItem->ActiveDeliveries == 0;
 123         citthread_mutex_unlock(&ActiveQItemsLock);
 124         return IDestructQueItem;
 125 }
 126
 127 void RemoveQItem(OneQueItem *MyQItem)
 128 {
 129         HashPos  *It;
 130
 131         It = GetNewHashPos(MyQItem->MailQEntries, 0);
 132         citthread_mutex_lock(&ActiveQItemsLock);
 133         {
 134                 GetHashPosFromKey(ActiveQItems, IKEY(MyQItem->MessageID), It);
 135                 DeleteEntryFromHash(ActiveQItems, It);
 136         }
 137         citthread_mutex_unlock(&ActiveQItemsLock);
 138         DeleteHashPos(&It);
 139 }
 140
 141
 142
 143 void FreeMailQEntry(void *qv)
 144 {
 145         MailQEntry *Q = qv;
 146         FreeStrBuf(&Q->Recipient);
 147         FreeStrBuf(&Q->StatusMessage);
 148         free(Q);
 149 }
 150 void FreeQueItem(OneQueItem **Item)
 151 {
 152         DeleteHash(&(*Item)->MailQEntries);
 153         FreeStrBuf(&(*Item)->EnvelopeFrom);
 154         FreeStrBuf(&(*Item)->BounceTo);
 155         free(*Item);
 156         Item = NULL;
 157 }
 158 void HFreeQueItem(void *Item)
 159 {
 160         FreeQueItem((OneQueItem**)&Item);
 161 }
 162
 163 /* inspect recipients with a status of:
 164  * - 0 (no delivery yet attempted)
 165  * - 3/4 (transient errors
 166  *        were experienced and it's time to try again)
 167  */
 168 int CountActiveQueueEntries(OneQueItem *MyQItem)
 169 {
 170         HashPos  *It;
 171         long len;
 172         const char *Key;
 173         void *vQE;
 174
 175         MyQItem->ActiveDeliveries = 0;
 176         It = GetNewHashPos(MyQItem->MailQEntries, 0);
 177         while (GetNextHashPos(MyQItem->MailQEntries, It, &len, &Key, &vQE))
 178         {
 179                 MailQEntry *ThisItem = vQE;
 180                 if ((ThisItem->Status == 0) ||
 181                     (ThisItem->Status == 3) ||
 182                     (ThisItem->Status == 4))
 183                 {
 184                         MyQItem->ActiveDeliveries++;
 185                         ThisItem->Active = 1;
 186                 }
 187                 else
 188                         ThisItem->Active = 0;
 189         }
 190         DeleteHashPos(&It);
 191         return MyQItem->ActiveDeliveries;
 192 }
 193
 194 OneQueItem *DeserializeQueueItem(StrBuf *RawQItem, long QueMsgID)
 195 {
 196         OneQueItem *Item;
 197         const char *pLine = NULL;
 198         StrBuf *Line;
 199         StrBuf *Token;
 200         void *v;
 201
 202         Item = (OneQueItem*)malloc(sizeof(OneQueItem));
 203         memset(Item, 0, sizeof(OneQueItem));
 204         Item->LastAttempt.retry = SMTP_RETRY_INTERVAL;
 205         Item->MessageID = -1;
 206         Item->QueMsgID = QueMsgID;
 207
 208         citthread_mutex_lock(&ActiveQItemsLock);
 209         if (GetHash(ActiveQItems,
 210                     IKEY(Item->QueMsgID),
 211                     &v))
 212         {
 213                 /* WHOOPS. somebody else is already working on this. */
 214                 citthread_mutex_unlock(&ActiveQItemsLock);
 215                 FreeQueItem(&Item);
 216                 return NULL;
 217         }
 218         else {
 219                 /* mark our claim on this. */
 220                 Put(ActiveQItems,
 221                     IKEY(Item->QueMsgID),
 222                     Item,
 223                     HFreeQueItem);
 224                 citthread_mutex_unlock(&ActiveQItemsLock);
 225         }
 226
 227         Token = NewStrBuf();
 228         Line = NewStrBufPlain(NULL, 128);
 229         while (pLine != StrBufNOTNULL) {
 230                 const char *pItemPart = NULL;
 231                 void *vHandler;
 232
 233                 StrBufExtract_NextToken(Line, RawQItem, &pLine, '\n');
 234                 if (StrLength(Line) == 0) continue;
 235                 StrBufExtract_NextToken(Token, Line, &pItemPart, '|');
 236                 if (GetHash(QItemHandlers, SKEY(Token), &vHandler))
 237                 {
 238                         QItemHandler H;
 239                         H = (QItemHandler) vHandler;
 240                         H(Item, Line, &pItemPart);
 241                 }
 242         }
 243         FreeStrBuf(&Line);
 244         FreeStrBuf(&Token);
 245         return Item;
 246 }
 247
 248 StrBuf *SerializeQueueItem(OneQueItem *MyQItem)
 249 {
 250         StrBuf *QMessage;
 251         HashPos  *It;
 252         const char *Key;
 253         long len;
 254         void *vQE;
 255
 256         QMessage = NewStrBufPlain(NULL, SIZ);
 257         StrBufPrintf(QMessage, "Content-type: %s\n", SPOOLMIME);
 258
 259 //                 "attempted|%ld\n"  "retry|%ld\n",, (long)time(NULL), (long)retry );
 260         StrBufAppendBufPlain(QMessage, HKEY("\nmsgid|"), 0);
 261         StrBufAppendPrintf(QMessage, "%ld", MyQItem->MessageID);
 262
 263         if (StrLength(MyQItem->BounceTo) > 0) {
 264                 StrBufAppendBufPlain(QMessage, HKEY("\nbounceto|"), 0);
 265                 StrBufAppendBuf(QMessage, MyQItem->BounceTo, 0);
 266         }
 267
 268         if (StrLength(MyQItem->EnvelopeFrom) > 0) {
 269                 StrBufAppendBufPlain(QMessage, HKEY("\nenvelope_from|"), 0);
 270                 StrBufAppendBuf(QMessage, MyQItem->EnvelopeFrom, 0);
 271         }
 272
 273         It = GetNewHashPos(MyQItem->MailQEntries, 0);
 274         while (GetNextHashPos(MyQItem->MailQEntries, It, &len, &Key, &vQE))
 275         {
 276                 MailQEntry *ThisItem = vQE;
 277                 int i;
 278
 279                 if (!ThisItem->Active)
 280                         continue; /* skip already sent ones from the spoolfile. */
 281
 282                 for (i=0; i < ThisItem->nAttempts; i++) {
 283                         StrBufAppendBufPlain(QMessage, HKEY("\nretry|"), 0);
 284                         StrBufAppendPrintf(QMessage, "%ld",
 285                                            ThisItem->Attempts[i].retry);
 286
 287                         StrBufAppendBufPlain(QMessage, HKEY("\nattempted|"), 0);
 288                         StrBufAppendPrintf(QMessage, "%ld",
 289                                            ThisItem->Attempts[i].when);
 290                 }
 291                 StrBufAppendBufPlain(QMessage, HKEY("\nremote|"), 0);
 292                 StrBufAppendBuf(QMessage, ThisItem->Recipient, 0);
 293                 StrBufAppendBufPlain(QMessage, HKEY("|"), 0);
 294                 StrBufAppendPrintf(QMessage, "%d", ThisItem->Status);
 295                 StrBufAppendBufPlain(QMessage, HKEY("|"), 0);
 296                 StrBufAppendBuf(QMessage, ThisItem->StatusMessage, 0);
 297         }
 298         DeleteHashPos(&It);
 299         StrBufAppendBufPlain(QMessage, HKEY("\n"), 0);
 300         return QMessage;
 301 }
 302
 303
 304
 305
 306
 307 void NewMailQEntry(OneQueItem *Item)
 308 {
 309         Item->Current = (MailQEntry*) malloc(sizeof(MailQEntry));
 310         memset(Item->Current, 0, sizeof(MailQEntry));
 311
 312         if (Item->MailQEntries == NULL)
 313                 Item->MailQEntries = NewHash(1, Flathash);
 314         Item->Current->n = GetCount(Item->MailQEntries);
 315         Put(Item->MailQEntries, IKEY(Item->Current->n), Item->Current, FreeMailQEntry);
 316 }
 317
 318 void QItem_Handle_MsgID(OneQueItem *Item, StrBuf *Line, const char **Pos)
 319 {
 320         Item->MessageID = StrBufExtractNext_int(Line, Pos, '|');
 321 }
 322
 323 void QItem_Handle_EnvelopeFrom(OneQueItem *Item, StrBuf *Line, const char **Pos)
 324 {
 325         if (Item->EnvelopeFrom == NULL)
 326                 Item->EnvelopeFrom = NewStrBufPlain(NULL, StrLength(Line));
 327         StrBufExtract_NextToken(Item->EnvelopeFrom, Line, Pos, '|');
 328 }
 329
 330 void QItem_Handle_BounceTo(OneQueItem *Item, StrBuf *Line, const char **Pos)
 331 {
 332         if (Item->BounceTo == NULL)
 333                 Item->BounceTo = NewStrBufPlain(NULL, StrLength(Line));
 334         StrBufExtract_NextToken(Item->BounceTo, Line, Pos, '|');
 335 }
 336
 337 void QItem_Handle_Recipient(OneQueItem *Item, StrBuf *Line, const char **Pos)
 338 {
 339         if (Item->Current == NULL)
 340                 NewMailQEntry(Item);
 341         if (Item->Current->Recipient == NULL)
 342                 Item->Current->Recipient =  NewStrBufPlain(NULL, StrLength(Line));
 343         StrBufExtract_NextToken(Item->Current->Recipient, Line, Pos, '|');
 344         Item->Current->Status = StrBufExtractNext_int(Line, Pos, '|');
 345         StrBufExtract_NextToken(Item->Current->StatusMessage, Line, Pos, '|');
 346         Item->Current = NULL; // TODO: is this always right?
 347 }
 348
 349
 350 void QItem_Handle_retry(OneQueItem *Item, StrBuf *Line, const char **Pos)
 351 {
 352         if (Item->Current == NULL)
 353                 NewMailQEntry(Item);
 354         if (Item->Current->Attempts[Item->Current->nAttempts].retry != 0)
 355                 Item->Current->nAttempts++;
 356         if (Item->Current->nAttempts > MaxAttempts) {
 357                 Item->FailNow = 1;
 358                 return;
 359         }
 360         Item->Current->Attempts[Item->Current->nAttempts].retry = StrBufExtractNext_int(Line, Pos, '|');
 361 }
 362
 363 void QItem_Handle_Attempted(OneQueItem *Item, StrBuf *Line, const char **Pos)
 364 {
 365         if (Item->Current == NULL)
 366                 NewMailQEntry(Item);
 367         if (Item->Current->Attempts[Item->Current->nAttempts].when != 0)
 368                 Item->Current->nAttempts++;
 369         if (Item->Current->nAttempts > MaxAttempts) {
 370                 Item->FailNow = 1;
 371                 return;
 372         }
 373
 374         Item->Current->Attempts[Item->Current->nAttempts].when = StrBufExtractNext_int(Line, Pos, '|');
 375         if (Item->Current->Attempts[Item->Current->nAttempts].when > Item->LastAttempt.when)
 376         {
 377                 Item->LastAttempt.when = Item->Current->Attempts[Item->Current->nAttempts].when;
 378                 Item->LastAttempt.retry = Item->Current->Attempts[Item->Current->nAttempts].retry * 2;
 379                 if (Item->LastAttempt.retry > SMTP_RETRY_MAX)
 380                         Item->LastAttempt.retry = SMTP_RETRY_MAX;
 381         }
 382 }
 383
 384
 385
 386 /**
 387  * this one has to have the context for loading the message via the redirect buffer...
 388  */
 389 StrBuf *smtp_load_msg(OneQueItem *MyQItem, int n)
 390 {
 391         CitContext *CCC=CC;
 392         StrBuf *SendMsg;
 393
 394         CCC->redirect_buffer = NewStrBufPlain(NULL, SIZ);
 395         CtdlOutputMsg(MyQItem->MessageID, MT_RFC822, HEADERS_ALL, 0, 1, NULL, (ESC_DOT|SUPPRESS_ENV_TO) );
 396         SendMsg = CCC->redirect_buffer;
 397         CCC->redirect_buffer = NULL;
 398         if ((StrLength(SendMsg) > 0) &&
 399             ChrPtr(SendMsg)[StrLength(SendMsg) - 1] != '\n') {
 400                 CtdlLogPrintf(CTDL_WARNING,
 401                               "SMTP client[%ld]: Possible problem: message did not "
 402                               "correctly terminate. (expecting 0x10, got 0x%02x)\n",
 403                               MsgCount, //yes uncool, but best choice here...
 404                               ChrPtr(SendMsg)[StrLength(SendMsg) - 1] );
 405                 StrBufAppendBufPlain(SendMsg, HKEY("\r\n"), 0);
 406         }
 407         return SendMsg;
 408 }
 409
 410
 411
 412 /*
 413  * smtp_do_procmsg()
 414  *
 415  * Called by smtp_do_queue() to handle an individual message.
 416  */
 417 void smtp_do_procmsg(long msgnum, void *userdata) {
 418         struct CtdlMessage *msg = NULL;
 419         char *instr = NULL;
 420         StrBuf *PlainQItem;
 421         OneQueItem *MyQItem;
 422         char *pch;
 423         HashPos  *It;
 424         void *vQE;
 425         long len;
 426         const char *Key;
 427
 428         CtdlLogPrintf(CTDL_DEBUG, "SMTP Queue: smtp_do_procmsg(%ld)\n", msgnum);
 429         ///strcpy(envelope_from, "");
 430
 431         msg = CtdlFetchMessage(msgnum, 1);
 432         if (msg == NULL) {
 433                 CtdlLogPrintf(CTDL_ERR, "SMTP Queue: tried %ld but no such message!\n", msgnum);
 434                 return;
 435         }
 436
 437         pch = instr = msg->cm_fields['M'];
 438
 439         /* Strip out the headers (no not amd any other non-instruction) line */
 440         while (pch != NULL) {
 441                 pch = strchr(pch, '\n');
 442                 if ((pch != NULL) && (*(pch + 1) == '\n')) {
 443                         instr = pch + 2;
 444                         pch = NULL;
 445                 }
 446         }
 447         PlainQItem = NewStrBufPlain(instr, -1);
 448         CtdlFreeMessage(msg);
 449         MyQItem = DeserializeQueueItem(PlainQItem, msgnum);
 450         FreeStrBuf(&PlainQItem);
 451
 452         if (MyQItem == NULL) {
 453                 CtdlLogPrintf(CTDL_ERR, "SMTP Queue: Msg No %ld: already in progress!\n", msgnum);
 454                 return; /* s.b. else is already processing... */
 455         }
 456
 457         /*
 458          * Postpone delivery if we've already tried recently.
 459          * /
 460         if (((time(NULL) - MyQItem->LastAttempt.when) < MyQItem->LastAttempt.retry) && (run_queue_now == 0)) {
 461                 CtdlLogPrintf(CTDL_DEBUG, "SMTP client: Retry time not yet reached.\n");
 462
 463                 It = GetNewHashPos(MyQItem->MailQEntries, 0);
 464                 citthread_mutex_lock(&ActiveQItemsLock);
 465                 {
 466                         GetHashPosFromKey(ActiveQItems, IKEY(MyQItem->MessageID), It);
 467                         DeleteEntryFromHash(ActiveQItems, It);
 468                 }
 469                 citthread_mutex_unlock(&ActiveQItemsLock);
 470                 ////FreeQueItem(&MyQItem); TODO: DeleteEntryFromHash frees this?
 471                 DeleteHashPos(&It);
 472                 return;
 473         }// TODO: reenable me.*/
 474
 475         /*
 476          * Bail out if there's no actual message associated with this
 477          */
 478         if (MyQItem->MessageID < 0L) {
 479                 CtdlLogPrintf(CTDL_ERR, "SMTP Queue: no 'msgid' directive found!\n");
 480                 It = GetNewHashPos(MyQItem->MailQEntries, 0);
 481                 citthread_mutex_lock(&ActiveQItemsLock);
 482                 {
 483                         GetHashPosFromKey(ActiveQItems, IKEY(MyQItem->MessageID), It);
 484                         DeleteEntryFromHash(ActiveQItems, It);
 485                 }
 486                 citthread_mutex_unlock(&ActiveQItemsLock);
 487                 DeleteHashPos(&It);
 488                 ////FreeQueItem(&MyQItem); TODO: DeleteEntryFromHash frees this?
 489                 return;
 490         }
 491
 492         It = GetNewHashPos(MyQItem->MailQEntries, 0);
 493         while (GetNextHashPos(MyQItem->MailQEntries, It, &len, &Key, &vQE))
 494         {
 495                 MailQEntry *ThisItem = vQE;
 496                 CtdlLogPrintf(CTDL_DEBUG, "SMTP Queue: Task: <%s> %d\n", ChrPtr(ThisItem->Recipient), ThisItem->Active);
 497         }
 498         DeleteHashPos(&It);
 499
 500         CountActiveQueueEntries(MyQItem);
 501         if (MyQItem->ActiveDeliveries > 0)
 502         {
 503                 int n = MsgCount++;
 504                 int i = 1;
 505                 StrBuf *Msg = smtp_load_msg(MyQItem, n);
 506                 It = GetNewHashPos(MyQItem->MailQEntries, 0);
 507                 while ((i <= MyQItem->ActiveDeliveries) &&
 508                        (GetNextHashPos(MyQItem->MailQEntries, It, &len, &Key, &vQE)))
 509                 {
 510                         MailQEntry *ThisItem = vQE;
 511                         if (ThisItem->Active == 1) {
 512                                 if (i > 1) n = MsgCount++;
 513                                 CtdlLogPrintf(CTDL_DEBUG, "SMTP Queue: Trying <%s>\n", ChrPtr(ThisItem->Recipient));
 514                                 smtp_try(MyQItem, ThisItem, Msg, (i == MyQItem->ActiveDeliveries), n);
 515                                 i++;
 516                         }
 517                 }
 518                 DeleteHashPos(&It);
 519         }
 520         else
 521         {
 522                 It = GetNewHashPos(MyQItem->MailQEntries, 0);
 523                 citthread_mutex_lock(&ActiveQItemsLock);
 524                 {
 525                         GetHashPosFromKey(ActiveQItems, IKEY(MyQItem->MessageID), It);
 526                         DeleteEntryFromHash(ActiveQItems, It);
 527                 }
 528                 citthread_mutex_unlock(&ActiveQItemsLock);
 529                 DeleteHashPos(&It);
 530                 ////FreeQueItem(&MyQItem); TODO: DeleteEntryFromHash frees this?
 531
 532 // TODO: bounce & delete?
 533
 534         }
 535 }
 536
 537
 538
 539 /*
 540  * smtp_queue_thread()
 541  *
 542  * Run through the queue sending out messages.
 543  */
 544 void *smtp_queue_thread(void *arg) {
 545         int num_processed = 0;
 546         struct CitContext smtp_queue_CC;
 547
 548         CtdlThreadSleep(10);
 549
 550         CtdlFillSystemContext(&smtp_queue_CC, "SMTP Send");
 551         citthread_setspecific(MyConKey, (void *)&smtp_queue_CC);
 552         CtdlLogPrintf(CTDL_DEBUG, "smtp_queue_thread() initializing\n");
 553
 554         while (!CtdlThreadCheckStop()) {
 555
 556                 CtdlLogPrintf(CTDL_INFO, "SMTP client: processing outbound queue\n");
 557
 558                 if (CtdlGetRoom(&CC->room, SMTP_SPOOLOUT_ROOM) != 0) {
 559                         CtdlLogPrintf(CTDL_ERR, "Cannot find room <%s>\n", SMTP_SPOOLOUT_ROOM);
 560                 }
 561                 else {
 562                         num_processed = CtdlForEachMessage(MSGS_ALL, 0L, NULL, SPOOLMIME, NULL, smtp_do_procmsg, NULL);
 563                 }
 564                 CtdlLogPrintf(CTDL_INFO, "SMTP client: queue run completed; %d messages processed\n", num_processed);
 565                 CtdlThreadSleep(60);
 566         }
 567
 568         CtdlClearSystemContext();
 569         return(NULL);
 570 }
 571
 572
 573
 574 /*
 575  * Initialize the SMTP outbound queue
 576  */
 577 void smtp_init_spoolout(void) {
 578         struct ctdlroom qrbuf;
 579
 580         /*
 581          * Create the room.  This will silently fail if the room already
 582          * exists, and that's perfectly ok, because we want it to exist.
 583          */
 584         CtdlCreateRoom(SMTP_SPOOLOUT_ROOM, 3, "", 0, 1, 0, VIEW_MAILBOX);
 585
 586         /*
 587          * Make sure it's set to be a "system room" so it doesn't show up
 588          * in the <K>nown rooms list for Aides.
 589          */
 590         if (CtdlGetRoomLock(&qrbuf, SMTP_SPOOLOUT_ROOM) == 0) {
 591                 qrbuf.QRflags2 |= QR2_SYSTEM;
 592                 CtdlPutRoomLock(&qrbuf);
 593         }
 594 }
 595
 596
 597
 598
 599 /*****************************************************************************/
 600 /*                          SMTP UTILITY COMMANDS                            */
 601 /*****************************************************************************/
 602
 603 void cmd_smtp(char *argbuf) {
 604         char cmd[64];
 605         char node[256];
 606         char buf[1024];
 607         int i;
 608         int num_mxhosts;
 609
 610         if (CtdlAccessCheck(ac_aide)) return;
 611
 612         extract_token(cmd, argbuf, 0, '|', sizeof cmd);
 613
 614         if (!strcasecmp(cmd, "mx")) {
 615                 extract_token(node, argbuf, 1, '|', sizeof node);
 616                 num_mxhosts = getmx(buf, node);
 617                 cprintf("%d %d MX hosts listed for %s\n",
 618                         LISTING_FOLLOWS, num_mxhosts, node);
 619                 for (i=0; i<num_mxhosts; ++i) {
 620                         extract_token(node, buf, i, '|', sizeof node);
 621                         cprintf("%s\n", node);
 622                 }
 623                 cprintf("000\n");
 624                 return;
 625         }
 626
 627         else if (!strcasecmp(cmd, "runqueue")) {
 628                 run_queue_now = 1;
 629                 cprintf("%d All outbound SMTP will be retried now.\n", CIT_OK);
 630                 return;
 631         }
 632
 633         else {
 634                 cprintf("%d Invalid command.\n", ERROR + ILLEGAL_VALUE);
 635         }
 636
 637 }
 638
 639
 640
 641
 642 CTDL_MODULE_INIT(smtp_queu)
 643 {
 644 #ifdef EXPERIMENTAL_SMTP_EVENT_CLIENT
 645         if (!threading)
 646         {
 647                 ActiveQItems = NewHash(1, Flathash);
 648                 citthread_mutex_init(&ActiveQItemsLock, NULL);
 649
 650                 QItemHandlers = NewHash(0, NULL);
 651
 652                 Put(QItemHandlers, HKEY("msgid"), QItem_Handle_MsgID, reference_free_handler);
 653                 Put(QItemHandlers, HKEY("envelope_from"), QItem_Handle_EnvelopeFrom, reference_free_handler);
 654                 Put(QItemHandlers, HKEY("retry"), QItem_Handle_retry, reference_free_handler);
 655                 Put(QItemHandlers, HKEY("attempted"), QItem_Handle_Attempted, reference_free_handler);
 656                 Put(QItemHandlers, HKEY("remote"), QItem_Handle_Recipient, reference_free_handler);
 657                 Put(QItemHandlers, HKEY("bounceto"), QItem_Handle_BounceTo, reference_free_handler);
 658 ///submitted /TODO: flush qitemhandlers on exit
 659                 smtp_init_spoolout();
 660
 661                 CtdlRegisterCleanupHook(smtp_evq_cleanup);
 662                 CtdlThreadCreate("SMTPEvent Send", CTDLTHREAD_BIGSTACK, smtp_queue_thread, NULL);
 663
 664                 CtdlRegisterProtoHook(cmd_smtp, "SMTP", "SMTP utility commands");
 665         }
 666 #endif
 667
 668         /* return our Subversion id for the Log */
 669         return "smtpeventclient";
 670 }