开通博客赚积分
发布资源赚积分
分类

源码开发语言/平台
当前位置：首页 > 源码/资料 > Internet/网络编程 > Email服务器 > 查看源码
MessQueue.cpp
资源名称：xmail-1.22.rar　[点击查看]
上传用户：woshihumen
上传日期：2013-07-18
资源大小：484k
文件大小：40k
源码类别：

Email服务器

开发平台：

Visual C++

MessQueue.cpp：源码内容
  1. /*
  2.  *  XMail by Davide Libenzi ( Intranet and Internet mail server )
  3.  *  Copyright (C) 1999,..,2004  Davide Libenzi
  4.  *
  5.  *  This program is free software; you can redistribute it and/or modify
  6.  *  it under the terms of the GNU General Public License as published by
  7.  *  the Free Software Foundation; either version 2 of the License, or
  8.  *  (at your option) any later version.
  9.  *
  10.  *  This program is distributed in the hope that it will be useful,
  11.  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
  12.  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13.  *  GNU General Public License for more details.
  14.  *
  15.  *  You should have received a copy of the GNU General Public License
  16.  *  along with this program; if not, write to the Free Software
  17.  *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  18.  *
  19.  *  Davide Libenzi <davidel@xmailserver.org>
  20.  *
  21.  */
  23. #include "SysInclude.h"
  24. #include "SysDep.h"
  25. #include "SvrDefines.h"
  26. #include "ShBlocks.h"
  27. #include "ResLocks.h"
  28. #include "StrUtils.h"
  29. #include "SList.h"
  30. #include "MD5.h"
  31. #include "Base64Enc.h"
  32. #include "BuffSock.h"
  33. #include "MailConfig.h"
  34. #include "UsrUtils.h"
  35. #include "SvrUtils.h"
  36. #include "AppDefines.h"
  37. #include "MiscUtils.h"
  38. #include "SMTPUtils.h"
  39. #include "MessQueue.h"
  40. #include "SMAILUtils.h"
  42. #define QUEF_SHUTDOWN               (1 << 0)
  44. #define QUMF_DELETED                (1 << 0)
  45. #define QUMF_FREEZE                 (1 << 1)
  47. #define QUE_MASK_TMPFLAGS(v)        ((v) & ~(QUMF_DELETED | QUMF_FREEZE))
  49. #define QUE_ARENA_SCAN_INTERVAL     15
  50. #define QUE_ARENA_SCAN_WAIT         2
  51. #define QUE_SCAN_THREAD_MAXWAIT     60
  53. struct MessageQueue {
  54. SysListHead ReadyQueue;
  55. SysListHead RsndArenaQueue;
  56. int iReadyCount;
  57. int iRsndArenaCount;
  58. SYS_MUTEX hMutex;
  59. SYS_EVENT hReadyEvent;
  60. char *pszRootPath;
  61. int iMaxRetry;
  62. int iRetryTimeout;
  63. int iRetryIncrRatio;
  64. int iNumDirsLevel;
  65. unsigned long ulFlags;
  66. SYS_THREAD hRsndScanThread;
  67. };
  69. struct QueueMessage {
  70. SysListHead LLink;
  71. int iLevel1;
  72. int iLevel2;
  73. char const *pszQueueDir;
  74. char *pszFileName;
  75. int iNumTries;
  76. time_t tLastTry;
  77. unsigned long ulFlags;
  78. };
  80. static int QueCreateStruct(char const *pszRootPath);
  81. static int QueLoad(MessageQueue * pMQ);
  82. static int QueLoadMessages(MessageQueue * pMQ, int iLevel1, int iLevel2);
  83. static QueueMessage *QueAllocMessage(int iLevel1, int iLevel2, char const *pszQueueDir,
  84.      char const *pszFileName, int iNumTries, time_t tLastTry);
  85. static int QueFreeMessage(QueueMessage * pQM);
  86. static int QueFreeMessList(SysListHead * pHead);
  87. static int QueLoadMessageStat(MessageQueue * pMQ, QueueMessage * pQM);
  88. static int QueStatMessage(MessageQueue * pMQ, QueueMessage * pQM);
  89. static int QueGetFilePath(MessageQueue * pMQ, QueueMessage * pQM, char *pszFilePath,
  90.   char const *pszQueueDir = NULL);
  91. static int QueDoMessageCleanup(QUEUE_HANDLE hQueue, QMSG_HANDLE hMessage);
  92. static int QueAddNew(MessageQueue * pMQ, QueueMessage * pQM);
  93. static bool QueMessageExpired(MessageQueue * pMQ, QueueMessage * pQM);
  94. static time_t QueNextRetryOp(int iNumTries, unsigned int uRetryTimeout,
  95.      unsigned int uRetryIncrRatio);
  96. static bool QueMessageReadyToSend(MessageQueue * pMQ, QueueMessage * pQM);
  97. static int QueAddRsnd(MessageQueue * pMQ, QueueMessage * pQM);
  98. static unsigned int QueRsndThread(void *pThreadData);
  99. static int QueScanRsndArena(MessageQueue * pMQ);
  100. static bool QueMessageDestMatch(MessageQueue * pMQ, QueueMessage * pQM,
  101. char const *pszAddressMatch);
  103. QUEUE_HANDLE QueOpen(char const *pszRootPath, int iMaxRetry, int iRetryTimeout,
  104.      int iRetryIncrRatio, int iNumDirsLevel)
  105. {
  107. MessageQueue *pMQ = (MessageQueue *) SysAlloc(sizeof(MessageQueue));
  109. if (pMQ == NULL)
  110. return (INVALID_QUEUE_HANDLE);
  112. SYS_INIT_LIST_HEAD(&pMQ->ReadyQueue);
  114. SYS_INIT_LIST_HEAD(&pMQ->RsndArenaQueue);
  116. pMQ->iReadyCount = 0;
  118. pMQ->iRsndArenaCount = 0;
  120. pMQ->iMaxRetry = iMaxRetry;
  122. pMQ->iRetryTimeout = iRetryTimeout;
  124. pMQ->iRetryIncrRatio = iRetryIncrRatio;
  126. pMQ->iNumDirsLevel = iNumDirsLevel;
  128. pMQ->ulFlags = 0;
  130. if ((pMQ->hMutex = SysCreateMutex()) == SYS_INVALID_MUTEX) {
  131. SysFree(pMQ);
  132. return (INVALID_QUEUE_HANDLE);
  133. }
  135. if ((pMQ->hReadyEvent = SysCreateEvent(1)) == SYS_INVALID_EVENT) {
  136. SysCloseMutex(pMQ->hMutex);
  137. SysFree(pMQ);
  138. return (INVALID_QUEUE_HANDLE);
  139. }
  140. ///////////////////////////////////////////////////////////////////////////////
  141. //  Set the queue root path
  142. ///////////////////////////////////////////////////////////////////////////////
  143. char szRootPath[SYS_MAX_PATH] = "";
  145. StrSNCpy(szRootPath, pszRootPath);
  146. AppendSlash(szRootPath);
  148. pMQ->pszRootPath = SysStrDup(szRootPath);
  150. ///////////////////////////////////////////////////////////////////////////////
  151. //  Load queue status
  152. ///////////////////////////////////////////////////////////////////////////////
  153. if (QueLoad(pMQ) < 0) {
  154. ErrorPush();
  155. SysFree(pMQ->pszRootPath);
  156. SysCloseEvent(pMQ->hReadyEvent);
  157. SysCloseMutex(pMQ->hMutex);
  158. SysFree(pMQ);
  160. ErrSetErrorCode(ErrorPop());
  161. return (INVALID_QUEUE_HANDLE);
  162. }
  163. ///////////////////////////////////////////////////////////////////////////////
  164. //  Start rsnd arena scan thread
  165. ///////////////////////////////////////////////////////////////////////////////
  166. if ((pMQ->hRsndScanThread = SysCreateThread(QueRsndThread, pMQ)) == SYS_INVALID_THREAD) {
  167. ErrorPush();
  168. QueFreeMessList(&pMQ->ReadyQueue);
  169. QueFreeMessList(&pMQ->RsndArenaQueue);
  170. SysFree(pMQ->pszRootPath);
  171. SysCloseEvent(pMQ->hReadyEvent);
  172. SysCloseMutex(pMQ->hMutex);
  173. SysFree(pMQ);
  175. ErrSetErrorCode(ErrorPop());
  176. return (INVALID_QUEUE_HANDLE);
  177. }
  179. return ((QUEUE_HANDLE) pMQ);
  181. }
  183. int QueClose(QUEUE_HANDLE hQueue)
  184. {
  186. MessageQueue *pMQ = (MessageQueue *) hQueue;
  188. ///////////////////////////////////////////////////////////////////////////////
  189. //  Set the shutdown flag and wait for rsnd scan thread to terminate
  190. ///////////////////////////////////////////////////////////////////////////////
  191. pMQ->ulFlags |= QUEF_SHUTDOWN;
  193. SysWaitThread(pMQ->hRsndScanThread, QUE_SCAN_THREAD_MAXWAIT);
  195. SysCloseThread(pMQ->hRsndScanThread, 1);
  197. ///////////////////////////////////////////////////////////////////////////////
  198. //  Clear queues
  199. ///////////////////////////////////////////////////////////////////////////////
  200. QueFreeMessList(&pMQ->ReadyQueue);
  202. QueFreeMessList(&pMQ->RsndArenaQueue);
  204. SysCloseEvent(pMQ->hReadyEvent);
  206. SysCloseMutex(pMQ->hMutex);
  208. SysFree(pMQ->pszRootPath);
  210. SysFree(pMQ);
  212. return (0);
  214. }
  216. int QueGetDirsLevel(QUEUE_HANDLE hQueue)
  217. {
  219. MessageQueue *pMQ = (MessageQueue *) hQueue;
  221. return (pMQ->iNumDirsLevel);
  223. }
  225. char const *QueGetRootPath(QUEUE_HANDLE hQueue)
  226. {
  228. MessageQueue *pMQ = (MessageQueue *) hQueue;
  230. return (pMQ->pszRootPath);
  232. }
  234. static int QueCreateStruct(char const *pszRootPath)
  235. {
  236. ///////////////////////////////////////////////////////////////////////////////
  237. //  Create message dir ( new messages queue )
  238. ///////////////////////////////////////////////////////////////////////////////
  239. char szDirPath[SYS_MAX_PATH] = "";
  241. StrSNCpy(szDirPath, pszRootPath);
  242. AppendSlash(szDirPath);
  243. StrSNCat(szDirPath, QUEUE_MESS_DIR);
  245. if (!SysExistDir(szDirPath) && (SysMakeDir(szDirPath) < 0))
  246. return (ErrGetErrorCode());
  248. ///////////////////////////////////////////////////////////////////////////////
  249. //  Create message resend dir ( resend messages queue )
  250. ///////////////////////////////////////////////////////////////////////////////
  251. StrSNCpy(szDirPath, pszRootPath);
  252. AppendSlash(szDirPath);
  253. StrSNCat(szDirPath, QUEUE_RSND_DIR);
  255. if (!SysExistDir(szDirPath) && (SysMakeDir(szDirPath) < 0))
  256. return (ErrGetErrorCode());
  258. ///////////////////////////////////////////////////////////////////////////////
  259. //  Create info dir
  260. ///////////////////////////////////////////////////////////////////////////////
  261. StrSNCpy(szDirPath, pszRootPath);
  262. AppendSlash(szDirPath);
  263. StrSNCat(szDirPath, QUEUE_INFO_DIR);
  265. if (!SysExistDir(szDirPath) && (SysMakeDir(szDirPath) < 0))
  266. return (ErrGetErrorCode());
  268. ///////////////////////////////////////////////////////////////////////////////
  269. //  Create temp dir
  270. ///////////////////////////////////////////////////////////////////////////////
  271. StrSNCpy(szDirPath, pszRootPath);
  272. AppendSlash(szDirPath);
  273. StrSNCat(szDirPath, QUEUE_TEMP_DIR);
  275. if (!SysExistDir(szDirPath) && (SysMakeDir(szDirPath) < 0))
  276. return (ErrGetErrorCode());
  278. ///////////////////////////////////////////////////////////////////////////////
  279. //  Create send log dir
  280. ///////////////////////////////////////////////////////////////////////////////
  281. StrSNCpy(szDirPath, pszRootPath);
  282. AppendSlash(szDirPath);
  283. StrSNCat(szDirPath, QUEUE_SLOG_DIR);
  285. if (!SysExistDir(szDirPath) && (SysMakeDir(szDirPath) < 0))
  286. return (ErrGetErrorCode());
  288. ///////////////////////////////////////////////////////////////////////////////
  289. //  Create custom message processing dir
  290. ///////////////////////////////////////////////////////////////////////////////
  291. StrSNCpy(szDirPath, pszRootPath);
  292. AppendSlash(szDirPath);
  293. StrSNCat(szDirPath, QUEUE_CUST_DIR);
  295. if (!SysExistDir(szDirPath) && (SysMakeDir(szDirPath) < 0))
  296. return (ErrGetErrorCode());
  298. ///////////////////////////////////////////////////////////////////////////////
  299. //  Create user custom message processing dir ( mailproc.tab cache )
  300. ///////////////////////////////////////////////////////////////////////////////
  301. StrSNCpy(szDirPath, pszRootPath);
  302. AppendSlash(szDirPath);
  303. StrSNCat(szDirPath, QUEUE_MPRC_DIR);
  305. if (!SysExistDir(szDirPath) && (SysMakeDir(szDirPath) < 0))
  306. return (ErrGetErrorCode());
  308. ///////////////////////////////////////////////////////////////////////////////
  309. //  Create frozen dir
  310. ///////////////////////////////////////////////////////////////////////////////
  311. StrSNCpy(szDirPath, pszRootPath);
  312. AppendSlash(szDirPath);
  313. StrSNCat(szDirPath, QUEUE_FROZ_DIR);
  315. if (!SysExistDir(szDirPath) && (SysMakeDir(szDirPath) < 0))
  316. return (ErrGetErrorCode());
  318. return (0);
  320. }
  322. static int QueLoad(MessageQueue * pMQ)
  323. {
  325. char szCurrPath[SYS_MAX_PATH] = "";
  327. for (int ii = 0; ii < pMQ->iNumDirsLevel; ii++) {
  328. SysSNPrintf(szCurrPath, sizeof(szCurrPath) - 1, "%s%d", pMQ->pszRootPath, ii);
  330. if (!SysExistDir(szCurrPath) && (SysMakeDir(szCurrPath) < 0))
  331. return (ErrGetErrorCode());
  333. for (int jj = 0; jj < pMQ->iNumDirsLevel; jj++) {
  334. SysSNPrintf(szCurrPath, sizeof(szCurrPath) - 1, "%s%d%s%d",
  335.     pMQ->pszRootPath, ii, SYS_SLASH_STR, jj);
  337. if (!SysExistDir(szCurrPath) && (SysMakeDir(szCurrPath) < 0))
  338. return (ErrGetErrorCode());
  340. if (QueCreateStruct(szCurrPath) < 0)
  341. return (ErrGetErrorCode());
  343. ///////////////////////////////////////////////////////////////////////////////
  344. //  Load queue directory
  345. ///////////////////////////////////////////////////////////////////////////////
  346. if (QueLoadMessages(pMQ, ii, jj) < 0)
  347. return (ErrGetErrorCode());
  349. }
  350. }
  352. return (0);
  354. }
  356. static int QueLoadMessages(MessageQueue * pMQ, int iLevel1, int iLevel2)
  357. {
  358. ///////////////////////////////////////////////////////////////////////////////
  359. //  File scan the new messages dir
  360. ///////////////////////////////////////////////////////////////////////////////
  361. char szDirPath[SYS_MAX_PATH] = "";
  363. SysSNPrintf(szDirPath, sizeof(szDirPath) - 1, "%s%d%s%d%s%s",
  364.     pMQ->pszRootPath, iLevel1, SYS_SLASH_STR, iLevel2,
  365.     SYS_SLASH_STR, QUEUE_MESS_DIR);
  367. char szMsgFileName[SYS_MAX_PATH] = "";
  368. FSCAN_HANDLE hFileScan = MscFirstFile(szDirPath, 0, szMsgFileName);
  370. if (hFileScan != INVALID_FSCAN_HANDLE) {
  371. do {
  373. if (!IsDotFilename(szMsgFileName)) {
  374. QueueMessage *pQM =
  375.     QueAllocMessage(iLevel1, iLevel2, QUEUE_MESS_DIR,
  376.     szMsgFileName, 0, 0);
  378. if (pQM != NULL) {
  379. ///////////////////////////////////////////////////////////////////////////////
  380. //  Add the file to the message queue
  381. ///////////////////////////////////////////////////////////////////////////////
  382. SYS_LIST_ADDT(&pQM->LLink, &pMQ->ReadyQueue);
  384. ++pMQ->iReadyCount;
  385. }
  386. }
  388. } while (MscNextFile(hFileScan, szMsgFileName));
  390. MscCloseFindFile(hFileScan);
  392. ///////////////////////////////////////////////////////////////////////////////
  393. //  Set the mess event if the queue is not empty
  394. ///////////////////////////////////////////////////////////////////////////////
  395. if (pMQ->iReadyCount > 0)
  396. SysSetEvent(pMQ->hReadyEvent);
  398. }
  399. ///////////////////////////////////////////////////////////////////////////////
  400. //  File scan the resend messages dir
  401. ///////////////////////////////////////////////////////////////////////////////
  402. SysSNPrintf(szDirPath, sizeof(szDirPath) - 1, "%s%d%s%d%s%s",
  403.     pMQ->pszRootPath, iLevel1, SYS_SLASH_STR, iLevel2,
  404.     SYS_SLASH_STR, QUEUE_RSND_DIR);
  406. if ((hFileScan = MscFirstFile(szDirPath, 0, szMsgFileName)) != INVALID_FSCAN_HANDLE) {
  407. do {
  409. if (!IsDotFilename(szMsgFileName)) {
  410. QueueMessage *pQM =
  411.     QueAllocMessage(iLevel1, iLevel2, QUEUE_RSND_DIR,
  412.     szMsgFileName, 0, 0);
  414. if (pQM != NULL) {
  415. ///////////////////////////////////////////////////////////////////////////////
  416. //  Load message statistics
  417. ///////////////////////////////////////////////////////////////////////////////
  418. if (QueLoadMessageStat(pMQ, pQM) < 0) {
  419. SysLogMessage(LOG_LEV_ERROR,
  420.       "Error loading queue file: '%s%d%s%d%s%s%s%s'n",
  421.       pMQ->pszRootPath, iLevel1,
  422.       SYS_SLASH_STR, iLevel2,
  423.       SYS_SLASH_STR, QUEUE_RSND_DIR,
  424.       SYS_SLASH_STR, szMsgFileName);
  426. QueFreeMessage(pQM);
  427. } else {
  428. ///////////////////////////////////////////////////////////////////////////////
  429. //  Add the file to the resend queue
  430. ///////////////////////////////////////////////////////////////////////////////
  431. SYS_LIST_ADDT(&pQM->LLink, &pMQ->RsndArenaQueue);
  433. ++pMQ->iRsndArenaCount;
  434. }
  436. }
  437. }
  439. } while (MscNextFile(hFileScan, szMsgFileName));
  441. MscCloseFindFile(hFileScan);
  442. }
  444. return (0);
  446. }
  448. static QueueMessage *QueAllocMessage(int iLevel1, int iLevel2, char const *pszQueueDir,
  449.      char const *pszFileName, int iNumTries, time_t tLastTry)
  450. {
  452. QueueMessage *pQM = (QueueMessage *) SysAlloc(sizeof(QueueMessage));
  454. if (pQM == NULL)
  455. return (NULL);
  457. SYS_INIT_LIST_LINK(&pQM->LLink);
  459. pQM->iLevel1 = iLevel1;
  461. pQM->iLevel2 = iLevel2;
  463. pQM->pszQueueDir = pszQueueDir;
  465. pQM->pszFileName = SysStrDup(pszFileName);
  467. pQM->iNumTries = iNumTries;
  469. pQM->tLastTry = tLastTry;
  471. pQM->ulFlags = 0;
  473. return (pQM);
  475. }
  477. static int QueFreeMessage(QueueMessage * pQM)
  478. {
  480. SysFree(pQM->pszFileName);
  482. SysFree(pQM);
  484. return (0);
  486. }
  488. static int QueFreeMessList(SysListHead * pHead)
  489. {
  491. SysListHead *pLLink;
  493. while ((pLLink = SYS_LIST_FIRST(pHead)) != NULL) {
  494. QueueMessage *pQM = SYS_LIST_ENTRY(pLLink, QueueMessage, LLink);
  496. SYS_LIST_DEL(pLLink);
  498. QueFreeMessage(pQM);
  499. }
  501. return (0);
  503. }
  505. char *QueLoadLastLogEntry(char const *pszLogFilePath)
  506. {
  508. FILE *pLogFile = fopen(pszLogFilePath, "rb");
  510. if (pLogFile == NULL) {
  511. ErrSetErrorCode(ERR_FILE_OPEN, pszLogFilePath);
  512. return (NULL);
  513. }
  514. ///////////////////////////////////////////////////////////////////////////////
  515. //  Walk log entries
  516. ///////////////////////////////////////////////////////////////////////////////
  517. unsigned long ulCurrOffset = 0;
  518. unsigned long ulBaseOffset = (unsigned long) -1;
  519. unsigned long ulEndOffset;
  520. unsigned long ulPeekTime;
  521. char szLogLine[1024] = "";
  523. for (;;) {
  524. ulCurrOffset = (unsigned long) ftell(pLogFile);
  526. if (MscFGets(szLogLine, sizeof(szLogLine) - 1, pLogFile) == NULL)
  527. break;
  529. if (sscanf(szLogLine, "[PeekTime] %lu", &ulPeekTime) == 1)
  530. ulBaseOffset = ulCurrOffset;
  531. }
  533. if (ulBaseOffset == (unsigned long) -1) {
  534. fclose(pLogFile);
  535. ErrSetErrorCode(ERR_EMPTY_LOG, pszLogFilePath);
  536. return (NULL);
  537. }
  538. ///////////////////////////////////////////////////////////////////////////////
  539. //  Get end offset ( end of file )
  540. ///////////////////////////////////////////////////////////////////////////////
  541. fseek(pLogFile, 0, SEEK_END);
  542. ulEndOffset = (unsigned long) ftell(pLogFile);
  544. ///////////////////////////////////////////////////////////////////////////////
  545. //  Load last entry
  546. ///////////////////////////////////////////////////////////////////////////////
  547. unsigned int uEntrySize = (unsigned int) (ulEndOffset - ulBaseOffset);
  548. char *pszEntry = (char *) SysAlloc(uEntrySize + 1);
  550. if (pszEntry == NULL) {
  551. fclose(pLogFile);
  552. return (NULL);
  553. }
  555. fseek(pLogFile, ulBaseOffset, SEEK_SET);
  557. if (!fread(pszEntry, uEntrySize, 1, pLogFile)) {
  558. SysFree(pszEntry);
  559. fclose(pLogFile);
  560. ErrSetErrorCode(ERR_FILE_READ, pszLogFilePath);
  561. return (NULL);
  562. }
  564. pszEntry[uEntrySize] = '';
  566. fclose(pLogFile);
  568. return (pszEntry);
  570. }
  572. static int QueLoadMessageStat(MessageQueue * pMQ, QueueMessage * pQM)
  573. {
  574. ///////////////////////////////////////////////////////////////////////////////
  575. //  Build the slog file path
  576. ///////////////////////////////////////////////////////////////////////////////
  577. char szSlogFilePath[SYS_MAX_PATH] = "";
  579. QueGetFilePath(pMQ, pQM, szSlogFilePath, QUEUE_SLOG_DIR);
  581. ///////////////////////////////////////////////////////////////////////////////
  582. //  Try to load message statistics
  583. ///////////////////////////////////////////////////////////////////////////////
  584. FILE *pLogFile = fopen(szSlogFilePath, "rt");
  586. if (pLogFile != NULL) {
  587. int iNumTries = 0;
  588. unsigned long ulLastTime = 0;
  589. unsigned long ulPeekTime;
  590. char szLogLine[1024] = "";
  592. while (MscFGets(szLogLine, sizeof(szLogLine) - 1, pLogFile) != NULL)
  593. if (sscanf(szLogLine, "[PeekTime] %lu", &ulPeekTime) == 1)
  594. ++iNumTries, ulLastTime = ulPeekTime;
  596. fclose(pLogFile);
  598. pQM->iNumTries = iNumTries;
  600. pQM->tLastTry = (time_t) ulLastTime;
  601. }
  603. return (0);
  605. }
  607. static int QueStatMessage(MessageQueue * pMQ, QueueMessage * pQM)
  608. {
  609. ///////////////////////////////////////////////////////////////////////////////
  610. //  Build the slog file path
  611. ///////////////////////////////////////////////////////////////////////////////
  612. char szSlogFilePath[SYS_MAX_PATH] = "";
  614. QueGetFilePath(pMQ, pQM, szSlogFilePath, QUEUE_SLOG_DIR);
  616. FILE *pLogFile = fopen(szSlogFilePath, "a+t");
  618. if (pLogFile == NULL) {
  619. ErrSetErrorCode(ERR_FILE_OPEN, szSlogFilePath);
  620. return (ERR_FILE_OPEN);
  621. }
  622. ///////////////////////////////////////////////////////////////////////////////
  623. //  Dump peek time
  624. ///////////////////////////////////////////////////////////////////////////////
  625. time_t tCurr = time(NULL);
  626. char szTime[128] = "";
  628. MscGetTimeStr(szTime, sizeof(szTime) - 1, tCurr);
  630. fprintf(pLogFile, "[PeekTime] %lu : %sn", (unsigned long) tCurr, szTime);
  632. fclose(pLogFile);
  634. return (0);
  636. }
  638. QMSG_HANDLE QueCreateMessage(QUEUE_HANDLE hQueue)
  639. {
  641. MessageQueue *pMQ = (MessageQueue *) hQueue;
  643. ///////////////////////////////////////////////////////////////////////////////
  644. //  Initialize random number generator
  645. ///////////////////////////////////////////////////////////////////////////////
  646. SRand();
  648. ///////////////////////////////////////////////////////////////////////////////
  649. //  Build message file path
  650. ///////////////////////////////////////////////////////////////////////////////
  651. int iLevel1 = rand() % pMQ->iNumDirsLevel;
  652. int iLevel2 = rand() % pMQ->iNumDirsLevel;
  653. char szSubPath[SYS_MAX_PATH] = "";
  654. char szMsgFilePath[SYS_MAX_PATH] = "";
  656. SysSNPrintf(szSubPath, sizeof(szSubPath) - 1, "%s%d%s%d%s%s",
  657.     pMQ->pszRootPath, iLevel1, SYS_SLASH_STR,
  658.     iLevel2, SYS_SLASH_STR, QUEUE_TEMP_DIR);
  660. if (MscUniqueFile(szSubPath, szMsgFilePath) < 0)
  661. return (INVALID_QMSG_HANDLE);
  663. ///////////////////////////////////////////////////////////////////////////////
  664. //  Extract file name
  665. ///////////////////////////////////////////////////////////////////////////////
  666. char szMsgFileName[SYS_MAX_PATH] = "";
  668. MscGetFileName(szMsgFilePath, szMsgFileName);
  670. ///////////////////////////////////////////////////////////////////////////////
  671. //  Create queue message data
  672. ///////////////////////////////////////////////////////////////////////////////
  673. QueueMessage *pQM = QueAllocMessage(iLevel1, iLevel2, QUEUE_TEMP_DIR,
  674.     szMsgFileName, 0, 0);
  676. if (pQM == NULL)
  677. return (INVALID_QMSG_HANDLE);
  679. return ((QMSG_HANDLE) pQM);
  681. }
  683. static int QueGetFilePath(MessageQueue * pMQ, QueueMessage * pQM, char *pszFilePath,
  684.   char const *pszQueueDir)
  685. {
  687. if (pszQueueDir == NULL)
  688. pszQueueDir = pQM->pszQueueDir;
  690. SysSNPrintf(pszFilePath, SYS_MAX_PATH - 1, "%s%d%s%d%s%s%s%s",
  691.     pMQ->pszRootPath, pQM->iLevel1, SYS_SLASH_STR,
  692.     pQM->iLevel2, SYS_SLASH_STR, pszQueueDir, SYS_SLASH_STR, pQM->pszFileName);
  694. return (0);
  696. }
  698. int QueGetFilePath(QUEUE_HANDLE hQueue, QMSG_HANDLE hMessage, char *pszFilePath,
  699.    char const *pszQueueDir)
  700. {
  702. MessageQueue *pMQ = (MessageQueue *) hQueue;
  703. QueueMessage *pQM = (QueueMessage *) hMessage;
  705. return (QueGetFilePath(pMQ, pQM, pszFilePath, pszQueueDir));
  707. }
  709. int QueCloseMessage(QUEUE_HANDLE hQueue, QMSG_HANDLE hMessage)
  710. {
  712. MessageQueue *pMQ = (MessageQueue *) hQueue;
  713. QueueMessage *pQM = (QueueMessage *) hMessage;
  715. if (pQM->ulFlags & QUMF_DELETED)
  716. QueDoMessageCleanup(hQueue, hMessage);
  718. QueFreeMessage(pQM);
  720. return (0);
  722. }
  724. QMSG_HANDLE QueGetHandle(QUEUE_HANDLE hQueue, int iLevel1, int iLevel2, char const *pszQueueDir,
  725.  char const *pszFileName)
  726. {
  728. MessageQueue *pMQ = (MessageQueue *) hQueue;
  729. QueueMessage *pQM = QueAllocMessage(iLevel1, iLevel2, pszQueueDir, pszFileName, 0, 0);
  731. if (pQM == NULL)
  732. return (INVALID_QMSG_HANDLE);
  734. ///////////////////////////////////////////////////////////////////////////////
  735. //  Load message statistics
  736. ///////////////////////////////////////////////////////////////////////////////
  737. QueLoadMessageStat(pMQ, pQM);
  739. return ((QMSG_HANDLE) pQM);
  741. }
  743. char const *QueGetFileName(QMSG_HANDLE hMessage)
  744. {
  746. QueueMessage *pQM = (QueueMessage *) hMessage;
  748. return (pQM->pszFileName);
  750. }
  752. int QueGetLevel1(QMSG_HANDLE hMessage)
  753. {
  755. QueueMessage *pQM = (QueueMessage *) hMessage;
  757. return (pQM->iLevel1);
  759. }
  761. int QueGetLevel2(QMSG_HANDLE hMessage)
  762. {
  764. QueueMessage *pQM = (QueueMessage *) hMessage;
  766. return (pQM->iLevel2);
  768. }
  770. int QueGetTryCount(QMSG_HANDLE hMessage)
  771. {
  773. QueueMessage *pQM = (QueueMessage *) hMessage;
  775. return (pQM->iNumTries);
  777. }
  779. time_t QueGetLastTryTime(QMSG_HANDLE hMessage)
  780. {
  782. QueueMessage *pQM = (QueueMessage *) hMessage;
  784. return (pQM->tLastTry);
  786. }
  788. time_t QueGetMessageNextOp(QUEUE_HANDLE hQueue, QMSG_HANDLE hMessage)
  789. {
  791. MessageQueue *pMQ = (MessageQueue *) hQueue;
  792. QueueMessage *pQM = (QueueMessage *) hMessage;
  794. return (pQM->tLastTry + QueNextRetryOp(pQM->iNumTries, (unsigned int) pMQ->iRetryTimeout,
  795.        (unsigned int) pMQ->iRetryIncrRatio));
  797. }
  799. int QueInitMessageStats(QUEUE_HANDLE hQueue, QMSG_HANDLE hMessage)
  800. {
  802. MessageQueue *pMQ = (MessageQueue *) hQueue;
  803. QueueMessage *pQM = (QueueMessage *) hMessage;
  805. ///////////////////////////////////////////////////////////////////////////////
  806. //  Clean 'slog' file
  807. ///////////////////////////////////////////////////////////////////////////////
  808. char szQueueFilePath[SYS_MAX_PATH] = "";
  810. QueGetFilePath(pMQ, pQM, szQueueFilePath, QUEUE_SLOG_DIR);
  812. CheckRemoveFile(szQueueFilePath);
  814. ///////////////////////////////////////////////////////////////////////////////
  815. //  Init message statistics
  816. ///////////////////////////////////////////////////////////////////////////////
  817. pQM->iNumTries = 0;
  819. pQM->tLastTry = 0;
  821. return (0);
  823. }
  825. static int QueDoMessageCleanup(QUEUE_HANDLE hQueue, QMSG_HANDLE hMessage)
  826. {
  828. MessageQueue *pMQ = (MessageQueue *) hQueue;
  829. QueueMessage *pQM = (QueueMessage *) hMessage;
  830. char szQueueFilePath[SYS_MAX_PATH] = "";
  832. if (pQM->ulFlags & QUMF_FREEZE) {
  833. ///////////////////////////////////////////////////////////////////////////////
  834. //  Move message file
  835. ///////////////////////////////////////////////////////////////////////////////
  836. char szTargetFile[SYS_MAX_PATH] = "";
  838. QueGetFilePath(pMQ, pQM, szQueueFilePath);
  839. QueGetFilePath(pMQ, pQM, szTargetFile, QUEUE_FROZ_DIR);
  841. if (SysMoveFile(szQueueFilePath, szTargetFile) < 0)
  842. return (ErrGetErrorCode());
  844. ///////////////////////////////////////////////////////////////////////////////
  845. //  Change message location
  846. ///////////////////////////////////////////////////////////////////////////////
  847. pQM->pszQueueDir = QUEUE_FROZ_DIR;
  848. } else {
  849. ///////////////////////////////////////////////////////////////////////////////
  850. //  Clean message file
  851. ///////////////////////////////////////////////////////////////////////////////
  852. QueGetFilePath(pMQ, pQM, szQueueFilePath);
  854. CheckRemoveFile(szQueueFilePath);
  856. ///////////////////////////////////////////////////////////////////////////////
  857. //  Clean 'info' file
  858. ///////////////////////////////////////////////////////////////////////////////
  859. QueGetFilePath(pMQ, pQM, szQueueFilePath, QUEUE_INFO_DIR);
  861. CheckRemoveFile(szQueueFilePath);
  863. ///////////////////////////////////////////////////////////////////////////////
  864. //  Clean 'slog' file
  865. ///////////////////////////////////////////////////////////////////////////////
  866. QueGetFilePath(pMQ, pQM, szQueueFilePath, QUEUE_SLOG_DIR);
  868. CheckRemoveFile(szQueueFilePath);
  870. }
  872. ///////////////////////////////////////////////////////////////////////////////
  873. //  Clean 'temp' file
  874. ///////////////////////////////////////////////////////////////////////////////
  875. QueGetFilePath(pMQ, pQM, szQueueFilePath, QUEUE_TEMP_DIR);
  877. CheckRemoveFile(szQueueFilePath);
  879. ///////////////////////////////////////////////////////////////////////////////
  880. //  Clean 'cust' file
  881. ///////////////////////////////////////////////////////////////////////////////
  882. QueGetFilePath(pMQ, pQM, szQueueFilePath, QUEUE_CUST_DIR);
  884. CheckRemoveFile(szQueueFilePath);
  886. ///////////////////////////////////////////////////////////////////////////////
  887. //  Clean 'mprc' file
  888. ///////////////////////////////////////////////////////////////////////////////
  889. QueGetFilePath(pMQ, pQM, szQueueFilePath, QUEUE_MPRC_DIR);
  891. CheckRemoveFile(szQueueFilePath);
  893. return (0);
  895. }
  897. int QueCleanupMessage(QUEUE_HANDLE hQueue, QMSG_HANDLE hMessage, bool bFreeze)
  898. {
  900. MessageQueue *pMQ = (MessageQueue *) hQueue;
  901. QueueMessage *pQM = (QueueMessage *) hMessage;
  903. pQM->ulFlags |= QUMF_DELETED;
  904. if (bFreeze)
  905. pQM->ulFlags |= QUMF_FREEZE;
  907. return (0);
  909. }
  911. int QueCommitMessage(QUEUE_HANDLE hQueue, QMSG_HANDLE hMessage)
  912. {
  914. MessageQueue *pMQ = (MessageQueue *) hQueue;
  915. QueueMessage *pQM = (QueueMessage *) hMessage;
  917. ///////////////////////////////////////////////////////////////////////////////
  918. //  Move message file ( if not in mess )
  919. ///////////////////////////////////////////////////////////////////////////////
  920. if (strcmp(pQM->pszQueueDir, QUEUE_MESS_DIR) != 0) {
  921. char szSourceFile[SYS_MAX_PATH] = "";
  922. char szTargetFile[SYS_MAX_PATH] = "";
  924. QueGetFilePath(pMQ, pQM, szSourceFile);
  925. QueGetFilePath(pMQ, pQM, szTargetFile, QUEUE_MESS_DIR);
  927. if (SysMoveFile(szSourceFile, szTargetFile) < 0)
  928. return (ErrGetErrorCode());
  930. ///////////////////////////////////////////////////////////////////////////////
  931. //  Change message location
  932. ///////////////////////////////////////////////////////////////////////////////
  933. pQM->pszQueueDir = QUEUE_MESS_DIR;
  934. }
  935. ///////////////////////////////////////////////////////////////////////////////
  936. //  Unmask temporary flags
  937. ///////////////////////////////////////////////////////////////////////////////
  938. pQM->ulFlags = QUE_MASK_TMPFLAGS(pQM->ulFlags);
  940. ///////////////////////////////////////////////////////////////////////////////
  941. //  Add to queue
  942. ///////////////////////////////////////////////////////////////////////////////
  943. if (QueAddNew(pMQ, pQM) < 0)
  944. return (ErrGetErrorCode());
  946. return (0);
  948. }
  950. static int QueAddNew(MessageQueue * pMQ, QueueMessage * pQM)
  951. {
  952. ///////////////////////////////////////////////////////////////////////////////
  953. //  Add the queue entry
  954. ///////////////////////////////////////////////////////////////////////////////
  955. if (SysLockMutex(pMQ->hMutex, SYS_INFINITE_TIMEOUT) < 0)
  956. return (ErrGetErrorCode());
  958. SYS_LIST_ADDT(&pQM->LLink, &pMQ->ReadyQueue);
  960. ++pMQ->iReadyCount;
  962. SysSetEvent(pMQ->hReadyEvent);
  964. SysUnlockMutex(pMQ->hMutex);
  966. return (0);
  968. }
  970. static bool QueMessageExpired(MessageQueue * pMQ, QueueMessage * pQM)
  971. {
  973. return (pQM->iNumTries >= pMQ->iMaxRetry);
  975. }
  977. static time_t QueNextRetryOp(int iNumTries, unsigned int uRetryTimeout,
  978.      unsigned int uRetryIncrRatio)
  979. {
  981. unsigned int uNextOp = uRetryTimeout;
  983. if (uRetryIncrRatio != 0)
  984. for (int ii = 1; ii < iNumTries; ii++)
  985. uNextOp += uNextOp / uRetryIncrRatio;
  987. return ((time_t) uNextOp);
  989. }
  991. static bool QueMessageReadyToSend(MessageQueue * pMQ, QueueMessage * pQM)
  992. {
  994. return (time(NULL) > (pQM->tLastTry +
  995.       QueNextRetryOp(pQM->iNumTries, (unsigned int) pMQ->iRetryTimeout,
  996.      (unsigned int) pMQ->iRetryIncrRatio)));
  998. }
  1000. int QueResendMessage(QUEUE_HANDLE hQueue, QMSG_HANDLE hMessage)
  1001. {
  1003. MessageQueue *pMQ = (MessageQueue *) hQueue;
  1004. QueueMessage *pQM = (QueueMessage *) hMessage;
  1006. ///////////////////////////////////////////////////////////////////////////////
  1007. //  Check for message expired
  1008. ///////////////////////////////////////////////////////////////////////////////
  1009. if (QueMessageExpired(pMQ, pQM)) {
  1010. ErrSetErrorCode(ERR_SPOOL_FILE_EXPIRED);
  1011. return (ERR_SPOOL_FILE_EXPIRED);
  1012. }
  1013. ///////////////////////////////////////////////////////////////////////////////
  1014. //  Move message file ( if not in rsnd dir )
  1015. ///////////////////////////////////////////////////////////////////////////////
  1016. if (strcmp(pQM->pszQueueDir, QUEUE_RSND_DIR) != 0) {
  1017. char szSourceFile[SYS_MAX_PATH] = "";
  1018. char szTargetFile[SYS_MAX_PATH] = "";
  1020. QueGetFilePath(pMQ, pQM, szSourceFile);
  1021. QueGetFilePath(pMQ, pQM, szTargetFile, QUEUE_RSND_DIR);
  1023. if (SysMoveFile(szSourceFile, szTargetFile) < 0)
  1024. return (ErrGetErrorCode());
  1026. ///////////////////////////////////////////////////////////////////////////////
  1027. //  Change message location
  1028. ///////////////////////////////////////////////////////////////////////////////
  1029. pQM->pszQueueDir = QUEUE_RSND_DIR;
  1030. }
  1031. ///////////////////////////////////////////////////////////////////////////////
  1032. //  Unmask temporary flags
  1033. ///////////////////////////////////////////////////////////////////////////////
  1034. pQM->ulFlags = QUE_MASK_TMPFLAGS(pQM->ulFlags);
  1036. ///////////////////////////////////////////////////////////////////////////////
  1037. //  Add to queue
  1038. ///////////////////////////////////////////////////////////////////////////////
  1039. if (QueAddRsnd(pMQ, pQM) < 0)
  1040. return (ErrGetErrorCode());
  1042. return (0);
  1044. }
  1046. static int QueAddRsnd(MessageQueue * pMQ, QueueMessage * pQM)
  1047. {
  1048. ///////////////////////////////////////////////////////////////////////////////
  1049. //  Add the queue entry
  1050. ///////////////////////////////////////////////////////////////////////////////
  1051. if (SysLockMutex(pMQ->hMutex, SYS_INFINITE_TIMEOUT) < 0)
  1052. return (ErrGetErrorCode());
  1054. SYS_LIST_ADDT(&pQM->LLink, &pMQ->RsndArenaQueue);
  1056. ++pMQ->iRsndArenaCount;
  1058. SysUnlockMutex(pMQ->hMutex);
  1060. return (0);
  1062. }
  1064. QMSG_HANDLE QueExtractMessage(QUEUE_HANDLE hQueue, int iTimeout)
  1065. {
  1067. MessageQueue *pMQ = (MessageQueue *) hQueue;
  1069. ///////////////////////////////////////////////////////////////////////////////
  1070. //  Wait for message available
  1071. ///////////////////////////////////////////////////////////////////////////////
  1072. if (SysWaitEvent(pMQ->hReadyEvent, iTimeout) < 0)
  1073. return (INVALID_QMSG_HANDLE);
  1075. if (SysLockMutex(pMQ->hMutex, SYS_INFINITE_TIMEOUT) < 0)
  1076. return (INVALID_QMSG_HANDLE);
  1078. ///////////////////////////////////////////////////////////////////////////////
  1079. //  Get the first message of the queue
  1080. ///////////////////////////////////////////////////////////////////////////////
  1081. SysListHead *pLLink = SYS_LIST_FIRST(&pMQ->ReadyQueue);
  1083. if (pLLink == NULL) {
  1084. SysUnlockMutex(pMQ->hMutex);
  1085. return (INVALID_QMSG_HANDLE);
  1086. }
  1087. ///////////////////////////////////////////////////////////////////////////////
  1088. //  Remove the message from the list
  1089. ///////////////////////////////////////////////////////////////////////////////
  1090. SYS_LIST_DEL(pLLink);
  1092. ///////////////////////////////////////////////////////////////////////////////
  1093. //  Decrement message count by resetting the event if no more messages are in
  1094. ///////////////////////////////////////////////////////////////////////////////
  1095. if (--pMQ->iReadyCount == 0)
  1096. SysResetEvent(pMQ->hReadyEvent);
  1098. SysUnlockMutex(pMQ->hMutex);
  1100. ///////////////////////////////////////////////////////////////////////////////
  1101. //  Get queue message pointer
  1102. ///////////////////////////////////////////////////////////////////////////////
  1103. QueueMessage *pQM = SYS_LIST_ENTRY(pLLink, QueueMessage, LLink);
  1105. ///////////////////////////////////////////////////////////////////////////////
  1106. //  Update message statistics
  1107. ///////////////////////////////////////////////////////////////////////////////
  1108. ++pQM->iNumTries;
  1110. pQM->tLastTry = time(NULL);
  1112. ///////////////////////////////////////////////////////////////////////////////
  1113. //  Update log file
  1114. ///////////////////////////////////////////////////////////////////////////////
  1115. QueStatMessage(pMQ, pQM);
  1117. return ((QMSG_HANDLE) pQM);
  1119. }
  1121. static unsigned int QueRsndThread(void *pThreadData)
  1122. {
  1124. MessageQueue *pMQ = (MessageQueue *) pThreadData;
  1126. ///////////////////////////////////////////////////////////////////////////////
  1127. //  Scan the rsnd arena to find out ready-to-send messages
  1128. ///////////////////////////////////////////////////////////////////////////////
  1129. int iElapsedTime = 0;
  1131. while ((pMQ->ulFlags & QUEF_SHUTDOWN) == 0) {
  1132. SysSleep(QUE_ARENA_SCAN_WAIT);
  1134. iElapsedTime += QUE_ARENA_SCAN_WAIT;
  1136. if (iElapsedTime > QUE_ARENA_SCAN_INTERVAL) {
  1137. iElapsedTime = 0;
  1139. ///////////////////////////////////////////////////////////////////////////////
  1140. //  Scan rsnd arena to prepare messages to resend
  1141. ///////////////////////////////////////////////////////////////////////////////
  1142. QueScanRsndArena(pMQ);
  1143. }
  1145. }
  1147. ///////////////////////////////////////////////////////////////////////////////
  1148. //  Reset the shutdown flag
  1149. ///////////////////////////////////////////////////////////////////////////////
  1150. pMQ->ulFlags &= ~QUEF_SHUTDOWN;
  1152. return (0);
  1154. }
  1156. static int QueScanRsndArena(MessageQueue * pMQ)
  1157. {
  1159. if (SysLockMutex(pMQ->hMutex, SYS_INFINITE_TIMEOUT) < 0)
  1160. return (ErrGetErrorCode());
  1162. SysListHead *pLLink;
  1164. SYS_LIST_FOR_EACH(pLLink, &pMQ->RsndArenaQueue) {
  1165. QueueMessage *pQM = SYS_LIST_ENTRY(pLLink, QueueMessage, LLink);
  1167. if (QueMessageReadyToSend(pMQ, pQM)) {
  1168. ///////////////////////////////////////////////////////////////////////////////
  1169. //  Set the list pointer to the next item
  1170. ///////////////////////////////////////////////////////////////////////////////
  1171. pLLink = pLLink->pPrev;
  1173. ///////////////////////////////////////////////////////////////////////////////
  1174. //  Remove item from resend arena
  1175. ///////////////////////////////////////////////////////////////////////////////
  1176. SYS_LIST_DEL(&pQM->LLink);
  1178. --pMQ->iRsndArenaCount;
  1180. ///////////////////////////////////////////////////////////////////////////////
  1181. //  Add item from resend queue
  1182. ///////////////////////////////////////////////////////////////////////////////
  1183. SYS_LIST_ADDT(&pQM->LLink, &pMQ->ReadyQueue);
  1185. ++pMQ->iReadyCount;
  1187. }
  1189. }
  1191. ///////////////////////////////////////////////////////////////////////////////
  1192. //  If the count of rsnd queue is not zero, set the event
  1193. ///////////////////////////////////////////////////////////////////////////////
  1194. if (pMQ->iReadyCount > 0)
  1195. SysSetEvent(pMQ->hReadyEvent);
  1197. SysUnlockMutex(pMQ->hMutex);
  1199. return (0);
  1201. }
  1203. int QueCheckMessage(QUEUE_HANDLE hQueue, QMSG_HANDLE hMessage)
  1204. {
  1206. MessageQueue *pMQ = (MessageQueue *) hQueue;
  1207. QueueMessage *pQM = (QueueMessage *) hMessage;
  1208. char szQueueFilePath[SYS_MAX_PATH] = "";
  1210. if (pQM->ulFlags & QUMF_DELETED) {
  1211. ErrSetErrorCode(ERR_MESSAGE_DELETED);
  1212. return (ERR_MESSAGE_DELETED);
  1213. }
  1215. QueGetFilePath(hQueue, hMessage, szQueueFilePath, QUEUE_MESS_DIR);
  1217. if (!SysExistFile(szQueueFilePath)) {
  1218. QueGetFilePath(hQueue, hMessage, szQueueFilePath, QUEUE_RSND_DIR);
  1220. if (!SysExistFile(szQueueFilePath)) {
  1221. ErrSetErrorCode(ERR_NO_MESSAGE_FILE);
  1222. return (ERR_NO_MESSAGE_FILE);
  1223. }
  1224. }
  1226. return (0);
  1228. }
  1230. static bool QueMessageDestMatch(MessageQueue * pMQ, QueueMessage * pQM,
  1231. char const *pszAddressMatch)
  1232. {
  1233. ///////////////////////////////////////////////////////////////////////////////
  1234. //  Get the queue file path
  1235. ///////////////////////////////////////////////////////////////////////////////
  1236. char szQueueFilePath[SYS_MAX_PATH] = "";
  1238. QueGetFilePath(pMQ, pQM, szQueueFilePath);
  1240. ///////////////////////////////////////////////////////////////////////////////
  1241. //  Load queue file header
  1242. ///////////////////////////////////////////////////////////////////////////////
  1243. SpoolFileHeader SFH;
  1245. if (USmlLoadSpoolFileHeader(szQueueFilePath, SFH) < 0)
  1246. return (false);
  1248. bool bAddressMatch = false;
  1250. if (strchr(pszAddressMatch, '@') == NULL) {
  1251. ///////////////////////////////////////////////////////////////////////////////
  1252. //  RFC style ETRN ( domain based )
  1253. ///////////////////////////////////////////////////////////////////////////////
  1254. char szDestUser[MAX_ADDR_NAME] = "";
  1255. char szDestDomain[MAX_ADDR_NAME] = "";
  1257. if ((StrStringsCount(SFH.ppszRcpt) < 1) ||
  1258.     (USmtpSplitEmailAddr(SFH.ppszRcpt[0], szDestUser, szDestDomain) < 0)) {
  1259. USmlCleanupSpoolFileHeader(SFH);
  1260. return (false);
  1261. }
  1263. bAddressMatch = (StrIWildMatch(szDestDomain, pszAddressMatch) != 0);
  1265. } else {
  1266. ///////////////////////////////////////////////////////////////////////////////
  1267. //  XMail style ETRN ( email based )
  1268. ///////////////////////////////////////////////////////////////////////////////
  1269. bAddressMatch = (StrIWildMatch(SFH.ppszRcpt[0], pszAddressMatch) != 0);
  1271. }
  1273. USmlCleanupSpoolFileHeader(SFH);
  1275. return (bAddressMatch);
  1277. }
  1279. int QueFlushRsndArena(QUEUE_HANDLE hQueue, char const *pszAddressMatch)
  1280. {
  1282. MessageQueue *pMQ = (MessageQueue *) hQueue;
  1284. if (SysLockMutex(pMQ->hMutex, SYS_INFINITE_TIMEOUT) < 0)
  1285. return (ErrGetErrorCode());
  1287. SysListHead *pLLink;
  1289. SYS_LIST_FOR_EACH(pLLink, &pMQ->RsndArenaQueue) {
  1290. QueueMessage *pQM = SYS_LIST_ENTRY(pLLink, QueueMessage, LLink);
  1292. if ((pszAddressMatch == NULL) || QueMessageDestMatch(pMQ, pQM, pszAddressMatch)) {
  1293. ///////////////////////////////////////////////////////////////////////////////
  1294. //  Set the list pointer to the next item
  1295. ///////////////////////////////////////////////////////////////////////////////
  1296. pLLink = pLLink->pPrev;
  1298. ///////////////////////////////////////////////////////////////////////////////
  1299. //  Remove item from resend arena
  1300. ///////////////////////////////////////////////////////////////////////////////
  1301. SYS_LIST_DEL(&pQM->LLink);
  1303. --pMQ->iRsndArenaCount;
  1305. ///////////////////////////////////////////////////////////////////////////////
  1306. //  Add item from resend queue
  1307. ///////////////////////////////////////////////////////////////////////////////
  1308. SYS_LIST_ADDT(&pQM->LLink, &pMQ->ReadyQueue);
  1310. ++pMQ->iReadyCount;
  1312. }
  1314. }
  1316. ///////////////////////////////////////////////////////////////////////////////
  1317. //  If the count of rsnd queue is not zero, set the event
  1318. ///////////////////////////////////////////////////////////////////////////////
  1319. if (pMQ->iReadyCount > 0)
  1320. SysSetEvent(pMQ->hReadyEvent);
  1322. SysUnlockMutex(pMQ->hMutex);
  1324. return (0);
  1326. }