diff options
| -rw-r--r-- | activitypub.c | 46 | ||||
| -rw-r--r-- | data.c | 72 | ||||
| -rw-r--r-- | snac.h | 1 |
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)); + } + } + } + } +} @@ -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 */ { @@ -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); |