Add queue hold release and listing controls

include/nmap.h

@@ -353,6 +353,7 @@ typedef enum _NMAPMessageFlags {
 #define	Q_OUTGOING				7
 #define	Q_RTS						8
 #define	Q_DIGEST					9
+#define Q_HOLD						999
 
 /* Delivery Status Notification and other Control File Flags */
 #define	DSN_FLAGS				(0x0000001F)

src/agents/queue/messages.h

@@ -49,6 +49,9 @@
 
 #define NMAP_QINFO_COMMAND "QINFO"
 
+#define NMAP_QLIST_COMMAND "QLIST"
+#define NMAP_QLIST_HELP "QLIST - Lists committed message queue entries.\r\n"
+
 #define NMAP_QMOD_FROM_COMMAND "QMOD FROM"
 
 #define NMAP_QMOD_FLAGS_COMMAND "QMOD FLAGS"

src/agents/queue/queue.c

@@ -2599,6 +2599,12 @@ CheckQueue(void *queueIn)
 
             found++;
             QueueFormat(path, "%.3s%.7s", dirEntry->d_nameDOS + 9, dirEntry->d_nameDOS + 1);
+
+            /* Held messages remain committed and inspectable, but neither
+             * the periodic monitor nor QFLUSH may process them. */
+            if (atoi(path) == Q_HOLD) {
+                continue;
+            }
                 
             /* skip recently-touched messages that are in the outgoing
              * queue (queue 7) */
@@ -2795,7 +2801,7 @@ CreateQueueThreads(BOOL failed)
                         count += 2;
                     } else if (failed) {
                         QueueFormat(path, "%s/c%s", Conf.spoolPath, dirEntry->d_nameDOS + 1);
-                        for (i = 0; i < 10; i++) {
+                        for (i = 0; i <= Q_HOLD; i++) {
                             if (strlen(path) >= 3) {
                                 size_t extensionOffset = strlen(path) - 3;
                                 QueueFormatBuffer(path + extensionOffset,
@@ -2809,7 +2815,7 @@ CreateQueueThreads(BOOL failed)
                             }
                         }
                     
-                        if (i == 10) {
+                        if (i > Q_HOLD) {
                             /* Check for matching C file */
                             fprintf(killFile, "%s/d%s\r\n", Conf.spoolPath, dirEntry->d_nameDOS + 1);
 
@@ -3707,6 +3713,99 @@ CommandQinfo(void *param)
         (intmax_t)(sb.st_size - count))); /* BodySize */
 }
 
+static int
+QueueListEntryCompare(const void *left, const void *right)
+{
+    const char *const *leftEntry = left;
+    const char *const *rightEntry = right;
+
+    return strcmp(*leftEntry, *rightEntry);
+}
+
+int
+CommandQlist(void *param)
+{
+    QueueClient *client = (QueueClient *)param;
+    XplDir *dirP;
+    XplDir *dirEntry;
+    GPtrArray *entries;
+    GString *listing;
+    unsigned int i;
+    int ccode;
+
+    if (client->buffer[5] != '\0') {
+        return ConnWrite(client->conn, MSG3010BADARGC,
+                         sizeof(MSG3010BADARGC) - 1);
+    }
+
+    entries = g_ptr_array_new_with_free_func(g_free);
+    listing = g_string_new(NULL);
+    dirP = XplOpenDir(Conf.spoolPath);
+    if (!dirP) {
+        g_ptr_array_free(entries, TRUE);
+        g_string_free(listing, TRUE);
+        return ConnWrite(client->conn, MSG4224CANTREAD,
+                         sizeof(MSG4224CANTREAD) - 1);
+    }
+
+    while ((dirEntry = XplReadDir(dirP)) != NULL) {
+        const char *name = dirEntry->d_nameDOS;
+        char queueID[12];
+        char dataPath[XPL_MAX_PATH + 1];
+        struct stat dataStat;
+        size_t position;
+        BOOL valid = TRUE;
+
+        if (strlen(name) != 12 || toupper(name[0]) != 'C' ||
+            name[8] != '.') {
+            continue;
+        }
+        for (position = 1; position < 8; position++) {
+            if (!isxdigit((unsigned char)name[position])) {
+                valid = FALSE;
+                break;
+            }
+        }
+        for (position = 9; valid && position < 12; position++) {
+            if (!isdigit((unsigned char)name[position])) {
+                valid = FALSE;
+            }
+        }
+        if (!valid) {
+            continue;
+        }
+
+        snprintf(queueID, sizeof(queueID), "%.3s-%.7s", name + 9, name + 1);
+        QueueFormat(dataPath, "%s/d%.7s.msg", Conf.spoolPath, name + 1);
+        if (stat(dataPath, &dataStat) != 0) {
+            continue;
+        }
+        g_ptr_array_add(
+            entries,
+            g_strdup_printf("%s %" PRIdMAX "\r\n", queueID,
+                            (intmax_t)dataStat.st_size));
+    }
+    XplCloseDir(dirP);
+
+    g_ptr_array_sort(entries, QueueListEntryCompare);
+    for (i = 0; i < entries->len; i++) {
+        g_string_append(listing, g_ptr_array_index(entries, i));
+    }
+
+    ccode = ConnWriteF(client->conn, "2024 %lu Queue list follows\r\n",
+                       (unsigned long)listing->len);
+    if (ccode != -1 && listing->len) {
+        ccode = ConnWrite(client->conn, listing->str, listing->len);
+    }
+    if (ccode != -1) {
+        ccode = ConnWrite(client->conn, MSG1000OK, sizeof(MSG1000OK) - 1);
+    }
+
+    g_ptr_array_free(entries, TRUE);
+    g_string_free(listing, TRUE);
+    return ccode;
+}
+
 int
 CommandQmodFrom(void *param)
 {
@@ -3942,6 +4041,8 @@ int
 CommandQmove(void *param)
 {
     int ccode;
+    unsigned long entryID;
+    unsigned long target;
     char *ptr;
     char *ptr2;
     char path[XPL_MAX_PATH + 1];
@@ -3966,20 +4067,45 @@ CommandQmove(void *param)
         return(ConnWrite(client->conn, MSG3010BADARGC, sizeof(MSG3010BADARGC) - 1));
     }
 
+    if (!*ptr2) {
+        return(ConnWrite(client->conn, MSG3010BADARGC,
+                         sizeof(MSG3010BADARGC) - 1));
+    }
+    {
+        char *end = NULL;
+        target = strtoul(ptr2, &end, 10);
+        if (!end || *end != '\0') {
+            return(ConnWrite(client->conn, MSG3010BADARGC,
+                             sizeof(MSG3010BADARGC) - 1));
+        }
+    }
+    if (target > Q_HOLD) {
+        return(ConnWrite(client->conn, MSG4000OUTOFRANGE,
+                         sizeof(MSG4000OUTOFRANGE) - 1));
+    }
+
+    entryID = strtoul(ptr + 4, NULL, 16);
+    if (!SpoolEntryIDLock(entryID)) {
+        return(ConnWrite(client->conn, MSG4000CANTUNLOCKENTRY,
+                         sizeof(MSG4000CANTUNLOCKENTRY) - 1));
+    }
+
     QueueFormat(client->path, "%s/c%s.%s", Conf.spoolPath, ptr + 4, ptr);
     if (stat(client->path, &sb) == 0) {
-        QueueFormat(path, "%s/c%s.%03d", Conf.spoolPath, ptr + 4, atoi(ptr));
+        QueueFormat(path, "%s/c%s.%03lu", Conf.spoolPath, ptr + 4, target);
 
-        RENAME_CHECK(client->path, path);
-
-        ccode = ConnWrite(client->conn, MSG1000OK, sizeof(MSG1000OK) - 1);
+        if (rename(client->path, path) == 0) {
+            ccode = ConnWriteF(client->conn, "1000 %03lu-%s OK\r\n",
+                               target, ptr + 4);
+        } else {
+            ccode = ConnWrite(client->conn, MSG4224CANTREAD,
+                              sizeof(MSG4224CANTREAD) - 1);
+        }
     } else {
-        QueueFormat(client->path, "%s/d%s.msg", Conf.spoolPath, ptr + 4);
-        UNLINK_CHECK(client->path);
-
         ccode = ConnWrite(client->conn, MSG4224NOENTRY, sizeof(MSG4224NOENTRY) - 1);
     }
 
+    SpoolEntryIDUnlock(entryID);
     return(ccode);
 }
 

src/agents/queue/queue.h

@@ -122,6 +122,7 @@ int CommandQend(void *param);
 int CommandQgrep(void *param);
 int CommandQhead(void *param);
 int CommandQinfo(void *param);
+int CommandQlist(void *param);
 int CommandQmodFrom(void *param);
 int CommandQmodFlags(void *param);
 int CommandQmodLocal(void *param);

src/agents/queue/queued.c

@@ -74,6 +74,7 @@ static ProtocolCommand commands[] = {
     { NMAP_QGREP_COMMAND, NMAP_HELP_NOT_DEFINED, sizeof(NMAP_QGREP_COMMAND) - 1, CommandQgrep, NULL, NULL, NULL, BlackCommand },
     { NMAP_QHEAD_COMMAND, NMAP_HELP_NOT_DEFINED, sizeof(NMAP_QHEAD_COMMAND) - 1, CommandQhead, NULL, NULL, NULL, BlackCommand },
     { NMAP_QINFO_COMMAND, NMAP_HELP_NOT_DEFINED, sizeof(NMAP_QINFO_COMMAND) - 1, CommandQinfo, NULL, NULL, NULL, BlackCommand },
+    { NMAP_QLIST_COMMAND, NMAP_QLIST_HELP, sizeof(NMAP_QLIST_COMMAND) - 1, CommandQlist, NULL, NULL, NULL, BlackCommand },
     { NMAP_QMOD_FROM_COMMAND, NMAP_HELP_NOT_DEFINED, sizeof(NMAP_QMOD_FROM_COMMAND) - 1, CommandQmodFrom, NULL, NULL, NULL, BlackCommand },
     { NMAP_QMOD_FLAGS_COMMAND, NMAP_HELP_NOT_DEFINED, sizeof(NMAP_QMOD_FLAGS_COMMAND) - 1, CommandQmodFlags, NULL, NULL, NULL, BlackCommand },
     { NMAP_QMOD_LOCAL_COMMAND, NMAP_HELP_NOT_DEFINED, sizeof(NMAP_QMOD_LOCAL_COMMAND) - 1, CommandQmodLocal, NULL, NULL, NULL, BlackCommand },

src/apps/queuetool/bongo-queuetool

@@ -94,6 +94,24 @@ class ShowCommand(Command):
         data = queue.RetrieveInfo(args[0])
         sys.stdout.buffer.write(data)
 
+class ListCommand(Command):
+    log = logging.getLogger("Bongo.QueueTool")
+
+    def __init__(self):
+        Command.__init__(self, "list", aliases=["l"],
+                         summary="List committed queue entries",
+                         usage="%prog %cmd")
+
+    def Run(self, options, args):
+        if args:
+            self.error("list does not accept arguments")
+        queue = QueueClient(host=options.host, port=options.port)
+        try:
+            for queue_id, size in queue.List():
+                print(f"{queue_id} {size}")
+        finally:
+            queue.Quit()
+
 class MessageCommand(Command):
     log = logging.getLogger("Bongo.QueueTool")
 
@@ -181,13 +199,58 @@ class DeleteCommand(Command):
         finally:
             queue.Quit()
 
+class HoldCommand(Command):
+    log = logging.getLogger("Bongo.QueueTool")
+
+    def __init__(self):
+        Command.__init__(self, "hold",
+                         summary="Move a committed entry to the hold queue",
+                         usage="%prog %cmd <queue-id>")
+
+    def Run(self, options, args):
+        if len(args) != 1:
+            self.error("exactly one queue ID is required")
+        queue = QueueClient(host=options.host, port=options.port)
+        try:
+            response = queue.Move(args[0], 999)
+            print(response.message.split(" ", 1)[0])
+        finally:
+            queue.Quit()
+
+class ReleaseCommand(Command):
+    log = logging.getLogger("Bongo.QueueTool")
+
+    def __init__(self):
+        Command.__init__(
+            self, "release",
+            summary="Release a held entry and process it immediately",
+            usage="%prog %cmd <queue-id> [<target>]")
+
+    def Run(self, options, args):
+        if len(args) < 1 or len(args) > 2:
+            self.error("a queue ID and optional target are required")
+        if not args[0].startswith("999-"):
+            self.error("release requires an entry from queue 999")
+        target = int(args[1]) if len(args) == 2 else 0
+        queue = QueueClient(host=options.host, port=options.port)
+        try:
+            response = queue.Move(args[0], target)
+            released_id = response.message.split(" ", 1)[0]
+            queue.Run(released_id)
+            print(released_id)
+        finally:
+            queue.Quit()
+
 parser.add_command(FlushCommand(), "Queue Management")
 parser.add_command(SpaceCommand(), "Queue Management")
+parser.add_command(ListCommand(), "Queue Management")
 parser.add_command(ShowCommand(), "Queue Management")
 parser.add_command(MessageCommand(), "Queue Management")
 parser.add_command(HoldLocalCommand(), "Queue Recovery and Diagnostics")
 parser.add_command(DeliverLocalCommand(), "Queue Recovery and Diagnostics")
 parser.add_command(DeleteCommand(), "Queue Recovery and Diagnostics")
+parser.add_command(HoldCommand(), "Queue Recovery and Diagnostics")
+parser.add_command(ReleaseCommand(), "Queue Recovery and Diagnostics")
 
 if __name__ == '__main__':
     (command, options, args) = parser.parse_args()

src/apps/queuetool/tests/test_queue_client.py

@@ -111,6 +111,38 @@ class QueueClientTests(unittest.TestCase):
 
         self.assertEqual(client.stream.commands, ["QDELE 999-deadbee"])
 
+    def test_list_reads_declared_payload(self):
+        data = b"007-deadbee 42\r\n999-cafebad 81\r\n"
+        client = QueueClient.__new__(QueueClient)
+        client.stream = Stream(
+            [Response(2024, f"{len(data)} Queue list follows"),
+             Response(1000, "OK")],
+            data,
+        )
+
+        self.assertEqual(
+            client.List(),
+            [("007-deadbee", 42), ("999-cafebad", 81)],
+        )
+        self.assertEqual(client.stream.commands, ["QLIST"])
+        self.assertEqual(client.stream.lengths, [len(data)])
+
+    def test_move_uses_qmove(self):
+        client = QueueClient.__new__(QueueClient)
+        client.stream = Stream([Response(1000, "999-deadbee OK")], b"")
+
+        client.Move("007-deadbee", 999)
+
+        self.assertEqual(client.stream.commands, ["QMOVE 007-deadbee 999"])
+
+    def test_run_accepts_existing_queue_id(self):
+        client = QueueClient.__new__(QueueClient)
+        client.stream = Stream([Response(1000, "000-deadbee OK")], b"")
+
+        client.Run("000-deadbee")
+
+        self.assertEqual(client.stream.commands, ["QRUN 000-deadbee"])
+
     def test_retrieve_info_reads_declared_payload(self):
         client = QueueClient.__new__(QueueClient)
         client.stream = Stream(

src/libs/python/bongo/store/QueueClient.py

@@ -69,6 +69,35 @@ class QueueClient:
             raise bongo.BongoError(r.message)
         return r
 
+    def List(self):
+        self.stream.Write("QLIST")
+        r = self.stream.GetResponse()
+        if r.code != 2024:
+            raise bongo.BongoError(r.message)
+        try:
+            length = int(r.message.split(" ", 1)[0])
+        except (ValueError, IndexError):
+            raise bongo.BongoError("Malformed queue list response")
+        data = self.stream.Read(length)
+        r = self.stream.GetResponse()
+        if r.code != 1000:
+            raise bongo.BongoError(r.message)
+        result = []
+        for line in data.decode("ascii").splitlines():
+            queue_id, size = line.split()
+            result.append((queue_id, int(size)))
+        return result
+
+    def Move(self, queue_id, target):
+        target = int(target)
+        if target < 0 or target > 999:
+            raise ValueError("queue target must be between 0 and 999")
+        self.stream.Write("QMOVE %s %d" % (queue_id, target))
+        r = self.stream.GetResponse()
+        if r.code != 1000:
+            raise bongo.BongoError(r.message)
+        return r
+
     def RetrieveInfo(self, queue_id):
         self.stream.Write("QRETR %s INFO" % queue_id)
         r = self.stream.GetResponse()
@@ -107,8 +136,11 @@ class QueueClient:
         self.stream.Write("QUIT")
         self.stream.close()
 
-    def Run(self):
-        self.stream.Write("QRUN")
+    def Run(self, queue_id=None):
+        command = "QRUN"
+        if queue_id is not None:
+            command += " %s" % queue_id
+        self.stream.Write(command)
         r = self.stream.GetResponse()
         if r.code != 1000:
             raise bongo.BongoError(r.message)