FS-10167: Updated publish and broadcast to support adding and removing channels…

FS-10167: Updated publish and broadcast to support adding and removing channels at runtime, as well as removing protocols at runtime. In effect this provides cleanup on subscriptions when an entire channel or protocol becomes unavailable. Committing to test linux build.

libs/libblade/src/blade_mastermgr.c

@@ -122,8 +122,12 @@ KS_DECLARE(ks_status_t) blade_mastermgr_purge(blade_mastermgr_t *bmmgr, const ch
 		ks_hash_this(it, (const void **)&key, NULL, (void **)&bp);
 
 		if (blade_protocol_purge(bp, nodeid)) {
-			if (!cleanup) ks_hash_create(&cleanup, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK, pool);
-			ks_hash_insert(cleanup, (void *)key, bp);
+			if (!blade_protocol_controller_available(bp)) {
+				if (!cleanup) ks_hash_create(&cleanup, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK, pool);
+				ks_hash_insert(cleanup, (void *)key, bp);
+			} else {
+				// @todo not the last controller, may need to propagate that the controller is no longer available?
+			}
 		}
 	}
 	if (cleanup) {
@@ -133,6 +137,8 @@ KS_DECLARE(ks_status_t) blade_mastermgr_purge(blade_mastermgr_t *bmmgr, const ch
 
 			ks_hash_this(it, (const void **)&key, NULL, (void **)&bp);
 
+			blade_subscriptionmgr_broadcast(blade_handle_subscriptionmgr_get(bmmgr->handle), BLADE_RPCBROADCAST_COMMAND_PROTOCOL_REMOVE, NULL, blade_protocol_name_get(bp), blade_protocol_realm_get(bp), NULL, NULL, NULL, NULL, NULL);
+
 			ks_log(KS_LOG_DEBUG, "Protocol Removed: %s\n", key);
 			ks_hash_remove(bmmgr->protocols, (void *)key);
 		}
@@ -192,12 +198,49 @@ KS_DECLARE(ks_status_t) blade_mastermgr_controller_add(blade_mastermgr_t *bmmgr,
 		ks_hash_insert(bmmgr->protocols, (void *)ks_pstrdup(pool, key), bp);
 	}
 
-	blade_protocol_controllers_add(bp, controller);
+	blade_protocol_controller_add(bp, controller);
+
+	ks_hash_write_unlock(bmmgr->protocols);
 
 	ks_pool_free(&key);
 
+	return KS_STATUS_SUCCESS;
+}
+
+KS_DECLARE(ks_status_t) blade_mastermgr_controller_remove(blade_mastermgr_t *bmmgr, const char *protocol, const char *realm, const char *controller)
+{
+	ks_pool_t *pool = NULL;
+	blade_protocol_t *bp = NULL;
+	char *key = NULL;
+
+	ks_assert(bmmgr);
+	ks_assert(protocol);
+	ks_assert(realm);
+	ks_assert(controller);
+
+	pool = ks_pool_get(bmmgr);
+
+	key = ks_psprintf(pool, "%s@%s", protocol, realm);
+
+	ks_hash_write_lock(bmmgr->protocols);
+
+	bp = (blade_protocol_t *)ks_hash_search(bmmgr->protocols, (void *)key, KS_UNLOCKED);
+	if (bp) {
+		if (blade_protocol_controller_remove(bp, controller)) {
+			if (!blade_protocol_controller_available(bp)) {
+				// @todo broadcast protocol removal to remove all channel subscriptions
+				ks_log(KS_LOG_DEBUG, "Protocol Removed: %s\n", key);
+				ks_hash_remove(bmmgr->protocols, (void *)key);
+			} else {
+				// @todo not the last controller, may need to propagate when a specific controller becomes unavailable though
+			}
+		}
+	}
+
 	ks_hash_write_unlock(bmmgr->protocols);
 
+	ks_pool_free(&key);
+
 	return KS_STATUS_SUCCESS;
 }
 
@@ -249,7 +292,9 @@ KS_DECLARE(ks_status_t) blade_mastermgr_channel_remove(blade_mastermgr_t *bmmgr,
 		goto done;
 	}
 
-	blade_protocol_channel_remove(bp, channel);
+	if (blade_protocol_channel_remove(bp, channel)) {
+		blade_subscriptionmgr_broadcast(blade_handle_subscriptionmgr_get(bmmgr->handle), BLADE_RPCBROADCAST_COMMAND_CHANNEL_REMOVE, NULL, blade_protocol_name_get(bp), blade_protocol_realm_get(bp), channel, NULL, NULL, NULL, NULL);
+	}
 
 done:
 	ks_pool_free(&key);

libs/libblade/src/blade_protocol.c

@@ -98,10 +98,20 @@ KS_DECLARE(ks_status_t) blade_protocol_destroy(blade_protocol_t **bpP)
 	return KS_STATUS_SUCCESS;
 }
 
-KS_DECLARE(ks_bool_t) blade_protocol_purge(blade_protocol_t *bp, const char *nodeid)
+KS_DECLARE(const char *) blade_protocol_name_get(blade_protocol_t *bp)
 {
-	ks_bool_t ret = KS_FALSE;
+	ks_assert(bp);
+	return bp->name;
+}
+
+KS_DECLARE(const char *) blade_protocol_realm_get(blade_protocol_t *bp)
+{
+	ks_assert(bp);
+	return bp->realm;
+}
 
+KS_DECLARE(ks_bool_t) blade_protocol_purge(blade_protocol_t *bp, const char *nodeid)
+{
 	ks_assert(bp);
 	ks_assert(nodeid);
 
@@ -119,22 +129,16 @@ KS_DECLARE(ks_bool_t) blade_protocol_purge(blade_protocol_t *bp, const char *nod
 	}
 	ks_hash_write_unlock(bp->channels);
 
-	ks_hash_write_lock(bp->controllers);
-	if (ks_hash_remove(bp->controllers, (void *)nodeid)) {
-		ks_log(KS_LOG_DEBUG, "Protocol Controller Removed: %s from %s@%s\n", nodeid, bp->name, bp->realm);
-	}
-	ret = ks_hash_count(bp->controllers) == 0;
-	ks_hash_write_unlock(bp->controllers);
-
-	return ret;
+	return blade_protocol_controller_remove(bp, nodeid);
 }
 
