monkee.ch/snac2
1
0
Fork 0
mirror of https://codeberg.org/grunfink/snac2.git synced 2024-11-22 13:25:04 +00:00
Code Issues Projects Releases Packages Wiki Activity

Unify enqueueing code.

Browse source
This commit is contained in:
default 2022-10-11 08:16:58 +02:00
parent e493ff962f
commit e8fbc94089
1 changed files with 38 additions and 38 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

48
data.c
View file

@ -892,35 +892,46 @@ d_char *history_list(snac *snac)
} }
static int _enqueue_put(char *fn, char *msg)
/* writes safely to the queue */
{
int ret = 1;
xs *tfn = xs_fmt("%s.tmp", fn);
FILE *f;
if ((f = fopen(tfn, "w")) != NULL) {
xs *j = xs_json_dumps_pp(msg, 4);
fwrite(j, strlen(j), 1, f);
fclose(f);
rename(tfn, fn);
}
else
ret = 0;
return ret;
}
void enqueue_input(snac *snac, char *msg, char *req, int retries) void enqueue_input(snac *snac, char *msg, char *req, int retries)
/* enqueues an input message */ /* enqueues an input message */
{ {
int qrt = xs_number_get(xs_dict_get(srv_config, "queue_retry_minutes")); int qrt = xs_number_get(xs_dict_get(srv_config, "queue_retry_minutes"));
xs *ntid = tid(retries * 60 * qrt); xs *ntid = tid(retries * 60 * qrt);
xs *fn = xs_fmt("%s/queue/%s.json", snac->basedir, ntid); xs *fn = xs_fmt("%s/queue/%s.json", snac->basedir, ntid);
xs *tfn = xs_fmt("%s.tmp", fn);
FILE *f;
if ((f = fopen(tfn, "w")) != NULL) {
xs *qmsg = xs_dict_new(); xs *qmsg = xs_dict_new();
xs *rn = xs_number_new(retries); xs *rn = xs_number_new(retries);
xs *j;
qmsg = xs_dict_append(qmsg, "type", "input"); qmsg = xs_dict_append(qmsg, "type", "input");
qmsg = xs_dict_append(qmsg, "object", msg); qmsg = xs_dict_append(qmsg, "object", msg);
qmsg = xs_dict_append(qmsg, "req", req); qmsg = xs_dict_append(qmsg, "req", req);
qmsg = xs_dict_append(qmsg, "retries", rn); qmsg = xs_dict_append(qmsg, "retries", rn);
j = xs_json_dumps_pp(qmsg, 4); _enqueue_put(fn, qmsg);
fwrite(j, strlen(j), 1, f);
fclose(f);
rename(tfn, fn);
snac_debug(snac, 1, xs_fmt("enqueue_input %s", fn)); snac_debug(snac, 1, xs_fmt("enqueue_input %s", fn));
} }
}
void enqueue_output(snac *snac, char *msg, char *actor, int retries) void enqueue_output(snac *snac, char *msg, char *actor, int retries)
@ -934,29 +945,18 @@ void enqueue_output(snac *snac, char *msg, char *actor, int retries)
int qrt = xs_number_get(xs_dict_get(srv_config, "queue_retry_minutes")); int qrt = xs_number_get(xs_dict_get(srv_config, "queue_retry_minutes"));
xs *ntid = tid(retries * 60 * qrt); xs *ntid = tid(retries * 60 * qrt);
xs *fn = xs_fmt("%s/queue/%s.json", snac->basedir, ntid); xs *fn = xs_fmt("%s/queue/%s.json", snac->basedir, ntid);
xs *tfn = xs_fmt("%s.tmp", fn);
FILE *f;
if ((f = fopen(tfn, "w")) != NULL) {
xs *qmsg = xs_dict_new(); xs *qmsg = xs_dict_new();
xs *rn = xs_number_new(retries); xs *rn = xs_number_new(retries);
xs *j;
qmsg = xs_dict_append(qmsg, "type", "output"); qmsg = xs_dict_append(qmsg, "type", "output");
qmsg = xs_dict_append(qmsg, "actor", actor); qmsg = xs_dict_append(qmsg, "actor", actor);
qmsg = xs_dict_append(qmsg, "object", msg); qmsg = xs_dict_append(qmsg, "object", msg);
qmsg = xs_dict_append(qmsg, "retries", rn); qmsg = xs_dict_append(qmsg, "retries", rn);
j = xs_json_dumps_pp(qmsg, 4); _enqueue_put(fn, qmsg);
fwrite(j, strlen(j), 1, f);
fclose(f);
rename(tfn, fn);
snac_debug(snac, 1, xs_fmt("enqueue_output %s %s %d", actor, fn, retries)); snac_debug(snac, 1, xs_fmt("enqueue_output %s %s %d", actor, fn, retries));
} }
}
d_char *queue(snac *snac) d_char *queue(snac *snac)