Output messages are now processed by the pool of threads.

1 files changed, 38 insertions, 0 deletions

diff --git a/activitypub.c b/activitypub.c
index af0a9a1..b54845d 100644
--- a/activitypub.c
+++ b/activitypub.c
@@ -1195,6 +1195,44 @@ void process_queue_item(xs_dict *q_item)
     char *type = xs_dict_get(q_item, "type");
     int queue_retry_max = xs_number_get(xs_dict_get(srv_config, "queue_retry_max"));
 
+    if (strcmp(type, "output") == 0) {
+        int status;
+        xs_str *inbox  = xs_dict_get(q_item, "inbox");
+        xs_str *keyid  = xs_dict_get(q_item, "keyid");
+        xs_str *seckey = xs_dict_get(q_item, "seckey");
+        xs_dict *msg   = xs_dict_get(q_item, "message");
+        int retries    = xs_number_get(xs_dict_get(q_item, "retries"));
+        xs *payload    = NULL;
+        int p_size     = 0;
+
+        if (xs_is_null(inbox) || xs_is_null(msg) || xs_is_null(keyid) || xs_is_null(seckey)) {
+            srv_log(xs_fmt("output message error: missing fields"));
+            return;
+        }
+
+        /* deliver */
+        status = send_to_inbox_raw(keyid, seckey, inbox, msg, &payload, &p_size, retries == 0 ? 3 : 8);
+
+        srv_log(xs_fmt("output message: sent to inbox %s %d", inbox, status));
+
+        if (!valid_status(status)) {
+            retries++;
+
+            /* error sending; requeue? */
+            if (status == 404 || status == 410)
+                /* explicit error: discard */
+                srv_log(xs_fmt("output message: fatal error %s %d", inbox, status));
+            else
+            if (retries > queue_retry_max)
+                srv_log(xs_fmt("output message: giving up %s %d", inbox, status));
+            else {
+                /* requeue */
+                enqueue_output_raw(keyid, seckey, msg, inbox, retries);
+                srv_log(xs_fmt("output message: requeue %s #%d", inbox, retries));
+            }
+        }
+    }
+    else
     if (strcmp(type, "email") == 0) {
         /* send this email */
         xs_str *msg = xs_dict_get(q_item, "message");