-KS_DECLARE(cJSON *) blade_protocol_controllers_pack(blade_protocol_t *bp)
+KS_DECLARE(cJSON *) blade_protocol_controller_pack(blade_protocol_t *bp)
 {
 	cJSON *controllers = cJSON_CreateObject();
 
 	ks_assert(bp);
 
+	ks_hash_read_lock(bp->controllers);
 	for (ks_hash_iterator_t *it = ks_hash_first(bp->controllers, KS_UNLOCKED); it; it = ks_hash_next(&it)) {
 		const char *key = NULL;
 		void *value = NULL;
@@ -143,11 +147,12 @@ KS_DECLARE(cJSON *) blade_protocol_controllers_pack(blade_protocol_t *bp)
 
 		cJSON_AddItemToArray(controllers, cJSON_CreateString(key));
 	}
+	ks_hash_read_unlock(bp->controllers);
 
 	return controllers;
 }
 
-KS_DECLARE(ks_status_t) blade_protocol_controllers_add(blade_protocol_t *bp, const char *nodeid)
+KS_DECLARE(ks_status_t) blade_protocol_controller_add(blade_protocol_t *bp, const char *nodeid)
 {
 	char *key = NULL;
 
@@ -162,6 +167,29 @@ KS_DECLARE(ks_status_t) blade_protocol_controllers_add(blade_protocol_t *bp, con
 	return KS_STATUS_SUCCESS;
 }
 
+KS_DECLARE(ks_bool_t) blade_protocol_controller_remove(blade_protocol_t *bp, const char *nodeid)
+{
+	ks_bool_t ret  = KS_FALSE;
+
+	ks_assert(bp);
+	ks_assert(nodeid);
+
+	ks_hash_write_lock(bp->controllers);
+	if (ks_hash_remove(bp->controllers, (void *)nodeid)) {
+		ret = KS_TRUE;
+		ks_log(KS_LOG_DEBUG, "Protocol Controller Removed: %s from %s@%s\n", nodeid, bp->name, bp->realm);
+	}
+	ks_hash_write_unlock(bp->controllers);
+
+	return ret;
+}
+
+KS_DECLARE(ks_bool_t) blade_protocol_controller_available(blade_protocol_t *bp)
+{
+	ks_assert(bp);
+	return ks_hash_count(bp->controllers) > 0;
+}
+
 KS_DECLARE(ks_status_t) blade_protocol_channel_add(blade_protocol_t *bp, const char *name)
 {
 	ks_status_t ret = KS_STATUS_SUCCESS;
@@ -195,16 +223,23 @@ done:
 	return ret;
 }
 
-KS_DECLARE(ks_status_t) blade_protocol_channel_remove(blade_protocol_t *bp, const char *name)
+KS_DECLARE(ks_bool_t) blade_protocol_channel_remove(blade_protocol_t *bp, const char *name)
 {
+	ks_bool_t ret = KS_FALSE;
+	ks_hash_t *authorized = NULL;
+
 	ks_assert(bp);
 	ks_assert(name);
 
-	ks_hash_remove(bp->channels, (void *)name);
-
-	ks_log(KS_LOG_DEBUG, "Protocol Channel Removed: %s from %s@%s\n", name, bp->name, bp->realm);
+	ks_hash_write_lock(bp->channels);
+	if ((authorized = ks_hash_remove(bp->channels, (void *)name))) {
+		ret = KS_TRUE;
+		ks_log(KS_LOG_DEBUG, "Protocol Channel Removed: %s from %s@%s\n", name, bp->name, bp->realm);
+		ks_hash_destroy(&authorized);
+	}
+	ks_hash_write_unlock(bp->channels);
 
-	return KS_STATUS_SUCCESS;
+	return ret;
 }
 
 KS_DECLARE(ks_status_t) blade_protocol_channel_authorize(blade_protocol_t *bp, ks_bool_t remove, const char *channel, const char *controller, const char *target)
@@ -233,8 +268,7 @@ KS_DECLARE(ks_status_t) blade_protocol_channel_authorize(blade_protocol_t *bp, k
 			if (ks_hash_remove(authorizations, (void *)target)) {
 				ks_log(KS_LOG_DEBUG, "Protocol Channel Authorization Removed: %s from %s@%s/%s\n", target, bp->name, bp->realm, channel);
 			} else ret = KS_STATUS_NOT_FOUND;
-		}
-		else {
+		} else {
 			ks_hash_insert(authorizations, (void *)ks_pstrdup(ks_pool_get(bp), target), (void *)KS_TRUE);
 			ks_log(KS_LOG_DEBUG, "Protocol Channel Authorization Added: %s to %s@%s/%s\n", target, bp->name, bp->realm, channel);
 		}

libs/libblade/src/blade_subscriptionmgr.c

@@ -145,6 +145,56 @@ KS_DECLARE(blade_subscription_t *) blade_subscriptionmgr_subscription_lookup(bla
 	return bsub;
 }
 
+KS_DECLARE(ks_status_t) blade_subscriptionmgr_subscription_remove(blade_subscriptionmgr_t *bsmgr, const char *protocol, const char *realm, const char *channel)
+{
+	ks_pool_t *pool = NULL;
+	char *bsub_key = NULL;
+	blade_subscription_t *bsub = NULL;
+	ks_hash_t *subscribers = NULL;
+	ks_hash_t *subscriptions = NULL;
+
+	ks_assert(bsmgr);
+	ks_assert(protocol);
+	ks_assert(realm);
+	ks_assert(channel);
+
+	pool = ks_pool_get(bsmgr);
+
+	bsub_key = ks_psprintf(pool, "%s@%s/%s", protocol, realm, channel);
+
+	ks_hash_write_lock(bsmgr->subscriptions);
+
+	bsub = (blade_subscription_t *)ks_hash_search(bsmgr->subscriptions, (void *)bsub_key, KS_UNLOCKED);
+
+	subscribers = blade_subscription_subscribers_get(bsub);
+	ks_assert(subscribers);
+
+	for (ks_hash_iterator_t *it = ks_hash_first(subscribers, KS_UNLOCKED); it; it = ks_hash_next(&it)) {
+		void *key = NULL;
+		void *value = NULL;
+
+		ks_hash_this(it, (const void **)&key, NULL, &value);
+
+		subscriptions = (ks_hash_t *)ks_hash_search(bsmgr->subscriptions_cleanup, key, KS_UNLOCKED);
+
+		ks_log(KS_LOG_DEBUG, "Subscriber Removed: %s from %s\n", key, bsub_key);
+		ks_hash_remove(subscriptions, bsub_key);
+
+		if (ks_hash_count(subscriptions) == 0) {
+			ks_hash_remove(bsmgr->subscriptions_cleanup, key);
+		}
+	}
+
+	ks_log(KS_LOG_DEBUG, "Subscription Removed: %s\n", bsub_key);
+	ks_hash_remove(bsmgr->subscriptions, (void *)bsub_key);
+
+	ks_hash_write_unlock(bsmgr->subscriptions);
+
+	ks_pool_free(&bsub_key);
+
+	return KS_STATUS_SUCCESS;
+}
+
 KS_DECLARE(ks_bool_t) blade_subscriptionmgr_subscriber_add(blade_subscriptionmgr_t *bsmgr, blade_subscription_t **bsubP, const char *protocol, const char *realm, const char *channel, const char *subscriber)
 {
 	ks_pool_t *pool = NULL;
@@ -295,7 +345,7 @@ KS_DECLARE(void) blade_subscriptionmgr_purge(blade_subscriptionmgr_t *bsmgr, con
 	}
 }
 
-KS_DECLARE(ks_status_t) blade_subscriptionmgr_broadcast(blade_subscriptionmgr_t *bsmgr, const char *excluded_nodeid, const char *protocol, const char *realm, const char *channel, const char *event, cJSON *params, blade_rpc_response_callback_t callback, void *data)
+KS_DECLARE(ks_status_t) blade_subscriptionmgr_broadcast(blade_subscriptionmgr_t *bsmgr, blade_rpcbroadcast_command_t command, const char *excluded_nodeid, const char *protocol, const char *realm, const char *channel, const char *event, cJSON *params, blade_rpc_response_callback_t callback, void *data)
 {
 	ks_pool_t *pool = NULL;
 	const char *bsub_key = NULL;
@@ -303,67 +353,152 @@ KS_DECLARE(ks_status_t) blade_subscriptionmgr_broadcast(blade_subscriptionmgr_t
 	blade_session_t *bs = NULL;
 	cJSON *req = NULL;
 	cJSON *req_params = NULL;
+	ks_hash_t *routers = NULL;
+	ks_hash_t *channels = NULL;
 
 	ks_assert(bsmgr);
 	ks_assert(protocol);
 	ks_assert(realm);
-	ks_assert(channel);
 
 	pool = ks_pool_get(bsmgr);
 
-	bsub_key = ks_psprintf(pool, "%s@%s/%s", protocol, realm, channel);
+	switch (command) {
+	case BLADE_RPCBROADCAST_COMMAND_EVENT:
+		ks_assert(event);
+	case BLADE_RPCBROADCAST_COMMAND_CHANNEL_REMOVE:
+		ks_assert(channel);
+		bsub_key = ks_psprintf(pool, "%s@%s/%s", protocol, realm, channel);
 
-	blade_rpc_request_raw_create(pool, &req, &req_params, NULL, "blade.broadcast");
-	cJSON_AddStringToObject(req_params, "protocol", protocol);
-	cJSON_AddStringToObject(req_params, "realm", realm);
-	cJSON_AddStringToObject(req_params, "channel", channel);
-	cJSON_AddStringToObject(req_params, "event", event);
-	if (params) cJSON_AddItemToObject(req_params, "params", cJSON_Duplicate(params, 1));
+		ks_hash_read_lock(bsmgr->subscriptions);
 
-	ks_hash_read_lock(bsmgr->subscriptions);
+		bsub = (blade_subscription_t *)ks_hash_search(bsmgr->subscriptions, (void *)bsub_key, KS_UNLOCKED);
+		if (bsub) {
+			ks_hash_t *subscribers = blade_subscription_subscribers_get(bsub);
 
-	bsub = (blade_subscription_t *)ks_hash_search(bsmgr->subscriptions, (void *)bsub_key, KS_UNLOCKED);
-	if (bsub) {
-		ks_hash_t *subscribers = blade_subscription_subscribers_get(bsub);
+			ks_assert(subscribers);
+
+			for (ks_hash_iterator_t *it = ks_hash_first(subscribers, KS_UNLOCKED); it; it = ks_hash_next(&it)) {
+				void *key = NULL;
+				void *value = NULL;
+
+				ks_hash_this(it, (const void **)&key, NULL, &value);
 
-		ks_assert(subscribers);
+				if (excluded_nodeid && !ks_safe_strcasecmp(excluded_nodeid, (const char *)key)) continue;
 
-		for (ks_hash_iterator_t *it = ks_hash_first(subscribers, KS_UNLOCKED); it; it = ks_hash_next(&it)) {
+				// @todo broadcast producer is also a local subscriber... requires special consideration with no session to request through
+				if (blade_upstreammgr_localid_compare(blade_handle_upstreammgr_get(bsmgr->handle), (const char *)key)) continue;
+
+				bs = blade_routemgr_route_lookup(blade_handle_routemgr_get(bsmgr->handle), (const char *)key);
+				if (bs) {
+					if (!routers) ks_hash_create(&routers, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_NOLOCK | KS_HASH_FLAG_DUP_CHECK, pool);
+					if (!ks_hash_search(routers, blade_session_id_get(bs), KS_UNLOCKED)) ks_hash_insert(routers, blade_session_id_get(bs), bs);
+					else blade_session_read_unlock(bs);
+				}
+			}
+			if (command == BLADE_RPCBROADCAST_COMMAND_CHANNEL_REMOVE) {
+				if (!channels) ks_hash_create(&channels, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_NOLOCK | KS_HASH_FLAG_DUP_CHECK, pool);
+				ks_hash_insert(channels, channel, (void *)KS_TRUE);
+			}
+		}
+
+		ks_hash_read_unlock(bsmgr->subscriptions);
+		break;
+	case BLADE_RPCBROADCAST_COMMAND_PROTOCOL_REMOVE:
+		bsub_key = ks_psprintf(pool, "%s@%s", protocol, realm);
+
+		ks_hash_read_lock(bsmgr->subscriptions);
+
+		for (ks_hash_iterator_t *it = ks_hash_first(bsmgr->subscriptions, KS_UNLOCKED); it; it = ks_hash_next(&it)) {
 			void *key = NULL;
 			void *value = NULL;
 
 			ks_hash_this(it, (const void **)&key, NULL, &value);
 
-			if (excluded_nodeid && !ks_safe_strcasecmp(excluded_nodeid, (const char *)key)) continue;
+			bsub = (blade_subscription_t *)value;
 
-			// @todo broadcast producer is also a local subscriber... requires special consideration with no session to request through
-			if (blade_upstreammgr_localid_compare(blade_handle_upstreammgr_get(bsmgr->handle), (const char *)key)) continue;
+			if (ks_stristr(bsub_key, (const char *)key) == (const char *)key) {
+				ks_hash_t *subscribers = blade_subscription_subscribers_get(bsub);
 
-			bs = blade_routemgr_route_lookup(blade_handle_routemgr_get(bsmgr->handle), (const char *)key);
-			if (bs) {
-				ks_log(KS_LOG_DEBUG, "Broadcasting: %s through %s\n", bsub_key, blade_session_id_get(bs));
+				ks_assert(subscribers);
 
-				blade_session_send(bs, req, callback, data);
+				for (ks_hash_iterator_t *it2 = ks_hash_first(subscribers, KS_UNLOCKED); it2; it2 = ks_hash_next(&it2)) {
+					void *key2 = NULL;
+					void *value2 = NULL;
 
-				blade_session_read_unlock(bs);
+					ks_hash_this(it2, (const void **)&key2, NULL, &value2);
+
+					if (excluded_nodeid && !ks_safe_strcasecmp(excluded_nodeid, (const char *)key2)) continue;
+
+					// @todo broadcast producer is also a local subscriber... requires special consideration with no session to request through
+					if (blade_upstreammgr_localid_compare(blade_handle_upstreammgr_get(bsmgr->handle), (const char *)key2)) continue;
+
+					bs = blade_routemgr_route_lookup(blade_handle_routemgr_get(bsmgr->handle), (const char *)key2);
+					if (bs) {
+						if (!routers) ks_hash_create(&routers, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_NOLOCK | KS_HASH_FLAG_DUP_CHECK, pool);
+						if (!ks_hash_search(routers, blade_session_id_get(bs), KS_UNLOCKED)) ks_hash_insert(routers, blade_session_id_get(bs), bs);
+						else blade_session_read_unlock(bs);
+					}
+				}
+
+				if (!channels) ks_hash_create(&channels, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_NOLOCK | KS_HASH_FLAG_DUP_CHECK, pool);
+				ks_hash_insert(channels, blade_subscription_channel_get(bsub), (void *)KS_TRUE);
 			}
 		}
+
+		ks_hash_read_unlock(bsmgr->subscriptions);
+		break;
+	default: return KS_STATUS_ARG_INVALID;
 	}
 
-	ks_hash_read_unlock(bsmgr->subscriptions);
 
 	bs = blade_upstreammgr_session_get(blade_handle_upstreammgr_get(bsmgr->handle));
 	if (bs) {
 		if (!excluded_nodeid || ks_safe_strcasecmp(blade_session_id_get(bs), excluded_nodeid)) {
+			if (!routers) ks_hash_create(&routers, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_NOLOCK | KS_HASH_FLAG_DUP_CHECK, pool);
+			ks_hash_insert(routers, blade_session_id_get(bs), bs);
+		}
+		else blade_session_read_unlock(bs);
+	}
+
+	if (channels) {
+		for (ks_hash_iterator_t *it = ks_hash_first(channels, KS_UNLOCKED); it; it = ks_hash_next(&it)) {
+			void *key = NULL;
+			void *value = NULL;
+
+			ks_hash_this(it, (const void **)&key, NULL, &value);
+
+			blade_subscriptionmgr_subscription_remove(bsmgr, protocol, realm, (const char *)key);
+		}
+		ks_hash_destroy(&channels);
+	}
+
+	if (routers) {
+		for (ks_hash_iterator_t *it = ks_hash_first(routers, KS_UNLOCKED); it; it = ks_hash_next(&it)) {
+			void *key = NULL;
+			void *value = NULL;
+
+			ks_hash_this(it, (const void **)&key, NULL, &value);
+
+			bs = (blade_session_t *)value;
+
+			blade_rpc_request_raw_create(pool, &req, &req_params, NULL, "blade.broadcast");
+			cJSON_AddNumberToObject(req_params, "command", command);
+			cJSON_AddStringToObject(req_params, "protocol", protocol);
+			cJSON_AddStringToObject(req_params, "realm", realm);
+			if (channel) cJSON_AddStringToObject(req_params, "channel", channel);
+			if (event) cJSON_AddStringToObject(req_params, "event", event);
+			if (params) cJSON_AddItemToObject(req_params, "params", cJSON_Duplicate(params, 1));
+
 			ks_log(KS_LOG_DEBUG, "Broadcasting: %s through %s\n", bsub_key, blade_session_id_get(bs));
 
 			blade_session_send(bs, req, callback, data);
-		}
 
-		blade_session_read_unlock(bs);
-	}
+			cJSON_Delete(req);
 
-	cJSON_Delete(req);
+			blade_session_read_unlock(bs);
+		}
+		ks_hash_destroy(&routers);
+	}
 
 	ks_pool_free(&bsub_key);
 

libs/libblade/src/include/blade_mastermgr.h

@@ -42,6 +42,7 @@ KS_DECLARE(blade_handle_t *) blade_mastermgr_handle_get(blade_mastermgr_t *bmmgr
 KS_DECLARE(ks_status_t) blade_mastermgr_purge(blade_mastermgr_t *bmmgr, const char *nodeid);
 KS_DECLARE(blade_protocol_t *) blade_mastermgr_protocol_lookup(blade_mastermgr_t *bmmgr, const char *protocol, const char *realm);
 KS_DECLARE(ks_status_t) blade_mastermgr_controller_add(blade_mastermgr_t *bmmgr, const char *protocol, const char *realm, const char *controller);
+KS_DECLARE(ks_status_t) blade_mastermgr_controller_remove(blade_mastermgr_t *bmmgr, const char *protocol, const char *realm, const char *controller);
 KS_DECLARE(ks_status_t) blade_mastermgr_channel_add(blade_mastermgr_t *bmmgr, const char *protocol, const char *realm, const char *channel);
 KS_DECLARE(ks_status_t) blade_mastermgr_channel_remove(blade_mastermgr_t *bmmgr, const char *protocol, const char *realm, const char *channel);
 KS_DECLARE(ks_status_t) blade_mastermgr_channel_authorize(blade_mastermgr_t *bmmgr, ks_bool_t remove, const char *protocol, const char *realm, const char *channel, const char *controller, const char *target);

libs/libblade/src/include/blade_protocol.h

@@ -38,11 +38,15 @@
 KS_BEGIN_EXTERN_C
 KS_DECLARE(ks_status_t) blade_protocol_create(blade_protocol_t **bpP, ks_pool_t *pool, const char *name, const char *realm);
 KS_DECLARE(ks_status_t) blade_protocol_destroy(blade_protocol_t **bpP);
+KS_DECLARE(const char *) blade_protocol_name_get(blade_protocol_t *bp);
+KS_DECLARE(const char *) blade_protocol_realm_get(blade_protocol_t *bp);
 KS_DECLARE(ks_bool_t) blade_protocol_purge(blade_protocol_t *bp, const char *nodeid);
-KS_DECLARE(ks_status_t) blade_protocol_controllers_add(blade_protocol_t *bp, const char *nodeid);
-KS_DECLARE(cJSON *) blade_protocol_controllers_pack(blade_protocol_t *bp);
+KS_DECLARE(cJSON *) blade_protocol_controller_pack(blade_protocol_t *bp);
+KS_DECLARE(ks_status_t) blade_protocol_controller_add(blade_protocol_t *bp, const char *nodeid);
+KS_DECLARE(ks_bool_t) blade_protocol_controller_remove(blade_protocol_t *bp, const char *nodeid);
+KS_DECLARE(ks_bool_t) blade_protocol_controller_available(blade_protocol_t *bp);
 KS_DECLARE(ks_status_t) blade_protocol_channel_add(blade_protocol_t *bp, const char *name);
-KS_DECLARE(ks_status_t) blade_protocol_channel_remove(blade_protocol_t *bp, const char *name);
+KS_DECLARE(ks_bool_t) blade_protocol_channel_remove(blade_protocol_t *bp, const char *name);
 KS_DECLARE(ks_status_t) blade_protocol_channel_authorize(blade_protocol_t *bp, ks_bool_t remove, const char *channel, const char *controller, const char *target);
 KS_DECLARE(ks_bool_t) blade_protocol_channel_verify(blade_protocol_t *bp, const char *channel, const char *target);
 KS_END_EXTERN_C

libs/libblade/src/include/blade_stack.h

@@ -60,7 +60,7 @@ KS_DECLARE(ks_status_t) blade_handle_connect(blade_handle_t *bh, blade_connectio
 
 KS_DECLARE(ks_status_t) blade_handle_rpcregister(blade_handle_t *bh, const char *nodeid, ks_bool_t remove, blade_rpc_response_callback_t callback, void *data);
 
-KS_DECLARE(ks_status_t) blade_handle_rpcpublish(blade_handle_t *bh, const char *protocol, const char *realm, cJSON *channels, blade_rpc_response_callback_t callback, void *data);
+KS_DECLARE(ks_status_t) blade_handle_rpcpublish(blade_handle_t *bh, blade_rpcpublish_command_t command, const char *protocol, const char *realm, cJSON *channels, blade_rpc_response_callback_t callback, void *data);
 
 KS_DECLARE(ks_status_t) blade_handle_rpcauthorize(blade_handle_t *bh, const char *nodeid, ks_bool_t remove, const char *protocol, const char *realm, cJSON *channels, blade_rpc_response_callback_t callback, void *data);
 
@@ -73,7 +73,7 @@ KS_DECLARE(cJSON *) blade_rpcexecute_request_params_get(blade_rpc_request_t *brp
 KS_DECLARE(cJSON *) blade_rpcexecute_response_result_get(blade_rpc_response_t *brpcres);
 KS_DECLARE(void) blade_rpcexecute_response_send(blade_rpc_request_t *brpcreq, cJSON *result);
 
-KS_DECLARE(ks_status_t) blade_handle_rpcsubscribe(blade_handle_t *bh, const char *protocol, const char *realm, cJSON *subscribe_channels, cJSON *unsubscribe_channels, blade_rpc_response_callback_t callback, void *data, blade_rpc_request_callback_t channel_callback, void *channel_data);
+KS_DECLARE(ks_status_t) blade_handle_rpcsubscribe(blade_handle_t *bh, blade_rpcsubscribe_command_t command, const char *protocol, const char *realm, cJSON *channels, blade_rpc_response_callback_t callback, void *data, blade_rpc_request_callback_t channel_callback, void *channel_data);
 
 KS_DECLARE(ks_status_t) blade_handle_rpcbroadcast(blade_handle_t *bh, const char *protocol, const char *realm, const char *channel, const char *event, cJSON *params, blade_rpc_response_callback_t callback, void *data);
 KS_DECLARE(cJSON *) blade_rpcbroadcast_request_params_get(blade_rpc_request_t *brpcreq);

libs/libblade/src/include/blade_subscriptionmgr.h

@@ -40,10 +40,11 @@ KS_DECLARE(ks_status_t) blade_subscriptionmgr_create(blade_subscriptionmgr_t **b
 KS_DECLARE(ks_status_t) blade_subscriptionmgr_destroy(blade_subscriptionmgr_t **bsmgrP);
 KS_DECLARE(blade_handle_t *) blade_subscriptionmgr_handle_get(blade_subscriptionmgr_t *bsmgr);
 KS_DECLARE(blade_subscription_t *) blade_subscriptionmgr_subscription_lookup(blade_subscriptionmgr_t *bsmgr, const char *protocol, const char *realm, const char *channel);
+KS_DECLARE(ks_status_t) blade_subscriptionmgr_subscription_remove(blade_subscriptionmgr_t *bsmgr, const char *protocol, const char *realm, const char *channel);
 KS_DECLARE(ks_bool_t) blade_subscriptionmgr_subscriber_add(blade_subscriptionmgr_t *bsmgr, blade_subscription_t **bsubP, const char *protocol, const char *realm, const char *channel, const char *subscriber);
 KS_DECLARE(ks_bool_t) blade_subscriptionmgr_subscriber_remove(blade_subscriptionmgr_t *bsmgr, blade_subscription_t **bsubP, const char *protocol, const char *realm, const char *channel, const char *subscriber);
 KS_DECLARE(void) blade_subscriptionmgr_purge(blade_subscriptionmgr_t *bsmgr, const char *target);
-KS_DECLARE(ks_status_t) blade_subscriptionmgr_broadcast(blade_subscriptionmgr_t *bsmgr, const char *excluded_nodeid, const char *protocol, const char *realm, const char *channel, const char *event, cJSON *params, blade_rpc_response_callback_t callback, void *data);
+KS_DECLARE(ks_status_t) blade_subscriptionmgr_broadcast(blade_subscriptionmgr_t *bsmgr, blade_rpcbroadcast_command_t command, const char *excluded_nodeid, const char *protocol, const char *realm, const char *channel, const char *event, cJSON *params, blade_rpc_response_callback_t callback, void *data);
 KS_END_EXTERN_C
 
 #endif

libs/libblade/src/include/blade_types.h

@@ -134,6 +134,28 @@ struct blade_transport_callbacks_s {
 typedef void (*blade_session_callback_t)(blade_session_t *bs, blade_session_state_condition_t condition, void *data);
 
 
+typedef enum {
+	BLADE_RPCPUBLISH_COMMAND_NONE,
+	BLADE_RPCPUBLISH_COMMAND_CONTROLLER_ADD,
+	BLADE_RPCPUBLISH_COMMAND_CONTROLLER_REMOVE,
+	BLADE_RPCPUBLISH_COMMAND_CHANNEL_ADD,
+	BLADE_RPCPUBLISH_COMMAND_CHANNEL_REMOVE,
+} blade_rpcpublish_command_t;
+
+typedef enum {
+	BLADE_RPCSUBSCRIBE_COMMAND_NONE,
+	BLADE_RPCSUBSCRIBE_COMMAND_SUBSCRIBER_ADD,
+	BLADE_RPCSUBSCRIBE_COMMAND_SUBSCRIBER_REMOVE,
+} blade_rpcsubscribe_command_t;
+
+typedef enum {
+	BLADE_RPCBROADCAST_COMMAND_NONE,
+	BLADE_RPCBROADCAST_COMMAND_EVENT,
+	BLADE_RPCBROADCAST_COMMAND_PROTOCOL_REMOVE,
+	BLADE_RPCBROADCAST_COMMAND_CHANNEL_REMOVE,
+} blade_rpcbroadcast_command_t;
+
+
 KS_END_EXTERN_C
 
 #endif

libs/libblade/test/bladec.c

@@ -297,7 +297,7 @@ void command_subscribe(blade_handle_t *bh, char *args)
 
 	channels = cJSON_CreateArray();
 	cJSON_AddItemToArray(channels, cJSON_CreateString("test"));
-	blade_handle_rpcsubscribe(bh, "test", "mydomain.com", channels, NULL, blade_subscribe_response_handler, NULL, test_event_request_handler, NULL);
+	blade_handle_rpcsubscribe(bh, BLADE_RPCSUBSCRIBE_COMMAND_SUBSCRIBER_ADD, "test", "mydomain.com", channels, blade_subscribe_response_handler, NULL, test_event_request_handler, NULL);
 	cJSON_Delete(channels);
 }
 

libs/libblade/test/blades.c

@@ -244,7 +244,7 @@ void command_publish(blade_handle_t *bh, char *args)
 	blade_rpcmgr_protocolrpc_add(blade_handle_rpcmgr_get(bh), brpc);
 
 	// @todo build up json-based method schema for each protocolrpc registered above, and pass into blade_handle_rpcpublish() to attach to the request, to be stored in the blade_protocol_t tracked by the master node
-	blade_handle_rpcpublish(bh, "test", "mydomain.com", NULL, blade_publish_response_handler, NULL);
+	blade_handle_rpcpublish(bh, BLADE_RPCPUBLISH_COMMAND_CONTROLLER_ADD, "test", "mydomain.com", NULL, blade_publish_response_handler, NULL);
 }
 
 void command_broadcast(blade_handle_t *bh, char *args)

libs/libblade/test/testcli.c

@@ -70,22 +70,25 @@ ks_bool_t test_locate_response_handler(blade_rpc_response_t *brpcres, void *data
 	ks_assert(res_result_realm);
 
 	res_result_controllers = cJSON_GetObjectItem(res_result, "controllers");
-	ks_assert(res_result_controllers);
 
 	ks_log(KS_LOG_DEBUG, "Session (%s) locate (%s@%s) response processing\n", blade_session_id_get(bs), res_result_protocol, res_result_realm);
 
-	for (int index = 0; index < cJSON_GetArraySize(res_result_controllers); ++index) {
-		cJSON *elem = cJSON_GetArrayItem(res_result_controllers, index);
-		if (elem->type == cJSON_String) {
-			nodeid = elem->valuestring;
+	if (res_result_controllers) {
+		for (int index = 0; index < cJSON_GetArraySize(res_result_controllers); ++index) {
+			cJSON *elem = cJSON_GetArrayItem(res_result_controllers, index);
+			if (elem->type == cJSON_String) {
+				nodeid = elem->valuestring;
+			}
 		}
 	}
 
 	blade_session_read_unlock(bs);
 
 	if (nodeid) {
+		if (g_testcon_nodeid) ks_pool_free(&g_testcon_nodeid);
 		g_testcon_nodeid = ks_pstrdup(ks_pool_get(bh), nodeid);
 	}
+
 	ks_log(KS_LOG_DEBUG, "Session (%s) locate (%s@%s) provider (%s)\n", blade_session_id_get(bs), res_result_protocol, res_result_realm, g_testcon_nodeid);
 
 	return KS_FALSE;
@@ -384,7 +387,8 @@ void command_subscribe(blade_handle_t *bh, char *args)
 
 	channels = cJSON_CreateArray();
 	cJSON_AddItemToArray(channels, cJSON_CreateString("channel"));
-	blade_handle_rpcsubscribe(bh, "test", "mydomain.com", channels, NULL, NULL, NULL, test_channel_handler, NULL);
+	if (args && args[0]) cJSON_AddItemToArray(channels, cJSON_CreateString(args));
+	blade_handle_rpcsubscribe(bh, BLADE_RPCSUBSCRIBE_COMMAND_SUBSCRIBER_ADD, "test", "mydomain.com", channels, NULL, NULL, test_channel_handler, NULL);
 	cJSON_Delete(channels);
 }
 
@@ -399,7 +403,8 @@ void command_unsubscribe(blade_handle_t *bh, char *args)
 
 	channels = cJSON_CreateArray();
 	cJSON_AddItemToArray(channels, cJSON_CreateString("channel"));
-	blade_handle_rpcsubscribe(bh, "test", "mydomain.com", NULL, channels, test_subscribe_response_handler, NULL, test_channel_handler, NULL);
+	if (args && args[0]) cJSON_AddItemToArray(channels, cJSON_CreateString(args));
+	blade_handle_rpcsubscribe(bh, BLADE_RPCSUBSCRIBE_COMMAND_SUBSCRIBER_REMOVE, "test", "mydomain.com", channels, test_subscribe_response_handler, NULL, test_channel_handler, NULL);
 	cJSON_Delete(channels);
 }
 

libs/libblade/test/testcon.c

@@ -16,9 +16,13 @@ struct command_def_s {
 };
 
 void command_quit(blade_handle_t *bh, char *args);
+void command_channeladd(blade_handle_t *bh, char *args);
+void command_channelremove(blade_handle_t *bh, char *args);
 
 static const struct command_def_s command_defs[] = {
 	{ "quit", command_quit },
+	{ "channeladd", command_channeladd },
+	{ "channelremove", command_channelremove },
 
 	{ NULL, NULL }
 };
@@ -27,9 +31,12 @@ struct testproto_s {
 	blade_handle_t *handle;
 	ks_pool_t *pool;
 	ks_hash_t *participants;
+	ks_hash_t *channels;
 };
 typedef struct testproto_s testproto_t;
 
+testproto_t *g_test = NULL;
+
 static void testproto_cleanup(void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
 {
 	//testproto_t *test = (testproto_t *)ptr;
@@ -62,6 +69,7 @@ ks_status_t testproto_create(testproto_t **testP, blade_handle_t *bh)
 	test->pool = pool;
 
 	ks_hash_create(&test->participants, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, pool);
+	ks_hash_create(&test->channels, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, pool);
 
 	ks_pool_set_cleanup(test, NULL, testproto_cleanup);
 
@@ -157,6 +165,13 @@ ks_bool_t test_join_request_handler(blade_rpc_request_t *brpcreq, void *data)
 	// authorize channels with the master for the requester
 	channels = cJSON_CreateArray();
 	cJSON_AddItemToArray(channels, cJSON_CreateString("channel"));
+	for (ks_hash_iterator_t *it = ks_hash_first(test->channels, KS_UNLOCKED); it; it = ks_hash_next(&it)) {
+		void *key = NULL;
+		void *value = NULL;
+
+		ks_hash_this(it, &key, NULL, &value);
+		cJSON_AddItemToArray(channels, cJSON_CreateString((const char *)key));
+	}
 
 	blade_handle_rpcauthorize(bh, requester_nodeid, KS_FALSE, "test", "mydomain.com", channels, NULL, NULL);
 
@@ -220,6 +235,13 @@ ks_bool_t test_leave_request_handler(blade_rpc_request_t *brpcreq, void *data)
 	// deauthorize channels with the master for the requester
 	channels = cJSON_CreateArray();
 	cJSON_AddItemToArray(channels, cJSON_CreateString("channel"));
+	for (ks_hash_iterator_t *it = ks_hash_first(test->channels, KS_UNLOCKED); it; it = ks_hash_next(&it)) {
+		void *key = NULL;
+		void *value = NULL;
+
+		ks_hash_this(it, &key, NULL, &value);
+		cJSON_AddItemToArray(channels, cJSON_CreateString((const char *)key));
+	}
 
 	blade_handle_rpcauthorize(bh, requester_nodeid, KS_TRUE, "test", "mydomain.com", channels, NULL, NULL);
 
@@ -310,7 +332,6 @@ int main(int argc, char **argv)
 	config_setting_t *config_blade = NULL;
 	const char *cfgpath = "testcon.cfg";
 	const char *autoconnect = NULL;
-	testproto_t *test = NULL;
 
 	ks_global_set_default_logger(KS_LOG_LEVEL_DEBUG);
 
@@ -346,7 +367,7 @@ int main(int argc, char **argv)
 		return EXIT_FAILURE;
 	}
 
-	testproto_create(&test, bh);
+	testproto_create(&g_test, bh);
 
 	if (autoconnect) {
 		blade_connection_t *bc = NULL;
@@ -366,19 +387,19 @@ int main(int argc, char **argv)
 			// @todo use session state change callback to know when the session is ready and the realm(s) available from blade.connect, this hack temporarily ensures it's ready before trying to publish upstream
 			ks_sleep_ms(3000);
 
-			blade_rpc_create(&brpc, bh, "test.join", "test", "mydomain.com", test_join_request_handler, (void *)test);
+			blade_rpc_create(&brpc, bh, "test.join", "test", "mydomain.com", test_join_request_handler, (void *)g_test);
 			blade_rpcmgr_protocolrpc_add(blade_handle_rpcmgr_get(bh), brpc);
 
-			blade_rpc_create(&brpc, bh, "test.leave", "test", "mydomain.com", test_leave_request_handler, (void *)test);
+			blade_rpc_create(&brpc, bh, "test.leave", "test", "mydomain.com", test_leave_request_handler, (void *)g_test);
 			blade_rpcmgr_protocolrpc_add(blade_handle_rpcmgr_get(bh), brpc);
 
-			blade_rpc_create(&brpc, bh, "test.talk", "test", "mydomain.com", test_talk_request_handler, (void *)test);
+			blade_rpc_create(&brpc, bh, "test.talk", "test", "mydomain.com", test_talk_request_handler, (void *)g_test);
 			blade_rpcmgr_protocolrpc_add(blade_handle_rpcmgr_get(bh), brpc);
 
 			channels = cJSON_CreateArray();
 			cJSON_AddItemToArray(channels, cJSON_CreateString("channel"));
 
-			blade_handle_rpcpublish(bh, "test", "mydomain.com", channels, test_publish_response_handler, (void *)test);
+			blade_handle_rpcpublish(bh, BLADE_RPCPUBLISH_COMMAND_CONTROLLER_ADD, "test", "mydomain.com", channels, test_publish_response_handler, (void *)g_test);
 
 			cJSON_Delete(channels);
 		}
@@ -388,7 +409,7 @@ int main(int argc, char **argv)
 
 	blade_handle_destroy(&bh);
 
-	testproto_destroy(&test);
+	testproto_destroy(&g_test);
 
 	config_destroy(&config);
 
@@ -459,6 +480,51 @@ void command_quit(blade_handle_t *bh, char *args)
 	g_shutdown = KS_TRUE;
 }
 
+void command_channeladd(blade_handle_t *bh, char *args)
+{
+	cJSON *channels = NULL;
+
+	ks_assert(bh);
+	ks_assert(args);
+
+	if (!args[0]) {
+		ks_log(KS_LOG_INFO, "Requires channel argument");
+		return;
+	}
+
+	ks_hash_insert(g_test->channels, (void *)ks_pstrdup(g_test->pool, args), (void *)KS_TRUE);
+
+	channels = cJSON_CreateArray();
+	cJSON_AddItemToArray(channels, cJSON_CreateString(args));
+
+	blade_handle_rpcpublish(bh, BLADE_RPCPUBLISH_COMMAND_CHANNEL_ADD, "test", "mydomain.com", channels, test_publish_response_handler, (void *)g_test);
+
+	cJSON_Delete(channels);
+}
+
+void command_channelremove(blade_handle_t *bh, char *args)
+{
+	cJSON *channels = NULL;
+
+	ks_assert(bh);
+	ks_assert(args);
+
+	if (!args[0]) {
+		ks_log(KS_LOG_INFO, "Requires channel argument");
+		return;
+	}
+
+
+	if (ks_hash_remove(g_test->channels, (void *)args)) {
+		channels = cJSON_CreateArray();
+		cJSON_AddItemToArray(channels, cJSON_CreateString(args));
+
+		blade_handle_rpcpublish(bh, BLADE_RPCPUBLISH_COMMAND_CHANNEL_REMOVE, "test", "mydomain.com", channels, test_publish_response_handler, (void *)g_test);
+
+		cJSON_Delete(channels);
+	}
+}
+
 /* For Emacs:
 * Local Variables:
 * mode:c

libs/libks/src/include/ks_pool.h

@@ -151,7 +151,7 @@ KS_DECLARE(ks_status_t) ks_pool_clear(ks_pool_t *pool);
 KS_DECLARE(ks_bool_t) ks_pool_verify(void *addr);
 
 // @todo fill in documentation
-KS_DECLARE(ks_pool_t *) ks_pool_get(void *addr);
+inline KS_DECLARE(ks_pool_t *) ks_pool_get(void *addr);
 
 /*
  * void *ks_pool_alloc

libs/libks/src/ks_pool.c

@@ -596,31 +596,26 @@ done:
 // @todo fill in documentation
 KS_DECLARE(ks_bool_t) ks_pool_verify(void *addr)
 {
-	ks_pool_prefix_t *prefix = NULL;
 	if (!addr) return KS_FALSE;
-	prefix = (ks_pool_prefix_t *)((uintptr_t)addr - KS_POOL_PREFIX_SIZE);
-	if (check_prefix(prefix) != KS_STATUS_SUCCESS) return KS_FALSE;
+	if (check_prefix((ks_pool_prefix_t *)((uintptr_t)addr - KS_POOL_PREFIX_SIZE)) != KS_STATUS_SUCCESS) return KS_FALSE;
 	return KS_TRUE;
 }
 
 // @todo fill in documentation
-KS_DECLARE(ks_pool_t *) ks_pool_get(void *addr)
+inline KS_DECLARE(ks_pool_t *) ks_pool_get(void *addr)
 {
-	ks_pool_prefix_t *prefix = NULL;
+	ks_assert(addr);
+#ifdef DEBUG
+	ks_pool_prefix_t *prefix = (ks_pool_prefix_t *)((uintptr_t)addr - KS_POOL_PREFIX_SIZE);
 	ks_status_t ret = KS_STATUS_SUCCESS;
-	ks_pool_t *pool = NULL;
-
-	if (!addr) goto done;
 
-	prefix = (ks_pool_prefix_t *)((uintptr_t)addr - KS_POOL_PREFIX_SIZE);
-	if (check_prefix(prefix) != KS_STATUS_SUCCESS) goto done;
-
-	if ((ret = check_pool(prefix->pool)) == KS_STATUS_SUCCESS) pool = prefix->pool;
-
-done:
+	ret = check_prefix(prefix);
 	ks_assert(ret == KS_STATUS_SUCCESS);
 
-	return pool;
+	ret = check_pool(prefix->pool);
+	ks_assert(ret == KS_STATUS_SUCCESS);
+#endif
+	return ((ks_pool_prefix_t *)((uintptr_t)addr - KS_POOL_PREFIX_SIZE))->pool;
 }
 
 /*