commit c252432
default
·
2023-01-31 21:30:34 +0000 UTC
parent f0ef1d4
New function process_queue_item().
1 files changed,
+96,
-88
+96,
-88
1@@ -845,7 +845,7 @@ void notify(snac *snac, char *type, char *utype, char *actor, char *msg)
2
3 /** queues **/
4
5-int process_message(snac *snac, char *msg, char *req)
6+int process_input_message(snac *snac, char *msg, char *req)
7 /* processes an ActivityPub message from the input queue */
8 {
9 /* actor and type exist, were checked previously */
10@@ -1065,114 +1065,122 @@ int send_email(char *msg)
11 }
12
13
14-void process_queue(snac *snac)
15-/* processes the queue */
16+void process_queue_item(snac *snac, xs_dict *q_item)
17+/* processes an item from the queue */
18 {
19- xs *list;
20- char *p, *fn;
21+ char *type;
22 int queue_retry_max = xs_number_get(xs_dict_get(srv_config, "queue_retry_max"));
23
24- list = queue(snac);
25+ if ((type = xs_dict_get(q_item, "type")) == NULL)
26+ type = "output";
27
28- p = list;
29- while (xs_list_iter(&p, &fn)) {
30- xs *q_item = dequeue(snac, fn);
31- char *type;
32+ if (strcmp(type, "message") == 0) {
33+ xs_dict *msg = xs_dict_get(q_item, "message");
34+ xs *inboxes = inbox_list(snac, msg);
35+ xs_list *p;
36+ xs_str *inbox;
37
38- if (q_item == NULL) {
39- snac_log(snac, xs_fmt("process_queue q_item error"));
40- continue;
41+ p = inboxes;
42+ while (xs_list_iter(&p, &inbox)) {
43+ enqueue_output(snac, msg, inbox, 0);
44 }
45+ }
46+ else
47+ if (strcmp(type, "output") == 0) {
48+ int status;
49+ xs_str *inbox = xs_dict_get(q_item, "inbox");
50+ xs_dict *msg = xs_dict_get(q_item, "message");
51+ int retries = xs_number_get(xs_dict_get(q_item, "retries"));
52+ xs *payload = NULL;
53+ int p_size = 0;
54+
55+ if (xs_is_null(inbox) || xs_is_null(msg))
56+ return;
57
58- if ((type = xs_dict_get(q_item, "type")) == NULL)
59- type = "output";
60+ /* deliver */
61+ status = send_to_inbox(snac, inbox, msg, &payload, &p_size, retries == 0 ? 3 : 8);
62
63- if (strcmp(type, "message") == 0) {
64- char *msg = xs_dict_get(q_item, "message");
65- xs *inboxes = inbox_list(snac, msg);
66- char *p, *v;
67+ snac_log(snac, xs_fmt("process_queue sent to inbox %s %d", inbox, status));
68
69- p = inboxes;
70- while (xs_list_iter(&p, &v)) {
71- enqueue_output(snac, msg, v, 0);
72- }
73- }
74- else
75- if (strcmp(type, "output") == 0) {
76- int status;
77- char *inbox = xs_dict_get(q_item, "inbox");
78- char *msg = xs_dict_get(q_item, "message");
79- int retries = xs_number_get(xs_dict_get(q_item, "retries"));
80- xs *payload = NULL;
81- int p_size = 0;
82-
83- if (xs_is_null(inbox) || xs_is_null(msg))
84- continue;
85-
86- /* deliver */
87- status = send_to_inbox(snac, inbox, msg, &payload, &p_size, retries == 0 ? 3 : 8);
88-
89- snac_log(snac, xs_fmt("process_queue sent to inbox %s %d", inbox, status));
90-
91- if (!valid_status(status)) {
92- /* error sending; requeue? */
93- if (status == 404 || status == 410)
94- /* explicit error: discard */
95- snac_log(snac, xs_fmt("process_queue error %s %d", inbox, status));
96- else
97- if (retries > queue_retry_max)
98- snac_log(snac, xs_fmt("process_queue giving up %s %d", inbox, status));
99- else {
100- /* requeue */
101- enqueue_output(snac, msg, inbox, retries + 1);
102- snac_log(snac, xs_fmt("process_queue requeue %s #%d", inbox, retries + 1));
103- }
104+ if (!valid_status(status)) {
105+ /* error sending; requeue? */
106+ if (status == 404 || status == 410)
107+ /* explicit error: discard */
108+ snac_log(snac, xs_fmt("process_queue error %s %d", inbox, status));
109+ else
110+ if (retries > queue_retry_max)
111+ snac_log(snac, xs_fmt("process_queue giving up %s %d", inbox, status));
112+ else {
113+ /* requeue */
114+ enqueue_output(snac, msg, inbox, retries + 1);
115+ snac_log(snac, xs_fmt("process_queue requeue %s #%d", inbox, retries + 1));
116 }
117 }
118- else
119- if (strcmp(type, "input") == 0) {
120- /* process the message */
121- char *msg = xs_dict_get(q_item, "message");
122- char *req = xs_dict_get(q_item, "req");
123- int retries = xs_number_get(xs_dict_get(q_item, "retries"));
124-
125- if (xs_is_null(msg))
126- continue;
127-
128- if (!process_message(snac, msg, req)) {
129- if (retries > queue_retry_max)
130- snac_log(snac, xs_fmt("process_queue input giving up"));
131- else {
132- /* reenqueue */
133- enqueue_input(snac, msg, req, retries + 1);
134- snac_log(snac, xs_fmt("process_queue input requeue #%d", retries + 1));
135- }
136+ }
137+ else
138+ if (strcmp(type, "input") == 0) {
139+ /* process the message */
140+ xs_dict *msg = xs_dict_get(q_item, "message");
141+ xs_dict *req = xs_dict_get(q_item, "req");
142+ int retries = xs_number_get(xs_dict_get(q_item, "retries"));
143+
144+ if (xs_is_null(msg))
145+ return;
146+
147+ if (!process_input_message(snac, msg, req)) {
148+ if (retries > queue_retry_max)
149+ snac_log(snac, xs_fmt("process_queue input giving up"));
150+ else {
151+ /* reenqueue */
152+ enqueue_input(snac, msg, req, retries + 1);
153+ snac_log(snac, xs_fmt("process_queue input requeue #%d", retries + 1));
154 }
155 }
156- else
157- if (strcmp(type, "email") == 0) {
158- /* send this email */
159- char *msg = xs_dict_get(q_item, "message");
160- int retries = xs_number_get(xs_dict_get(q_item, "retries"));
161+ }
162+ else
163+ if (strcmp(type, "email") == 0) {
164+ /* send this email */
165+ xs_str *msg = xs_dict_get(q_item, "message");
166+ int retries = xs_number_get(xs_dict_get(q_item, "retries"));
167
168- if (!send_email(msg))
169- snac_debug(snac, 1, xs_fmt("email message sent"));
170+ if (!send_email(msg))
171+ snac_debug(snac, 1, xs_fmt("email message sent"));
172+ else {
173+ if (retries > queue_retry_max)
174+ snac_log(snac, xs_fmt("process_queue email giving up (errno: %d)", errno));
175 else {
176- if (retries > queue_retry_max)
177- snac_log(snac, xs_fmt("process_queue email giving up (errno: %d)", errno));
178- else {
179- /* requeue */
180- snac_log(snac, xs_fmt(
181- "process_queue email requeue #%d (errno: %d)", retries + 1, errno));
182-
183- enqueue_email(snac, msg, retries + 1);
184- }
185+ /* requeue */
186+ snac_log(snac, xs_fmt(
187+ "process_queue email requeue #%d (errno: %d)", retries + 1, errno));
188+
189+ enqueue_email(snac, msg, retries + 1);
190 }
191 }
192 }
193 }
194
195
196+void process_queue(snac *snac)
197+/* processes the queue */
198+{
199+ xs *list = queue(snac);
200+
201+ xs_list *p = list;
202+ xs_str *fn;
203+
204+ while (xs_list_iter(&p, &fn)) {
205+ xs *q_item = dequeue(snac, fn);
206+
207+ if (q_item == NULL) {
208+ snac_log(snac, xs_fmt("process_queue q_item error"));
209+ continue;
210+ }
211+
212+ process_queue_item(snac, q_item);
213+ }
214+}
215+
216+
217 /** HTTP handlers */
218
219 int activitypub_get_handler(d_char *req, char *q_path,