X-Git-Url: http://git.archive.openwrt.org/?p=project%2Fubus.git;a=blobdiff_plain;f=libubus.c;h=bfbaeb6565905f2614e202fc8c25d265fd1ca52c;hp=fee3ee8f6c9a7df1616ddb89f1338ce84e91ae9e;hb=f09d18878b09d7d6c1ca7ef7da51131f46ddb2c6;hpb=22246585804a2f6546a329a9e626598a82feb554 diff --git a/libubus.c b/libubus.c index fee3ee8..bfbaeb6 100644 --- a/libubus.c +++ b/libubus.c @@ -1,5 +1,5 @@ /* - * Copyright (C) 2011-2012 Felix Fietkau + * Copyright (C) 2011-2014 Felix Fietkau * * This program is free software; you can redistribute it and/or modify * it under the terms of the GNU Lesser General Public License version 2.1 @@ -40,7 +40,7 @@ struct blob_buf b __hidden = {}; struct ubus_pending_msg { struct list_head list; - struct ubus_msghdr hdr; + struct ubus_msghdr_buf hdr; }; static int ubus_cmp_id(const void *k1, const void *k2, void *ptr) @@ -71,51 +71,57 @@ out: } static void -ubus_queue_msg(struct ubus_context *ctx, struct ubus_msghdr *hdr) +ubus_queue_msg(struct ubus_context *ctx, struct ubus_msghdr_buf *buf) { struct ubus_pending_msg *pending; + void *data; - pending = calloc(1, sizeof(*pending) + blob_raw_len(hdr->data)); - if (!pending) - return; + pending = calloc_a(sizeof(*pending), &data, blob_raw_len(buf->data)); - memcpy(&pending->hdr, hdr, sizeof(*hdr) + blob_raw_len(hdr->data)); - list_add(&pending->list, &ctx->pending); + pending->hdr.data = data; + memcpy(&pending->hdr.hdr, &buf->hdr, sizeof(buf->hdr)); + memcpy(data, buf->data, blob_raw_len(buf->data)); + list_add_tail(&pending->list, &ctx->pending); + if (ctx->sock.registered) + uloop_timeout_set(&ctx->pending_timer, 1); } void __hidden -ubus_process_msg(struct ubus_context *ctx, struct ubus_msghdr *hdr) +ubus_process_msg(struct ubus_context *ctx, struct ubus_msghdr_buf *buf, int fd) { - - switch(hdr->type) { + switch(buf->hdr.type) { case UBUS_MSG_STATUS: case UBUS_MSG_DATA: - ubus_process_req_msg(ctx, hdr); + ubus_process_req_msg(ctx, buf, fd); break; case UBUS_MSG_INVOKE: case UBUS_MSG_UNSUBSCRIBE: case UBUS_MSG_NOTIFY: - if (ctx->stack_depth > 2) { - ubus_queue_msg(ctx, hdr); + if (ctx->stack_depth) { + ubus_queue_msg(ctx, buf); break; } - ubus_process_obj_msg(ctx, hdr); + ubus_process_obj_msg(ctx, buf); + break; + case UBUS_MSG_MONITOR: + if (ctx->monitor_cb) + ctx->monitor_cb(ctx, buf->hdr.seq, buf->data); break; } } -void __hidden ubus_process_pending_msg(struct ubus_context *ctx) +static void ubus_process_pending_msg(struct uloop_timeout *timeout) { - struct ubus_pending_msg *pending, *tmp; + struct ubus_context *ctx = container_of(timeout, struct ubus_context, pending_timer); + struct ubus_pending_msg *pending; - list_for_each_entry_safe(pending, tmp, &ctx->pending, list) { + while (!ctx->stack_depth && !list_empty(&ctx->pending)) { + pending = list_first_entry(&ctx->pending, struct ubus_pending_msg, list); list_del(&pending->list); - ubus_process_msg(ctx, &pending->hdr); + ubus_process_msg(ctx, &pending->hdr, -1); free(pending); - if (ctx->stack_depth > 2) - break; } } @@ -236,8 +242,11 @@ int ubus_register_event_handler(struct ubus_context *ctx, if (pattern) blobmsg_add_string(&b2, "pattern", pattern); - return ubus_invoke(ctx, UBUS_SYSTEM_OBJECT_EVENT, "register", b2.head, + ret = ubus_invoke(ctx, UBUS_SYSTEM_OBJECT_EVENT, "register", b2.head, NULL, NULL, 0); + blob_buf_free(&b2); + + return ret; } int ubus_send_event(struct ubus_context *ctx, const char *id, @@ -266,22 +275,79 @@ static void ubus_default_connection_lost(struct ubus_context *ctx) uloop_end(); } -struct ubus_context *ubus_connect(const char *path) +int ubus_connect_ctx(struct ubus_context *ctx, const char *path) { - struct ubus_context *ctx; - - ctx = calloc(1, sizeof(*ctx)); - if (!ctx) - return NULL; + memset(ctx, 0, sizeof(*ctx)); ctx->sock.fd = -1; ctx->sock.cb = ubus_handle_data; ctx->connection_lost = ubus_default_connection_lost; + ctx->pending_timer.cb = ubus_process_pending_msg; + + ctx->msgbuf.data = calloc(UBUS_MSG_CHUNK_SIZE, sizeof(char)); + if (!ctx->msgbuf.data) + return -1; + ctx->msgbuf_data_len = UBUS_MSG_CHUNK_SIZE; INIT_LIST_HEAD(&ctx->requests); INIT_LIST_HEAD(&ctx->pending); avl_init(&ctx->objects, ubus_cmp_id, false, NULL); if (ubus_reconnect(ctx, path)) { + free(ctx->msgbuf.data); + return -1; + } + + return 0; +} + +static void ubus_auto_reconnect_cb(struct uloop_timeout *timeout) +{ + struct ubus_auto_conn *conn = container_of(timeout, struct ubus_auto_conn, timer); + + if (!ubus_reconnect(&conn->ctx, conn->path)) + ubus_add_uloop(&conn->ctx); + else + uloop_timeout_set(timeout, 1000); +} + +static void ubus_auto_disconnect_cb(struct ubus_context *ctx) +{ + struct ubus_auto_conn *conn = container_of(ctx, struct ubus_auto_conn, ctx); + + conn->timer.cb = ubus_auto_reconnect_cb; + uloop_timeout_set(&conn->timer, 1000); +} + +static void ubus_auto_connect_cb(struct uloop_timeout *timeout) +{ + struct ubus_auto_conn *conn = container_of(timeout, struct ubus_auto_conn, timer); + + if (ubus_connect_ctx(&conn->ctx, conn->path)) { + uloop_timeout_set(timeout, 1000); + fprintf(stderr, "failed to connect to ubus\n"); + return; + } + conn->ctx.connection_lost = ubus_auto_disconnect_cb; + if (conn->cb) + conn->cb(&conn->ctx); + ubus_add_uloop(&conn->ctx); +} + +void ubus_auto_connect(struct ubus_auto_conn *conn) +{ + conn->timer.cb = ubus_auto_connect_cb; + ubus_auto_connect_cb(&conn->timer); +} + +struct ubus_context *ubus_connect(const char *path) +{ + struct ubus_context *ctx; + + ctx = calloc(1, sizeof(*ctx)); + if (!ctx) + return NULL; + + if (ubus_connect_ctx(ctx, path)) { free(ctx); ctx = NULL; } @@ -289,8 +355,15 @@ struct ubus_context *ubus_connect(const char *path) return ctx; } -void ubus_free(struct ubus_context *ctx) +void ubus_shutdown(struct ubus_context *ctx) { + blob_buf_free(&b); close(ctx->sock.fd); + free(ctx->msgbuf.data); +} + +void ubus_free(struct ubus_context *ctx) +{ + ubus_shutdown(ctx); free(ctx); }