commit 4cca157

Output messages are now processed by the pool of threads.

 1@@ -1195,6 +1195,44 @@ void process_queue_item(xs_dict *q_item)
 2     char *type = xs_dict_get(q_item, "type");
 3     int queue_retry_max = xs_number_get(xs_dict_get(srv_config, "queue_retry_max"));
 4 
 5+    if (strcmp(type, "output") == 0) {
 6+        int status;
 7+        xs_str *inbox  = xs_dict_get(q_item, "inbox");
 8+        xs_str *keyid  = xs_dict_get(q_item, "keyid");
 9+        xs_str *seckey = xs_dict_get(q_item, "seckey");
10+        xs_dict *msg   = xs_dict_get(q_item, "message");
11+        int retries    = xs_number_get(xs_dict_get(q_item, "retries"));
12+        xs *payload    = NULL;
13+        int p_size     = 0;
14+
15+        if (xs_is_null(inbox) || xs_is_null(msg) || xs_is_null(keyid) || xs_is_null(seckey)) {
16+            srv_log(xs_fmt("output message error: missing fields"));
17+            return;
18+        }
19+
20+        /* deliver */
21+        status = send_to_inbox_raw(keyid, seckey, inbox, msg, &payload, &p_size, retries == 0 ? 3 : 8);
22+
23+        srv_log(xs_fmt("output message: sent to inbox %s %d", inbox, status));
24+
25+        if (!valid_status(status)) {
26+            retries++;
27+
28+            /* error sending; requeue? */
29+            if (status == 404 || status == 410)
30+                /* explicit error: discard */
31+                srv_log(xs_fmt("output message: fatal error %s %d", inbox, status));
32+            else
33+            if (retries > queue_retry_max)
34+                srv_log(xs_fmt("output message: giving up %s %d", inbox, status));
35+            else {
36+                /* requeue */
37+                enqueue_output_raw(keyid, seckey, msg, inbox, retries);
38+                srv_log(xs_fmt("output message: requeue %s #%d", inbox, retries));
39+            }
40+        }
41+    }
42+    else
43     if (strcmp(type, "email") == 0) {
44         /* send this email */
45         xs_str *msg = xs_dict_get(q_item, "message");

 1@@ -1373,25 +1373,35 @@ void enqueue_input(snac *snac, xs_dict *msg, xs_dict *req, int retries)
 2 }
 3 
 4 
 5-void enqueue_output(snac *snac, xs_dict *msg, xs_str *inbox, int retries)
 6+void enqueue_output_raw(const char *keyid, const char *seckey,
 7+                        xs_dict *msg, xs_str *inbox, int retries)
 8 /* enqueues an output message to an inbox */
 9 {
10-    if (xs_startswith(inbox, snac->actor)) {
11-        snac_debug(snac, 1, xs_str_new("refusing enqueue to myself"));
12-        return;
13-    }
14-
15     xs *qmsg   = _new_qmsg("output", msg, retries);
16     char *ntid = xs_dict_get(qmsg, "ntid");
17-    xs *fn     = xs_fmt("%s/queue/%s.json", snac->basedir, ntid);
18+    xs *fn     = xs_fmt("%s/queue/%s.json", srv_basedir, ntid);
19 
20     qmsg = xs_dict_append(qmsg, "inbox",  inbox);
21-    qmsg = xs_dict_append(qmsg, "keyid",  snac->actor);
22-    qmsg = xs_dict_append(qmsg, "seckey", xs_dict_get(snac->key, "secret"));
23+    qmsg = xs_dict_append(qmsg, "keyid",  keyid);
24+    qmsg = xs_dict_append(qmsg, "seckey", seckey);
25 
26     qmsg = _enqueue_put(fn, qmsg);
27 
28-    snac_debug(snac, 1, xs_fmt("enqueue_output %s %s %d", inbox, fn, retries));
29+    srv_debug(1, xs_fmt("enqueue_output %s %s %d", inbox, fn, retries));
30+}
31+
32+
33+void enqueue_output(snac *snac, xs_dict *msg, xs_str *inbox, int retries)
34+/* enqueues an output message to an inbox */
35+{
36+    if (xs_startswith(inbox, snac->actor)) {
37+        snac_debug(snac, 1, xs_str_new("refusing enqueue to myself"));
38+        return;
39+    }
40+
41+    char *seckey = xs_dict_get(snac->key, "secret");
42+
43+    enqueue_output_raw(snac->actor, seckey, msg, inbox, retries);
44 }
45 
46 

1@@ -129,6 +129,8 @@ int history_del(snac *snac, char *id);
2 d_char *history_list(snac *snac);
3 
4 void enqueue_input(snac *snac, xs_dict *msg, xs_dict *req, int retries);
5+void enqueue_output_raw(const char *keyid, const char *seckey,
6+                        xs_dict *msg, xs_str *inbox, int retries);
7 void enqueue_output(snac *snac, xs_dict *msg, xs_str *inbox, int retries);
8 void enqueue_output_by_actor(snac *snac, xs_dict *msg, xs_str *actor, int retries);
9 void enqueue_email(xs_str *msg, int retries);