New function process_queue().

3 files changed, 82 insertions, 37 deletions

diff --git a/activitypub.c b/activitypub.c
index 5d54833..b654beb 100644
--- a/activitypub.c
+++ b/activitypub.c
@@ -80,9 +80,10 @@ int send_to_inbox(snac *snac, char *inbox, char *msg, d_char **payload, int *p_s
 {
     int status;
     d_char *response;
+    xs *j_msg = xs_json_dumps_pp(msg, 4);
 
     response = http_signed_request(snac, "POST", inbox,
-        NULL, msg, strlen(msg), &status, payload, p_size);
+        NULL, j_msg, strlen(j_msg), &status, payload, p_size);
 
     free(response);
 
@@ -108,5 +109,48 @@ int send_to_actor(snac *snac, char *actor, char *msg, d_char **payload, int *p_s
             status = 400;
     }
 
+    snac_log(snac, xs_fmt("send_to_actor %s %d", actor, status));
+
     return status;
 }
+
+
+void process_queue(snac *snac)
+/* processes the queue */
+{
+    xs *list;
+    char *p, *fn;
+    int queue_retry_max = xs_number_get(xs_dict_get(srv_config, "queue_retry_max"));
+
+    list = queue(snac);
+
+    p = list;
+    while (xs_list_iter(&p, &fn)) {
+        xs *q_item = dequeue(snac, fn);
+        char *type;
+
+        if ((type = xs_dict_get(q_item, "type")) == NULL)
+            type = "output";
+
+        if (strcmp(type, "output") == 0) {
+            int status;
+            char *actor = xs_dict_get(q_item, "actor");
+            char *msg   = xs_dict_get(q_item, "object");
+            int retries = xs_number_get(xs_dict_get(q_item, "retries"));
+
+            /* deliver */
+            status = send_to_actor(snac, actor, msg, NULL, 0);
+
+            if (!valid_status(status)) {
+                /* error sending; reenqueue? */
+                if (retries > queue_retry_max)
+                    snac_log(snac, xs_fmt("process_queue giving up %s %d", actor, status));
+                else {
+                    /* reenqueue */
+                    enqueue_output(snac, actor, msg, retries + 1);
+                    snac_log(snac, xs_fmt("process_queue requeue %s %d", actor, retries + 1));
+                }
+            }
+        }
+    }
+}
diff --git a/data.c b/data.c
index c9ba389..3f5d5e0 100644
--- a/data.c
+++ b/data.c
@@ -631,42 +631,6 @@ int is_muted(snac *snac, char *actor)
 }
 
 
-void enqueue_output(snac *snac, char *actor, char *msg, int retries)
-/* enqueues an output message for an actor */
-{
-    if (strcmp(actor, snac->actor) == 0) {
-        snac_debug(snac, 1, xs_str_new("enqueue refused to myself"));
-        return;
-    }
-
-    int qrt  = xs_number_get(xs_dict_get(srv_config, "query_retry_minutes"));
-    xs *ntid = tid(retries * 60 * qrt);
-    xs *fn   = xs_fmt("%s/queue/%s.json", snac->basedir, ntid);
-    xs *tfn  = xs_str_cat(fn, ".tmp");
-    FILE *f;
-
-    if ((f = fopen(tfn, "w")) != NULL) {
-        xs *qmsg = xs_dict_new();
-        xs *rn   = xs_number_new(retries);
-        xs *j;
-
-        qmsg = xs_dict_append(qmsg, "type",    "output");
-        qmsg = xs_dict_append(qmsg, "actor",   actor);
-        qmsg = xs_dict_append(qmsg, "object",  msg);
-        qmsg = xs_dict_append(qmsg, "retries", rn);
-
-        j = xs_json_dumps_pp(qmsg, 4);
-
-        fwrite(j, strlen(j), 1, f);
-        fclose(f);
-
-        rename(tfn, fn);
-
-        snac_debug(snac, 2, xs_fmt("enqueue %s %s %d", actor, fn, retries));
-    }
-}
-
-
 d_char *_actor_fn(snac *snac, char *actor)
 /* returns the file name for an actor */
 {
@@ -745,6 +709,42 @@ int actor_get(snac *snac, char *actor, d_char **data)
 }
 
 
+void enqueue_output(snac *snac, char *actor, char *msg, int retries)
+/* enqueues an output message for an actor */
+{
+    if (strcmp(actor, snac->actor) == 0) {
+        snac_debug(snac, 1, xs_str_new("enqueue refused to myself"));
+        return;
+    }
+
+    int qrt  = xs_number_get(xs_dict_get(srv_config, "query_retry_minutes"));
+    xs *ntid = tid(retries * 60 * qrt);
+    xs *fn   = xs_fmt("%s/queue/%s.json", snac->basedir, ntid);
+    xs *tfn  = xs_str_cat(fn, ".tmp");
+    FILE *f;
+
+    if ((f = fopen(tfn, "w")) != NULL) {
+        xs *qmsg = xs_dict_new();
+        xs *rn   = xs_number_new(retries);
+        xs *j;
+
+        qmsg = xs_dict_append(qmsg, "type",    "output");
+        qmsg = xs_dict_append(qmsg, "actor",   actor);
+        qmsg = xs_dict_append(qmsg, "object",  msg);
+        qmsg = xs_dict_append(qmsg, "retries", rn);
+
+        j = xs_json_dumps_pp(qmsg, 4);
+
+        fwrite(j, strlen(j), 1, f);
+        fclose(f);
+
+        rename(tfn, fn);
+
+        snac_debug(snac, 2, xs_fmt("enqueue %s %s %d", actor, fn, retries));
+    }
+}
+
+
 d_char *queue(snac *snac)
 /* returns a list with filenames that can be dequeued */
 {
diff --git a/snac.h b/snac.h
index a10169e..166806b 100644
--- a/snac.h
+++ b/snac.h
@@ -84,3 +84,4 @@ int activitypub_request(snac *snac, char *url, d_char **data);
 int actor_request(snac *snac, char *actor, d_char **data);
 int send_to_inbox(snac *snac, char *inbox, char *msg, d_char **payload, int *p_size);
 int send_to_actor(snac *snac, char *actor, char *msg, d_char **payload, int *p_size);
+void process_queue(snac *snac);