add auto (re)connect logic to libubus
[project/ubus.git] / libubus-req.c
1 /*
2  * Copyright (C) 2011-2012 Felix Fietkau <nbd@openwrt.org>
3  *
4  * This program is free software; you can redistribute it and/or modify
5  * it under the terms of the GNU Lesser General Public License version 2.1
6  * as published by the Free Software Foundation
7  *
8  * This program is distributed in the hope that it will be useful,
9  * but WITHOUT ANY WARRANTY; without even the implied warranty of
10  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
11  * GNU General Public License for more details.
12  */
13
14 #include "libubus.h"
15 #include "libubus-internal.h"
16
17 struct ubus_pending_data {
18         struct list_head list;
19         int type;
20         struct blob_attr data[];
21 };
22
23 static void req_data_cb(struct ubus_request *req, int type, struct blob_attr *data)
24 {
25         struct blob_attr **attr;
26
27         if (req->raw_data_cb)
28                 req->raw_data_cb(req, type, data);
29
30         if (!req->data_cb)
31                 return;
32
33         attr = ubus_parse_msg(data);
34         req->data_cb(req, type, attr[UBUS_ATTR_DATA]);
35 }
36
37 static void __ubus_process_req_data(struct ubus_request *req)
38 {
39         struct ubus_pending_data *data;
40
41         while (!list_empty(&req->pending)) {
42                 data = list_first_entry(&req->pending,
43                         struct ubus_pending_data, list);
44                 list_del(&data->list);
45                 if (!req->cancelled)
46                         req_data_cb(req, data->type, data->data);
47                 free(data);
48         }
49 }
50
51 int __hidden ubus_start_request(struct ubus_context *ctx, struct ubus_request *req,
52                                 struct blob_attr *msg, int cmd, uint32_t peer)
53 {
54         memset(req, 0, sizeof(*req));
55
56         if (msg && blob_pad_len(msg) > UBUS_MAX_MSGLEN)
57                 return -1;
58
59         INIT_LIST_HEAD(&req->list);
60         INIT_LIST_HEAD(&req->pending);
61         req->ctx = ctx;
62         req->peer = peer;
63         req->seq = ++ctx->request_seq;
64         return ubus_send_msg(ctx, req->seq, msg, cmd, peer);
65 }
66
67 void ubus_abort_request(struct ubus_context *ctx, struct ubus_request *req)
68 {
69         if (list_empty(&req->list))
70                 return;
71
72         req->cancelled = true;
73         __ubus_process_req_data(req);
74         list_del_init(&req->list);
75 }
76
77 void ubus_complete_request_async(struct ubus_context *ctx, struct ubus_request *req)
78 {
79         if (!list_empty(&req->list))
80                 return;
81
82         list_add(&req->list, &ctx->requests);
83 }
84
85 static void
86 ubus_req_complete_cb(struct ubus_request *req)
87 {
88         ubus_complete_handler_t cb = req->complete_cb;
89
90         if (!cb)
91                 return;
92
93         req->complete_cb = NULL;
94         cb(req, req->status_code);
95 }
96
97 static void
98 ubus_set_req_status(struct ubus_request *req, int ret)
99 {
100         if (!list_empty(&req->list))
101                 list_del_init(&req->list);
102
103         req->status_msg = true;
104         req->status_code = ret;
105         if (!req->blocked)
106                 ubus_req_complete_cb(req);
107 }
108
109 static void ubus_sync_req_cb(struct ubus_request *req, int ret)
110 {
111         req->status_msg = true;
112         req->status_code = ret;
113         uloop_end();
114 }
115
116 struct ubus_sync_req_cb {
117         struct uloop_timeout timeout;
118         struct ubus_request *req;
119 };
120
121 static void ubus_sync_req_timeout_cb(struct uloop_timeout *timeout)
122 {
123         struct ubus_sync_req_cb *cb;
124
125         cb = container_of(timeout, struct ubus_sync_req_cb, timeout);
126         ubus_set_req_status(cb->req, UBUS_STATUS_TIMEOUT);
127 }
128
129 int ubus_complete_request(struct ubus_context *ctx, struct ubus_request *req,
130                           int timeout)
131 {
132         struct ubus_sync_req_cb cb;
133         ubus_complete_handler_t complete_cb = req->complete_cb;
134         bool registered = ctx->sock.registered;
135         int status = UBUS_STATUS_NO_DATA;
136
137         if (!registered) {
138                 uloop_init();
139                 ubus_add_uloop(ctx);
140         }
141
142         if (timeout) {
143                 memset(&cb, 0, sizeof(cb));
144                 cb.req = req;
145                 cb.timeout.cb = ubus_sync_req_timeout_cb;
146                 uloop_timeout_set(&cb.timeout, timeout);
147         }
148
149         ubus_complete_request_async(ctx, req);
150         req->complete_cb = ubus_sync_req_cb;
151
152         ctx->stack_depth++;
153         while (!req->status_msg) {
154                 bool cancelled = uloop_cancelled;
155                 uloop_cancelled = false;
156                 uloop_run();
157                 uloop_cancelled = cancelled;
158         }
159         ctx->stack_depth--;
160         if (ctx->stack_depth)
161                 uloop_cancelled = true;
162
163         if (timeout)
164                 uloop_timeout_cancel(&cb.timeout);
165
166         if (req->status_msg)
167                 status = req->status_code;
168
169         req->complete_cb = complete_cb;
170         if (req->complete_cb)
171                 req->complete_cb(req, status);
172
173         if (!registered)
174                 uloop_fd_delete(&ctx->sock);
175
176         if (!ctx->stack_depth)
177                 ubus_process_pending_msg(ctx);
178
179         return status;
180 }
181
182 void ubus_complete_deferred_request(struct ubus_context *ctx, struct ubus_request_data *req, int ret)
183 {
184         blob_buf_init(&b, 0);
185         blob_put_int32(&b, UBUS_ATTR_STATUS, ret);
186         blob_put_int32(&b, UBUS_ATTR_OBJID, req->object);
187         ubus_send_msg(ctx, req->seq, b.head, UBUS_MSG_STATUS, req->peer);
188 }
189
190 int ubus_send_reply(struct ubus_context *ctx, struct ubus_request_data *req,
191                     struct blob_attr *msg)
192 {
193         int ret;
194
195         blob_buf_init(&b, 0);
196         blob_put_int32(&b, UBUS_ATTR_OBJID, req->object);
197         blob_put(&b, UBUS_ATTR_DATA, blob_data(msg), blob_len(msg));
198         ret = ubus_send_msg(ctx, req->seq, b.head, UBUS_MSG_DATA, req->peer);
199         if (ret < 0)
200                 return UBUS_STATUS_NO_DATA;
201
202         return 0;
203 }
204
205 int ubus_invoke_async(struct ubus_context *ctx, uint32_t obj, const char *method,
206                        struct blob_attr *msg, struct ubus_request *req)
207 {
208         blob_buf_init(&b, 0);
209         blob_put_int32(&b, UBUS_ATTR_OBJID, obj);
210         blob_put_string(&b, UBUS_ATTR_METHOD, method);
211         if (msg)
212                 blob_put(&b, UBUS_ATTR_DATA, blob_data(msg), blob_len(msg));
213
214         if (ubus_start_request(ctx, req, b.head, UBUS_MSG_INVOKE, obj) < 0)
215                 return UBUS_STATUS_INVALID_ARGUMENT;
216
217         return 0;
218 }
219
220 int ubus_invoke(struct ubus_context *ctx, uint32_t obj, const char *method,
221                 struct blob_attr *msg, ubus_data_handler_t cb, void *priv,
222                 int timeout)
223 {
224         struct ubus_request req;
225
226         ubus_invoke_async(ctx, obj, method, msg, &req);
227         req.data_cb = cb;
228         req.priv = priv;
229         return ubus_complete_request(ctx, &req, timeout);
230 }
231
232 static void
233 ubus_notify_complete_cb(struct ubus_request *req, int ret)
234 {
235         struct ubus_notify_request *nreq;
236
237         nreq = container_of(req, struct ubus_notify_request, req);
238         if (!nreq->complete_cb)
239                 return;
240
241         nreq->complete_cb(nreq, 0, 0);
242 }
243
244 static int
245 __ubus_notify_async(struct ubus_context *ctx, struct ubus_object *obj,
246                     const char *type, struct blob_attr *msg,
247                     struct ubus_notify_request *req, bool reply)
248 {
249         memset(req, 0, sizeof(*req));
250
251         blob_buf_init(&b, 0);
252         blob_put_int32(&b, UBUS_ATTR_OBJID, obj->id);
253         blob_put_string(&b, UBUS_ATTR_METHOD, type);
254
255         if (!reply)
256                 blob_put_int8(&b, UBUS_ATTR_NO_REPLY, true);
257
258         if (msg)
259                 blob_put(&b, UBUS_ATTR_DATA, blob_data(msg), blob_len(msg));
260
261         if (ubus_start_request(ctx, &req->req, b.head, UBUS_MSG_NOTIFY, obj->id) < 0)
262                 return UBUS_STATUS_INVALID_ARGUMENT;
263
264         /* wait for status message from ubusd first */
265         req->req.notify = true;
266         req->pending = 1;
267         req->id[0] = obj->id;
268         req->req.complete_cb = ubus_notify_complete_cb;
269
270         return 0;
271 }
272
273 int ubus_notify_async(struct ubus_context *ctx, struct ubus_object *obj,
274                       const char *type, struct blob_attr *msg,
275                       struct ubus_notify_request *req)
276 {
277         return __ubus_notify_async(ctx, obj, type, msg, req, true);
278 }
279
280 int ubus_notify(struct ubus_context *ctx, struct ubus_object *obj,
281                 const char *type, struct blob_attr *msg, int timeout)
282 {
283         struct ubus_notify_request req;
284         int ret;
285
286         ret = __ubus_notify_async(ctx, obj, type, msg, &req, timeout >= 0);
287         if (ret < 0)
288                 return ret;
289
290         if (timeout < 0) {
291                 ubus_abort_request(ctx, &req.req);
292                 return 0;
293         }
294
295         return ubus_complete_request(ctx, &req.req, timeout);
296 }
297
298 static bool ubus_get_status(struct ubus_msghdr *hdr, int *ret)
299 {
300         struct blob_attr **attrbuf = ubus_parse_msg(ubus_msghdr_data(hdr));
301
302         if (!attrbuf[UBUS_ATTR_STATUS])
303                 return false;
304
305         *ret = blob_get_u32(attrbuf[UBUS_ATTR_STATUS]);
306         return true;
307 }
308
309 static int
310 ubus_process_req_status(struct ubus_request *req, struct ubus_msghdr *hdr)
311 {
312         int ret = UBUS_STATUS_INVALID_ARGUMENT;
313
314         ubus_get_status(hdr, &ret);
315         req->peer = hdr->peer;
316         ubus_set_req_status(req, ret);
317
318         return ret;
319 }
320
321 static void
322 ubus_process_req_data(struct ubus_request *req, struct ubus_msghdr *hdr)
323 {
324         struct blob_attr *msg_data = ubus_msghdr_data(hdr);
325         struct ubus_pending_data *data;
326         int len;
327
328         if (!req->blocked) {
329                 req->blocked = true;
330                 req_data_cb(req, hdr->type, msg_data);
331                 __ubus_process_req_data(req);
332                 req->blocked = false;
333
334                 if (req->status_msg)
335                         ubus_req_complete_cb(req);
336
337                 return;
338         }
339
340         len = blob_raw_len(msg_data);
341         data = calloc(1, sizeof(*data) + len);
342         if (!data)
343                 return;
344
345         data->type = hdr->type;
346         memcpy(data->data, msg_data, len);
347         list_add(&data->list, &req->pending);
348 }
349
350 static int
351 ubus_find_notify_id(struct ubus_notify_request *n, uint32_t objid)
352 {
353         uint32_t pending = n->pending;
354         int i;
355
356         for (i = 0; pending; i++, pending >>= 1) {
357                 if (!(pending & 1))
358                         continue;
359
360                 if (n->id[i] == objid)
361                         return i;
362         }
363
364         return -1;
365 }
366
367 static struct ubus_request *
368 ubus_find_request(struct ubus_context *ctx, uint32_t seq, uint32_t peer, int *id)
369 {
370         struct ubus_request *req;
371
372         list_for_each_entry(req, &ctx->requests, list) {
373                 struct ubus_notify_request *nreq;
374                 nreq = container_of(req, struct ubus_notify_request, req);
375
376                 if (seq != req->seq)
377                         continue;
378
379                 if (req->notify) {
380                         if (!nreq->pending)
381                                 continue;
382
383                         *id = ubus_find_notify_id(nreq, peer);
384                         if (*id < 0)
385                                 continue;
386                 } else if (peer != req->peer)
387                         continue;
388
389                 return req;
390         }
391         return NULL;
392 }
393
394 static void ubus_process_notify_status(struct ubus_request *req, int id, struct ubus_msghdr *hdr)
395 {
396         struct ubus_notify_request *nreq;
397         struct blob_attr **tb;
398         struct blob_attr *cur;
399         int rem, idx = 1;
400         int ret = 0;
401
402         nreq = container_of(req, struct ubus_notify_request, req);
403         nreq->pending &= ~(1 << id);
404
405         if (!id) {
406                 /* first id: ubusd's status message with a list of ids */
407                 tb = ubus_parse_msg(ubus_msghdr_data(hdr));
408                 if (tb[UBUS_ATTR_SUBSCRIBERS]) {
409                         blob_for_each_attr(cur, tb[UBUS_ATTR_SUBSCRIBERS], rem) {
410                                 if (!blob_check_type(blob_data(cur), blob_len(cur), BLOB_ATTR_INT32))
411                                         continue;
412
413                                 nreq->pending |= (1 << idx);
414                                 nreq->id[idx] = blob_get_int32(cur);
415                                 idx++;
416
417                                 if (idx == UBUS_MAX_NOTIFY_PEERS + 1)
418                                         break;
419                         }
420                 }
421         } else {
422                 ubus_get_status(hdr, &ret);
423                 if (nreq->status_cb)
424                         nreq->status_cb(nreq, id, ret);
425         }
426
427         if (!nreq->pending)
428                 ubus_set_req_status(req, 0);
429 }
430
431 void __hidden ubus_process_req_msg(struct ubus_context *ctx, struct ubus_msghdr *hdr)
432 {
433         struct ubus_request *req;
434         int id = -1;
435
436         switch(hdr->type) {
437         case UBUS_MSG_STATUS:
438                 req = ubus_find_request(ctx, hdr->seq, hdr->peer, &id);
439                 if (!req)
440                         break;
441
442                 if (id >= 0)
443                         ubus_process_notify_status(req, id, hdr);
444                 else
445                         ubus_process_req_status(req, hdr);
446                 break;
447
448         case UBUS_MSG_DATA:
449                 req = ubus_find_request(ctx, hdr->seq, hdr->peer, &id);
450                 if (req && (req->data_cb || req->raw_data_cb))
451                         ubus_process_req_data(req, hdr);
452                 break;
453         }
454 }