ubus: add notification for subscribers present/gone

Signed-off-by: Felix Fietkau <nbd@openwrt.org>

examples/client.c

@@ -18,7 +18,14 @@
 static struct ubus_context *ctx;
 static struct blob_buf b;
 
-static struct ubus_object test_client_object = {};
+static void test_client_subscribe_cb(struct ubus_context *ctx, struct ubus_object *obj)
+{
+	fprintf(stderr, "Subscribers active: %d\n", obj->has_subscribers);
+}
+
+static struct ubus_object test_client_object = {
+	.subscribe_cb = test_client_subscribe_cb,
+};
 
 static void client_main(void)
 {

libubus-internal.h

@@ -27,5 +27,6 @@ void ubus_process_invoke(struct ubus_context *ctx, struct ubus_msghdr *hdr);
 int __hidden ubus_start_request(struct ubus_context *ctx, struct ubus_request *req,
 				struct blob_attr *msg, int cmd, uint32_t peer);
 void ubus_process_unsubscribe(struct ubus_context *ctx, struct ubus_msghdr *hdr);
+void ubus_process_notify(struct ubus_context *ctx, struct ubus_msghdr *hdr);
 
 #endif

libubus-sub.c

@@ -91,4 +91,22 @@ void __hidden ubus_process_unsubscribe(struct ubus_context *ctx, struct ubus_msg
 	s->remove_cb(ctx, s, blob_get_u32(attrbuf[UBUS_ATTR_TARGET]));
 }
 
+void __hidden ubus_process_notify(struct ubus_context *ctx, struct ubus_msghdr *hdr)
+{
+	struct blob_attr **attrbuf;
+	struct ubus_object *obj;
+	uint32_t objid;
 
+	attrbuf = ubus_parse_msg(hdr->data);
+	if (!attrbuf[UBUS_ATTR_OBJID] || !attrbuf[UBUS_ATTR_ACTIVE])
+		return;
+
+	objid = blob_get_u32(attrbuf[UBUS_ATTR_OBJID]);
+	obj = avl_find_element(&ctx->objects, &objid, obj, avl);
+	if (!obj)
+		return;
+
+	obj->has_subscribers = blob_get_u8(attrbuf[UBUS_ATTR_ACTIVE]);
+	if (obj->subscribe_cb)
+		obj->subscribe_cb(ctx, obj);
+}

libubus.c

@@ -232,6 +232,10 @@ void __hidden ubus_process_msg(struct ubus_context *ctx, struct ubus_msghdr *hdr
 	case UBUS_MSG_UNSUBSCRIBE:
 		ubus_process_unsubscribe(ctx, hdr);
 		break;
+
+	case UBUS_MSG_NOTIFY:
+		ubus_process_notify(ctx, hdr);
+		break;
 	}
 }
 

libubus.h

@@ -37,6 +37,7 @@ typedef void (*ubus_lookup_handler_t)(struct ubus_context *ctx,
 typedef int (*ubus_handler_t)(struct ubus_context *ctx, struct ubus_object *obj,
 			      struct ubus_request_data *req,
 			      const char *method, struct blob_attr *msg);
+typedef void (*ubus_state_handler_t)(struct ubus_context *ctx, struct ubus_object *obj);
 typedef void (*ubus_remove_handler_t)(struct ubus_context *ctx,
 				      struct ubus_subscriber *obj, uint32_t id);
 typedef void (*ubus_event_handler_t)(struct ubus_context *ctx, struct ubus_event_handler *ev,
@@ -86,6 +87,9 @@ struct ubus_object {
 	const char *path;
 	struct ubus_object_type *type;
 
+	ubus_state_handler_t subscribe_cb;
+	bool has_subscribers;
+
 	const struct ubus_method *methods;
 	int n_methods;
 };
@@ -134,6 +138,7 @@ struct ubus_request_data {
 	uint32_t peer;
 	uint32_t seq;
 	bool deferred;
+	bool notify;
 };
 
 struct ubus_request {

ubusd_obj.c

@@ -167,6 +167,7 @@ free:
 void ubus_subscribe(struct ubus_object *obj, struct ubus_object *target, const char *method)
 {
 	struct ubus_subscription *s;
+	bool first = list_empty(&target->subscribers);
 
 	s = calloc(1, sizeof(*s) + strlen(method) + 1);
 	if (!s)
@@ -177,13 +178,21 @@ void ubus_subscribe(struct ubus_object *obj, struct ubus_object *target, const c
 	list_add(&s->list, &target->subscribers);
 	list_add(&s->target_list, &obj->target_list);
 	strcpy(s->method, method);
+
+	if (first)
+		ubus_notify_subscription(target);
 }
 
 void ubus_unsubscribe(struct ubus_subscription *s)
 {
+	struct ubus_object *obj = s->target;
+
 	list_del(&s->list);
 	list_del(&s->target_list);
 	free(s);
+
+	if (list_empty(&obj->subscribers))
+		ubus_notify_subscription(obj);
 }
 
 void ubusd_free_object(struct ubus_object *obj)

ubusd_obj.h

@@ -79,5 +79,6 @@ static inline struct ubus_object *ubusd_find_object(uint32_t objid)
 void ubus_subscribe(struct ubus_object *obj, struct ubus_object *target, const char *method);
 void ubus_unsubscribe(struct ubus_subscription *s);
 void ubus_notify_unsubscribe(struct ubus_subscription *s);
+void ubus_notify_subscription(struct ubus_object *obj);
 
 #endif

ubusd_proto.c

@@ -416,6 +416,20 @@ void ubusd_proto_free_client(struct ubus_client *cl)
 	ubus_free_id(&clients, &cl->id);
 }
 
+void ubus_notify_subscription(struct ubus_object *obj)
+{
+	bool active = !list_empty(&obj->subscribers);
+	struct ubus_msg_buf *ub;
+
+	blob_buf_init(&b, 0);
+	blob_put_int32(&b, UBUS_ATTR_OBJID, obj->id.id);
+	blob_put_int8(&b, UBUS_ATTR_ACTIVE, active);
+
+	ub = ubus_msg_from_blob(false);
+	ubus_msg_init(ub, UBUS_MSG_NOTIFY, ++obj->invoke_seq, 0);
+	ubus_msg_send(obj->client, ub, true);
+}
+
 void ubus_notify_unsubscribe(struct ubus_subscription *s)
 {
 	struct ubus_msg_buf *ub;

ubusmsg.h

@@ -62,6 +62,13 @@ enum ubus_msg_type {
 	UBUS_MSG_SUBSCRIBE,
 	UBUS_MSG_UNSUBSCRIBE,
 
+	/*
+	 * send a notification to all subscribers of an object.
+	 * when sent from the server, it indicates a subscription
+	 * status change
+	 */
+	UBUS_MSG_NOTIFY,
+
 	/* must be last */
 	__UBUS_MSG_LAST,
 };
@@ -81,6 +88,8 @@ enum ubus_msg_attr {
 	UBUS_ATTR_DATA,
 	UBUS_ATTR_TARGET,
 
+	UBUS_ATTR_ACTIVE,
+
 	/* must be last */
 	UBUS_ATTR_MAX,
 };