mirrors/snac2
3
0
Fork 0
mirror of https://codeberg.org/grunfink/snac2.git synced 2024-09-19 18:00:23 +03:00
Code Issues Packages Projects Releases Wiki Activity

New function process_queue().

Browse Source
This commit is contained in:
default 2022-09-23 19:37:01 +02:00
parent cf59d68491
commit bbf5471039
3 changed files with 82 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
activitypub.c
View File

@ -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));
}
}
}
}
}

72
data.c
View File

@ -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 */
{

1
snac.h
View File

@ -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);