Skip to content

F
freeswitch
提交 c6e60de3 authored 作者: Shane Bryldt's avatar Shane Bryldt
浏览文件
操作

FS-10167: Temporary commit for peer review

上级 f1ae0b38
全部展开 隐藏空白字符变更
内嵌 并排
正在显示 包含 1066 行增加 和 403 行删除
libs/libblade/src/blade_mastermgr.c
浏览文件 @ c6e60de3
......@@ -38,8 +38,7 @@ struct blade_mastermgr_s {
ks_pool_t *pool;
// @todo how does "exclusive" play into the controllers, does "exclusive" mean only one provider can exist for a given protocol and realm? what does non exclusive mean?
ks_hash_t *protocols; // protocols that have been published with blade.publish, and the details to locate a protocol provider with blade.locate
ks_hash_t *protocols_cleanup; // keyed by the nodeid, each value is a hash_t* of which contains string keys matching the "protocol@realm" keys to remove each nodeid from as a provider during cleanup
ks_hash_t *protocols; // protocols that have been published with blade.publish, and the details to locate a protocol controller with blade.locate
};
......@@ -76,9 +75,6 @@ KS_DECLARE(ks_status_t) blade_mastermgr_create(blade_mastermgr_t **bmmgrP, blade
ks_hash_create(&bmmgr->protocols, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY | KS_HASH_FLAG_FREE_VALUE, bmmgr->pool);
ks_assert(bmmgr->protocols);
ks_hash_create(&bmmgr->protocols_cleanup, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY | KS_HASH_FLAG_FREE_VALUE, bmmgr->pool);
ks_assert(bmmgr->protocols_cleanup);
ks_pool_set_cleanup(pool, bmmgr, NULL, blade_mastermgr_cleanup);
*bmmgrP = bmmgr;
......@@ -113,6 +109,40 @@ KS_DECLARE(blade_handle_t *) blade_mastermgr_handle_get(blade_mastermgr_t *bmmgr
return bmmgr->handle;
}
KS_DECLARE(ks_status_t) blade_mastermgr_purge(blade_mastermgr_t *bmmgr, const char *nodeid)
{
ks_hash_t *cleanup = NULL;
ks_hash_write_lock(bmmgr->protocols);
for (ks_hash_iterator_t *it = ks_hash_first(bmmgr->protocols, KS_UNLOCKED); it; it = ks_hash_next(&it)) {
const char *key = NULL;
blade_protocol_t *bp = NULL;
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, bmmgr->pool);
ks_hash_insert(cleanup, key, bp);
}
}
if (cleanup) {
for (ks_hash_iterator_t *it = ks_hash_first(cleanup, KS_UNLOCKED); it; it = ks_hash_next(&it)) {
const char *key = NULL;
blade_protocol_t *bp = NULL;
ks_hash_this(it, (const void **)&key, NULL, (void **)&bp);
ks_log(KS_LOG_DEBUG, "Protocol Removed: %s\n", key);
ks_hash_remove(bmmgr->protocols, key);
}
ks_hash_destroy(&cleanup);
}
ks_hash_write_unlock(bmmgr->protocols);
return KS_STATUS_SUCCESS;
}
KS_DECLARE(blade_protocol_t *) blade_mastermgr_protocol_lookup(blade_mastermgr_t *bmmgr, const char *protocol, const char *realm)
{
blade_protocol_t *bp = NULL;
......@@ -135,7 +165,6 @@ KS_DECLARE(ks_status_t) blade_mastermgr_controller_add(blade_mastermgr_t *bmmgr,
{
blade_protocol_t *bp = NULL;
char *key = NULL;
ks_hash_t *cleanup = NULL;
ks_assert(bmmgr);
ks_assert(protocol);
......@@ -159,60 +188,131 @@ KS_DECLARE(ks_status_t) blade_mastermgr_controller_add(blade_mastermgr_t *bmmgr,
ks_hash_insert(bmmgr->protocols, (void *)ks_pstrdup(bmmgr->pool, key), bp);
}
cleanup = (ks_hash_t *)ks_hash_search(bmmgr->protocols_cleanup, (void *)controller, KS_UNLOCKED);
if (!cleanup) {
ks_hash_create(&cleanup, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, bmmgr->pool);
ks_assert(cleanup);
ks_hash_insert(bmmgr->protocols_cleanup, (void *)ks_pstrdup(bmmgr->pool, controller), cleanup);
}
ks_hash_insert(cleanup, (void *)ks_pstrdup(bmmgr->pool, key), (void *)KS_TRUE);
blade_protocol_controllers_add(bp, controller);
ks_log(KS_LOG_DEBUG, "Protocol Controller Added: %s to %s\n", controller, key);
ks_pool_free(bmmgr->pool, &key);
ks_hash_write_unlock(bmmgr->protocols);
return KS_STATUS_SUCCESS;
}
KS_DECLARE(ks_status_t) blade_mastermgr_controller_remove(blade_mastermgr_t *bmmgr, 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_hash_t *cleanup = NULL;
ks_status_t ret = KS_STATUS_SUCCESS;
blade_protocol_t *bp = NULL;
char *key = NULL;
ks_assert(bmmgr);
ks_assert(controller);
ks_assert(protocol);
ks_assert(realm);
ks_assert(channel);
ks_hash_write_lock(bmmgr->protocols);
cleanup = (ks_hash_t *)ks_hash_search(bmmgr->protocols_cleanup, (void *)controller, KS_UNLOCKED);
if (cleanup) {
for (ks_hash_iterator_t *it = ks_hash_first(cleanup, KS_UNLOCKED); it; it = ks_hash_next(&it)) {
void *key = NULL;
void *value = NULL;
blade_protocol_t *bp = NULL;
ks_hash_t *controllers = NULL;
key = ks_psprintf(bmmgr->pool, "%s@%s", protocol, realm);
ks_hash_this(it, (const void **)&key, NULL, &value);
bp = (blade_protocol_t *)ks_hash_search(bmmgr->protocols, (void *)key, KS_READLOCKED);
if (!bp) {
ret = KS_STATUS_NOT_FOUND;
goto done;
}
bp = (blade_protocol_t *)ks_hash_search(bmmgr->protocols, key, KS_UNLOCKED);
ks_assert(bp); // should not happen when a cleanup still has a provider tracked for a protocol
blade_protocol_channel_add(bp, channel);
ks_log(KS_LOG_DEBUG, "Protocol Controller Removed: %s from %s\n", controller, key);
blade_protocol_controllers_remove(bp, controller);
done:
ks_pool_free(bmmgr->pool, &key);
controllers = blade_protocol_controllers_get(bp);
if (ks_hash_count(controllers) == 0) {
// @note this depends on locking something outside of the protocol that won't be destroyed, like the top level
// protocols hash, but assumes then that any reader keeps the top level hash read locked while using the protocol
// so it cannot be deleted
ks_log(KS_LOG_DEBUG, "Protocol Removed: %s\n", key);
ks_hash_remove(bmmgr->protocols, key);
}
}
ks_hash_remove(bmmgr->protocols_cleanup, (void *)controller);
ks_hash_read_unlock(bmmgr->protocols);
return ret;
}
KS_DECLARE(ks_status_t) blade_mastermgr_channel_remove(blade_mastermgr_t *bmmgr, const char *protocol, const char *realm, const char *channel)
{
ks_status_t ret = KS_STATUS_SUCCESS;
blade_protocol_t *bp = NULL;
char *key = NULL;
ks_assert(bmmgr);
ks_assert(protocol);
ks_assert(realm);
ks_assert(channel);
key = ks_psprintf(bmmgr->pool, "%s@%s", protocol, realm);
bp = (blade_protocol_t *)ks_hash_search(bmmgr->protocols, (void *)key, KS_READLOCKED);
if (!bp) {
ret = KS_STATUS_NOT_FOUND;
goto done;
}
ks_hash_write_unlock(bmmgr->protocols);
return KS_STATUS_SUCCESS;
blade_protocol_channel_remove(bp, channel);
done:
ks_pool_free(bmmgr->pool, &key);
ks_hash_read_unlock(bmmgr->protocols);
return ret;
}
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)
{
ks_status_t ret = KS_STATUS_SUCCESS;
blade_protocol_t *bp = NULL;
char *key = NULL;
//ks_hash_t *cleanup = NULL;
ks_assert(bmmgr);
ks_assert(protocol);
ks_assert(realm);
ks_assert(channel);
ks_assert(controller);
ks_assert(target);
key = ks_psprintf(bmmgr->pool, "%s@%s", protocol, realm);
bp = (blade_protocol_t *)ks_hash_search(bmmgr->protocols, (void *)key, KS_READLOCKED);
if (!bp) {
ret = KS_STATUS_NOT_FOUND;
goto done;
}
ret = blade_protocol_channel_authorize(bp, remove, channel, controller, target);
done:
ks_pool_free(bmmgr->pool, &key);
ks_hash_read_unlock(bmmgr->protocols);
return ret;
}
KS_DECLARE(ks_bool_t) blade_mastermgr_channel_verify(blade_mastermgr_t *bmmgr, const char *protocol, const char *realm, const char *channel, const char *target)
{
ks_bool_t ret = KS_FALSE;
blade_protocol_t *bp = NULL;
char *key = NULL;
//ks_hash_t *cleanup = NULL;
ks_assert(bmmgr);
ks_assert(protocol);
ks_assert(realm);
ks_assert(channel);
ks_assert(target);
key = ks_psprintf(bmmgr->pool, "%s@%s", protocol, realm);
bp = (blade_protocol_t *)ks_hash_search(bmmgr->protocols, (void *)key, KS_READLOCKED);
if (!bp) goto done;
ret = blade_protocol_channel_verify(bp, channel, target);
done:
ks_pool_free(bmmgr->pool, &key);
ks_hash_read_unlock(bmmgr->protocols);
return ret;
}
/* For Emacs:
......
libs/libblade/src/blade_protocol.c
浏览文件 @ c6e60de3
......@@ -39,6 +39,7 @@ struct blade_protocol_s {
const char *name;
const char *realm;
ks_hash_t *controllers;
ks_hash_t *channels;
// @todo descriptors (schema, etc) for each method within a protocol
};
......@@ -56,6 +57,7 @@ static void blade_protocol_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_poo
if (bp->name) ks_pool_free(bp->pool, &bp->name);
if (bp->realm) ks_pool_free(bp->pool, &bp->realm);
if (bp->controllers) ks_hash_destroy(&bp->controllers);
if (bp->channels) ks_hash_destroy(&bp->channels);
break;
case KS_MPCL_DESTROY:
break;
......@@ -76,9 +78,12 @@ KS_DECLARE(ks_status_t) blade_protocol_create(blade_protocol_t **bpP, ks_pool_t
bp->name = ks_pstrdup(pool, name);
bp->realm = ks_pstrdup(pool, realm);
ks_hash_create(&bp->controllers, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_NOLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, bp->pool);
ks_hash_create(&bp->controllers, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, bp->pool);
ks_assert(bp->controllers);
ks_hash_create(&bp->channels, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, bp->pool);
ks_assert(bp->channels);
ks_pool_set_cleanup(pool, bp, NULL, blade_protocol_cleanup);
*bpP = bp;
......@@ -100,11 +105,53 @@ KS_DECLARE(ks_status_t) blade_protocol_destroy(blade_protocol_t **bpP)
return KS_STATUS_SUCCESS;
}
KS_DECLARE(ks_hash_t *) blade_protocol_controllers_get(blade_protocol_t *bp)
KS_DECLARE(ks_bool_t) blade_protocol_purge(blade_protocol_t *bp, const char *nodeid)
{
ks_bool_t ret = KS_FALSE;
ks_assert(bp);
ks_assert(nodeid);
// @todo iterate all channels, remove the nodeid from all authorized hashes
ks_hash_write_lock(bp->channels);
for (ks_hash_iterator_t *it = ks_hash_first(bp->channels, KS_UNLOCKED); it; it = ks_hash_next(&it)) {
const char *key = NULL;
ks_hash_t *authorizations = NULL;
ks_hash_this(it, (const void **)&key, NULL, (void **)&authorizations);
if (ks_hash_remove(authorizations, nodeid)) {
ks_log(KS_LOG_DEBUG, "Protocol Channel Authorization Removed: %s from %s@%s/%s\n", nodeid, bp->name, bp->realm, key);
}
}
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;
}
KS_DECLARE(cJSON *) blade_protocol_controllers_pack(blade_protocol_t *bp)
{
cJSON *controllers = cJSON_CreateObject();
ks_assert(bp);
return 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;
ks_hash_this(it, (const void **)&key, NULL, &value);
cJSON_AddItemToArray(controllers, cJSON_CreateString(key));
}
return controllers;
}
KS_DECLARE(ks_status_t) blade_protocol_controllers_add(blade_protocol_t *bp, const char *nodeid)
......@@ -117,19 +164,108 @@ KS_DECLARE(ks_status_t) blade_protocol_controllers_add(blade_protocol_t *bp, con
key = ks_pstrdup(bp->pool, nodeid);
ks_hash_insert(bp->controllers, (void *)key, (void *)KS_TRUE);
ks_log(KS_LOG_DEBUG, "Protocol Controller Added: %s to %s@%s\n", nodeid, bp->name, bp->realm);
return KS_STATUS_SUCCESS;
}
KS_DECLARE(ks_status_t) blade_protocol_channel_add(blade_protocol_t *bp, const char *name)
{
ks_status_t ret = KS_STATUS_SUCCESS;
ks_hash_t *authorized = NULL;
char *key = NULL;
ks_assert(bp);
ks_assert(name);
ks_hash_write_lock(bp->channels);
if (ks_hash_search(bp->channels, name, KS_UNLOCKED)) {
ret = KS_STATUS_DUPLICATE_OPERATION;
goto done;
}
ks_hash_create(&authorized, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, bp->pool);
key = ks_pstrdup(bp->pool, name);
ks_hash_insert(bp->channels, (void *)key, (void *)authorized);
ks_log(KS_LOG_DEBUG, "Protocol Channel Added: %s to %s@%s\n", key, bp->name, bp->realm);
done:
ks_hash_write_unlock(bp->channels);
return ret;
}
KS_DECLARE(ks_status_t) blade_protocol_controllers_remove(blade_protocol_t *bp, const char *nodeid)
KS_DECLARE(ks_status_t) blade_protocol_channel_remove(blade_protocol_t *bp, const char *name)
{
ks_assert(bp);
ks_assert(nodeid);
ks_assert(name);
ks_hash_remove(bp->controllers, (void *)nodeid);
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);
return KS_STATUS_SUCCESS;
}
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_status_t ret = KS_STATUS_SUCCESS;
ks_hash_t *authorizations = NULL;
ks_bool_t allowed = KS_FALSE;
ks_assert(bp);
ks_assert(channel);
ks_assert(controller);
ks_assert(target);
allowed = (ks_bool_t)(intptr_t)ks_hash_search(bp->controllers, (void *)controller, KS_READLOCKED);
ks_hash_read_unlock(bp->controllers);
if (!allowed) {
ret = KS_STATUS_NOT_ALLOWED;
goto done;
}
// @todo verify controller, get ks_hash_t* value based on channel, add target to the channels hash
authorizations = (ks_hash_t *)ks_hash_search(bp->channels, (void *)channel, KS_READLOCKED);
if (authorizations) {
if (remove) {
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 {
ks_hash_insert(authorizations, (void *)ks_pstrdup(bp->pool, 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);
}
}
ks_hash_read_unlock(bp->channels);
if (!authorizations) ret = KS_STATUS_NOT_FOUND;
done:
return ret;
}
KS_DECLARE(ks_bool_t) blade_protocol_channel_verify(blade_protocol_t *bp, const char *channel, const char *target)
{
ks_bool_t ret = KS_FALSE;
ks_hash_t *authorizations = NULL;
ks_assert(bp);
ks_assert(channel);
ks_assert(target);
// @todo verify controller, get ks_hash_t* value based on channel, add target to the channels hash
authorizations = (ks_hash_t *)ks_hash_search(bp->channels, (void *)channel, KS_READLOCKED);
if (authorizations) ret = ks_hash_search(authorizations, target, KS_UNLOCKED) != NULL;
ks_hash_read_unlock(bp->channels);
return ret;
}
/* For Emacs:
......
libs/libblade/src/blade_routemgr.c
浏览文件 @ c6e60de3
......@@ -157,12 +157,14 @@ KS_DECLARE(ks_status_t) blade_routemgr_route_remove(blade_routemgr_t *brmgr, con
blade_handle_rpcregister(brmgr->handle, target, KS_TRUE, NULL, NULL);
blade_subscriptionmgr_purge(blade_handle_subscriptionmgr_get(brmgr->handle), target);
// @note protocols are cleaned up here because routes can be removed that are not locally connected with a session but still
// have protocols published to the master node from further downstream, in which case if a route is announced upstream to be
// removed, a master node is still able to catch that here even when there is no direct session, but is also hit when there
// is a direct session being terminated
blade_mastermgr_controller_remove(blade_handle_mastermgr_get(brmgr->handle), target);
blade_mastermgr_purge(blade_handle_mastermgr_get(brmgr->handle), target);
return KS_STATUS_SUCCESS;
}
......
libs/libblade/src/blade_rpc.c
浏览文件 @ c6e60de3
......@@ -42,7 +42,7 @@ struct blade_rpc_s {
const char *realm;
blade_rpc_request_callback_t callback;
void *data;
cJSON *data;
};
struct blade_rpc_request_s {
......@@ -54,7 +54,7 @@ struct blade_rpc_request_s {
cJSON *message;
const char *message_id; // pulled from message for easier keying
blade_rpc_response_callback_t callback;
void *data;
cJSON *data;
// @todo ttl to wait for response before injecting an error response locally
};
......@@ -72,21 +72,22 @@ struct blade_rpc_response_s {
static void blade_rpc_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_pool_cleanup_action_t action, ks_pool_cleanup_type_t type)
{
//blade_rpc_t *brpc = (blade_rpc_t *)ptr;
blade_rpc_t *brpc = (blade_rpc_t *)ptr;
//ks_assert(brpc);
ks_assert(brpc);
switch (action) {
case KS_MPCL_ANNOUNCE:
break;
case KS_MPCL_TEARDOWN:
if (brpc->data) cJSON_Delete(brpc->data);
break;
case KS_MPCL_DESTROY:
break;
}
}
KS_DECLARE(ks_status_t) blade_rpc_create(blade_rpc_t **brpcP, blade_handle_t *bh, const char *method, const char *protocol, const char *realm, blade_rpc_request_callback_t callback, void *data)
KS_DECLARE(ks_status_t) blade_rpc_create(blade_rpc_t **brpcP, blade_handle_t *bh, const char *method, const char *protocol, const char *realm, blade_rpc_request_callback_t callback, cJSON *data)
{
blade_rpc_t *brpc = NULL;
ks_pool_t *pool = NULL;
......@@ -169,7 +170,7 @@ KS_DECLARE(blade_rpc_request_callback_t) blade_rpc_callback_get(blade_rpc_t *brp
return brpc->callback;
}
KS_DECLARE(void *) blade_rpc_data_get(blade_rpc_t *brpc)
KS_DECLARE(cJSON *) blade_rpc_data_get(blade_rpc_t *brpc)
{
ks_assert(brpc);
......@@ -189,6 +190,7 @@ static void blade_rpc_request_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks_
case KS_MPCL_TEARDOWN:
ks_pool_free(brpcreq->pool, (void **)&brpcreq->session_id);
cJSON_Delete(brpcreq->message);
if (brpcreq->data) cJSON_Delete(brpcreq->data);
break;
case KS_MPCL_DESTROY:
break;
......@@ -201,7 +203,7 @@ KS_DECLARE(ks_status_t) blade_rpc_request_create(blade_rpc_request_t **brpcreqP,
const char *session_id,
cJSON *json,
blade_rpc_response_callback_t callback,
void *data)
cJSON *data)
{
blade_rpc_request_t *brpcreq = NULL;
......@@ -276,7 +278,7 @@ KS_DECLARE(blade_rpc_response_callback_t) blade_rpc_request_callback_get(blade_r
return brpcreq->callback;
}
KS_DECLARE(void *) blade_rpc_request_data_get(blade_rpc_request_t *brpcreq)
KS_DECLARE(cJSON *) blade_rpc_request_data_get(blade_rpc_request_t *brpcreq)
{
ks_assert(brpcreq);
return brpcreq->data;
......
libs/libblade/src/blade_session.c
浏览文件 @ c6e60de3
......@@ -582,7 +582,7 @@ ks_status_t blade_session_onstate_run(blade_session_t *bs)
}
KS_DECLARE(ks_status_t) blade_session_send(blade_session_t *bs, cJSON *json, blade_rpc_response_callback_t callback, void *data)
KS_DECLARE(ks_status_t) blade_session_send(blade_session_t *bs, cJSON *json, blade_rpc_response_callback_t callback, cJSON *data)
{
blade_rpc_request_t *brpcreq = NULL;
const char *method = NULL;
......@@ -706,6 +706,12 @@ ks_status_t blade_session_process(blade_session_t *bs, cJSON *json)
blade_session_send(bs_router, json, NULL, NULL);
blade_session_read_unlock(bs_router);
// @todo if this is a subscribe request to remove subscriptions, it must carry a field unsubscribed-channels for which
// subscriptions should be removed along the request path, regardless of whether it's the consumer requesting or the
// master due to a deauthorization, without respect to waiting for the response as it should be a gaurenteed operation
// even when requested by the subscriber. This unsubscribed-channels field is simply treated as a special field, regardless
// of the actual method of the request.
return KS_STATUS_SUCCESS;
}
}
......@@ -763,6 +769,11 @@ ks_status_t blade_session_process(blade_session_t *bs, cJSON *json)
blade_session_send(bs_router, json, NULL, NULL);
blade_session_read_unlock(bs_router);
// @todo if this is a subscribe response to add a subscriber with the master as the responder-nodeid, it must have a
// subscribed-channels field which should be added for the requester-nodeid, this will ensure the subscriptions are
// added with respect to the master response and not immediately upon request. This subscribed-channels field is
// simply treated as a special field, regardless of the actual method of the request which is unavailable here.
return KS_STATUS_SUCCESS;
}
}
......
libs/libblade/src/blade_sessionmgr.c
浏览文件 @ c6e60de3
......@@ -200,8 +200,6 @@ KS_DECLARE(ks_status_t) blade_sessionmgr_session_remove(blade_sessionmgr_t *bsmg
ks_log(KS_LOG_DEBUG, "Session Removed: %s\n", id);
blade_subscriptionmgr_subscriber_cleanup(blade_handle_subscriptionmgr_get(bsmgr->handle), id);
if (blade_upstreammgr_localid_compare(blade_handle_upstreammgr_get(bsmgr->handle), id)) {
blade_upstreammgr_localid_set(blade_handle_upstreammgr_get(bsmgr->handle), NULL);
blade_upstreammgr_masterid_set(blade_handle_upstreammgr_get(bsmgr->handle), NULL);
......
libs/libblade/src/blade_subscription.c
浏览文件 @ c6e60de3
......@@ -36,9 +36,9 @@
struct blade_subscription_s {
ks_pool_t *pool;
const char *event;
const char *protocol;
const char *realm;
const char *channel;
ks_hash_t *subscribers;
blade_rpc_request_callback_t callback;
......@@ -56,9 +56,9 @@ static void blade_subscription_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks
case KS_MPCL_ANNOUNCE:
break;
case KS_MPCL_TEARDOWN:
if (bsub->event) ks_pool_free(bsub->pool, &bsub->event);
if (bsub->protocol) ks_pool_free(bsub->pool, &bsub->protocol);
if (bsub->realm) ks_pool_free(bsub->pool, &bsub->subscribers);
if (bsub->channel) ks_pool_free(bsub->pool, &bsub->channel);
if (bsub->subscribers) ks_hash_destroy(&bsub->subscribers);
break;
case KS_MPCL_DESTROY:
......@@ -66,21 +66,21 @@ static void blade_subscription_cleanup(ks_pool_t *pool, void *ptr, void *arg, ks
}
}
KS_DECLARE(ks_status_t) blade_subscription_create(blade_subscription_t **bsubP, ks_pool_t *pool, const char *event, const char *protocol, const char *realm)
KS_DECLARE(ks_status_t) blade_subscription_create(blade_subscription_t **bsubP, ks_pool_t *pool, const char *protocol, const char *realm, const char *channel)
{
blade_subscription_t *bsub = NULL;
ks_assert(bsubP);
ks_assert(pool);
ks_assert(event);
ks_assert(protocol);
ks_assert(realm);
ks_assert(channel);
bsub = ks_pool_alloc(pool, sizeof(blade_subscription_t));
bsub->pool = pool;
bsub->event = ks_pstrdup(pool, event);
bsub->protocol = ks_pstrdup(pool, protocol);
bsub->realm = ks_pstrdup(pool, realm);
bsub->channel = ks_pstrdup(pool, channel);
ks_hash_create(&bsub->subscribers, KS_HASH_MODE_CASE_INSENSITIVE, KS_HASH_FLAG_NOLOCK | KS_HASH_FLAG_DUP_CHECK | KS_HASH_FLAG_FREE_KEY, bsub->pool);
ks_assert(bsub->subscribers);
......@@ -106,27 +106,27 @@ KS_DECLARE(ks_status_t) blade_subscription_destroy(blade_subscription_t **bsubP)
return KS_STATUS_SUCCESS;
}
KS_DECLARE(const char *) blade_subscription_event_get(blade_subscription_t *bsub)
KS_DECLARE(const char *) blade_subscription_protocol_get(blade_subscription_t *bsub)
{
ks_assert(bsub);
return bsub->event;
return bsub->protocol;
}
KS_DECLARE(const char *) blade_subscription_protocol_get(blade_subscription_t *bsub)
KS_DECLARE(const char *) blade_subscription_realm_get(blade_subscription_t *bsub)
{
ks_assert(bsub);
return bsub->protocol;
return bsub->realm;
}
KS_DECLARE(const char *) blade_subscription_realm_get(blade_subscription_t *bsub)
KS_DECLARE(const char *) blade_subscription_channel_get(blade_subscription_t *bsub)
{
ks_assert(bsub);
return bsub->realm;
return bsub->channel;
}
......
libs/libblade/src/blade_upstreammgr.c
浏览文件 @ c6e60de3
......@@ -250,6 +250,21 @@ KS_DECLARE(ks_status_t) blade_upstreammgr_masterid_copy(blade_upstreammgr_t *bum
return KS_STATUS_SUCCESS;
}
KS_DECLARE(ks_bool_t) blade_upstreammgr_masterlocal(blade_upstreammgr_t *bumgr)
{
ks_bool_t ret = KS_FALSE;
ks_assert(bumgr);
ks_rwl_read_lock(bumgr->masterid_rwl);
ks_rwl_read_lock(bumgr->localid_rwl);
ret = bumgr->masterid && bumgr->localid && !ks_safe_strcasecmp(bumgr->masterid, bumgr->localid);
ks_rwl_read_unlock(bumgr->localid_rwl);
ks_rwl_read_unlock(bumgr->masterid_rwl);
return ret;
}
KS_DECLARE(ks_status_t) blade_upstreammgr_realm_add(blade_upstreammgr_t *bumgr, const char *realm)
{
char *key = NULL;
......
libs/libblade/src/include/blade_mastermgr.h
浏览文件 @ c6e60de3
......@@ -39,9 +39,13 @@ KS_BEGIN_EXTERN_C
KS_DECLARE(ks_status_t) blade_mastermgr_create(blade_mastermgr_t **bmmgrP, blade_handle_t *bh);
KS_DECLARE(ks_status_t) blade_mastermgr_destroy(blade_mastermgr_t **bmmgrP);
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 *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);
KS_DECLARE(ks_bool_t) blade_mastermgr_channel_verify(blade_mastermgr_t *bmmgr, const char *protocol, const char *realm, const char *channel, const char *target);
KS_END_EXTERN_C
#endif
......
libs/libblade/src/include/blade_protocol.h
浏览文件 @ c6e60de3
......@@ -38,9 +38,13 @@
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(ks_hash_t *) blade_protocol_controllers_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(ks_status_t) blade_protocol_controllers_remove(blade_protocol_t *bp, const char *nodeid);
KS_DECLARE(cJSON *) blade_protocol_controllers_pack(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_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
#endif
......
libs/libblade/src/include/blade_rpc.h
浏览文件 @ c6e60de3
......@@ -36,14 +36,14 @@
#include <blade.h>
KS_BEGIN_EXTERN_C
KS_DECLARE(ks_status_t) blade_rpc_create(blade_rpc_t **brpcP, blade_handle_t *bh, const char *method, const char *protocol, const char *realm, blade_rpc_request_callback_t callback, void *data);
KS_DECLARE(ks_status_t) blade_rpc_create(blade_rpc_t **brpcP, blade_handle_t *bh, const char *method, const char *protocol, const char *realm, blade_rpc_request_callback_t callback, cJSON *data);
KS_DECLARE(ks_status_t) blade_rpc_destroy(blade_rpc_t **brpcP);
KS_DECLARE(blade_handle_t *) blade_rpc_handle_get(blade_rpc_t *brpc);
KS_DECLARE(const char *) blade_rpc_method_get(blade_rpc_t *brpc);
KS_DECLARE(const char *) blade_rpc_protocol_get(blade_rpc_t *brpc);
KS_DECLARE(const char *) blade_rpc_realm_get(blade_rpc_t *brpc);
KS_DECLARE(blade_rpc_request_callback_t) blade_rpc_callback_get(blade_rpc_t *brpc);
KS_DECLARE(void *) blade_rpc_data_get(blade_rpc_t *brpc);
KS_DECLARE(cJSON *) blade_rpc_data_get(blade_rpc_t *brpc);
KS_DECLARE(ks_status_t) blade_rpc_request_create(blade_rpc_request_t **brpcreqP,
blade_handle_t *bh,
......@@ -51,14 +51,15 @@ KS_DECLARE(ks_status_t) blade_rpc_request_create(blade_rpc_request_t **brpcreqP,
const char *session_id,
cJSON *json,
blade_rpc_response_callback_t callback,
void *data);
cJSON *data);
KS_DECLARE(ks_status_t) blade_rpc_request_destroy(blade_rpc_request_t **brpcreqP);
KS_DECLARE(ks_status_t) blade_rpc_request_duplicate(blade_rpc_request_t **brpcreqP, blade_rpc_request_t *brpcreq);
KS_DECLARE(blade_handle_t *) blade_rpc_request_handle_get(blade_rpc_request_t *brpcreq);
KS_DECLARE(const char *) blade_rpc_request_sessionid_get(blade_rpc_request_t *brpcreq);
KS_DECLARE(cJSON *) blade_rpc_request_message_get(blade_rpc_request_t *brpcreq);
KS_DECLARE(const char *) blade_rpc_request_messageid_get(blade_rpc_request_t *brpcreq);
KS_DECLARE(blade_rpc_response_callback_t) blade_rpc_request_callback_get(blade_rpc_request_t *brpcreq);
KS_DECLARE(void *) blade_rpc_request_data_get(blade_rpc_request_t *brpcreq);
KS_DECLARE(cJSON *) blade_rpc_request_data_get(blade_rpc_request_t *brpcreq);
KS_DECLARE(ks_status_t) blade_rpc_request_raw_create(ks_pool_t *pool, cJSON **json, cJSON **params, const char **id, const char *method);
......
libs/libblade/src/include/blade_session.h
浏览文件 @ c6e60de3
......@@ -62,7 +62,7 @@ KS_DECLARE(void) blade_session_hangup(blade_session_t *bs);
KS_DECLARE(ks_bool_t) blade_session_terminating(blade_session_t *bs);
KS_DECLARE(const char *) blade_session_connection_get(blade_session_t *bs);
KS_DECLARE(ks_status_t) blade_session_connection_set(blade_session_t *bs, const char *id);
KS_DECLARE(ks_status_t) blade_session_send(blade_session_t *bs, cJSON *json, blade_rpc_response_callback_t callback, void *data);
KS_DECLARE(ks_status_t) blade_session_send(blade_session_t *bs, cJSON *json, blade_rpc_response_callback_t callback, cJSON *data);
KS_DECLARE(ks_status_t) blade_session_sending_push(blade_session_t *bs, cJSON *json);
KS_DECLARE(ks_status_t) blade_session_sending_pop(blade_session_t *bs, cJSON **json);
KS_DECLARE(ks_status_t) blade_session_receiving_push(blade_session_t *bs, cJSON *json);
......
libs/libblade/src/include/blade_stack.h
浏览文件 @ c6e60de3
......@@ -59,24 +59,25 @@ KS_DECLARE(blade_sessionmgr_t *) blade_handle_sessionmgr_get(blade_handle_t *bh)
KS_DECLARE(ks_status_t) blade_handle_connect(blade_handle_t *bh, blade_connection_t **bcP, blade_identity_t *target, const char *session_id);
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_rpcregister(blade_handle_t *bh, const char *nodeid, ks_bool_t remove, blade_rpc_response_callback_t callback, cJSON *data);
KS_DECLARE(ks_status_t) blade_handle_rpcpublish(blade_handle_t *bh, const char *name, const char *realm, 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, cJSON *data);
KS_DECLARE(ks_status_t) blade_handle_rpclocate(blade_handle_t *bh, const char *name, const char *realm, 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, cJSON *data);
KS_DECLARE(ks_status_t) blade_handle_rpcexecute(blade_handle_t *bh, const char *nodeid, const char *method, const char *protocol, const char *realm, cJSON *params, blade_rpc_response_callback_t callback, void *data);
KS_DECLARE(ks_status_t) blade_handle_rpclocate(blade_handle_t *bh, const char *protocol, const char *realm, blade_rpc_response_callback_t callback, cJSON *data);
KS_DECLARE(ks_status_t) blade_handle_rpcexecute(blade_handle_t *bh, const char *nodeid, const char *method, const char *protocol, const char *realm, cJSON *params, blade_rpc_response_callback_t callback, cJSON *data);
KS_DECLARE(const char *) blade_rpcexecute_request_requester_nodeid_get(blade_rpc_request_t *brpcreq);
KS_DECLARE(const char *) blade_rpcexecute_request_responder_nodeid_get(blade_rpc_request_t *brpcreq);
KS_DECLARE(cJSON *) blade_rpcexecute_request_params_get(blade_rpc_request_t *brpcreq);
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 *event, const char *protocol, const char *realm, ks_bool_t remove, blade_rpc_response_callback_t callback, void *data, blade_rpc_request_callback_t event_callback, void *event_data);
KS_DECLARE(ks_status_t) blade_handle_rpcsubscribe_raw(blade_handle_t *bh, const char *event, const char *protocol, const char *realm, ks_bool_t remove, blade_rpc_response_callback_t callback, void *data);
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, cJSON *data, blade_rpc_request_callback_t channel_callback, cJSON *channel_data);
KS_DECLARE(ks_status_t) blade_handle_rpcsubscribe_raw(blade_handle_t *bh, const char *protocol, const char *realm, cJSON *subscribe_channels, cJSON *unsubscribe_channels, const char *subscriber, ks_bool_t downstream, blade_rpc_response_callback_t callback, cJSON *data);
KS_DECLARE(ks_status_t) blade_handle_rpcbroadcast(blade_handle_t *bh, const char *broadcaster_nodeid, const char *event, const char *protocol, const char *realm, cJSON *params, blade_rpc_response_callback_t callback, void *data);
KS_DECLARE(const char *) blade_rpcbroadcast_request_broadcaster_nodeid_get(blade_rpc_request_t *brpcreq);
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, cJSON *data);
KS_DECLARE(cJSON *) blade_rpcbroadcast_request_params_get(blade_rpc_request_t *brpcreq);
KS_END_EXTERN_C
......
libs/libblade/src/include/blade_subscription.h
浏览文件 @ c6e60de3
......@@ -36,11 +36,11 @@
#include <blade.h>
KS_BEGIN_EXTERN_C
KS_DECLARE(ks_status_t) blade_subscription_create(blade_subscription_t **bsubP, ks_pool_t *pool, const char *event, const char *protocol, const char *realm);
KS_DECLARE(ks_status_t) blade_subscription_create(blade_subscription_t **bsubP, ks_pool_t *pool, const char *protocol, const char *realm, const char *channel);
KS_DECLARE(ks_status_t) blade_subscription_destroy(blade_subscription_t **bsubP);
KS_DECLARE(const char *) blade_subscription_event_get(blade_subscription_t *bsub);
KS_DECLARE(const char *) blade_subscription_protocol_get(blade_subscription_t *bsub);
KS_DECLARE(const char *) blade_subscription_realm_get(blade_subscription_t *bsub);
KS_DECLARE(const char *) blade_subscription_channel_get(blade_subscription_t *bsub);
KS_DECLARE(ks_hash_t *) blade_subscription_subscribers_get(blade_subscription_t *bsub);
KS_DECLARE(ks_status_t) blade_subscription_subscribers_add(blade_subscription_t *bsub, const char *nodeid);
KS_DECLARE(ks_status_t) blade_subscription_subscribers_remove(blade_subscription_t *bsub, const char *nodeid);
......
libs/libblade/src/include/blade_subscriptionmgr.h
浏览文件 @ c6e60de3
......@@ -39,11 +39,11 @@ KS_BEGIN_EXTERN_C
KS_DECLARE(ks_status_t) blade_subscriptionmgr_create(blade_subscriptionmgr_t **bsmgrP, blade_handle_t *bh);
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 *event, const char *protocol, const char *realm);
KS_DECLARE(ks_bool_t) blade_subscriptionmgr_subscriber_add(blade_subscriptionmgr_t *bsmgr, blade_subscription_t **bsubP, const char *event, const char *protocol, const char *realm, const char *target);
KS_DECLARE(ks_bool_t) blade_subscriptionmgr_subscriber_remove(blade_subscriptionmgr_t *bsmgr, blade_subscription_t **bsubP, const char *event, const char *protocol, const char *realm, const char *target);
KS_DECLARE(void) blade_subscriptionmgr_subscriber_cleanup(blade_subscriptionmgr_t *bsmgr, const char *target);
KS_DECLARE(ks_status_t) blade_subscriptionmgr_broadcast(blade_subscriptionmgr_t *bsmgr, const char *broadcaster_nodeid, const char *excluded_nodeid, const char *event, const char *protocol, const char *realm, cJSON *params, blade_rpc_response_callback_t callback, void *data);
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_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, cJSON *data);
KS_END_EXTERN_C
#endif
......
libs/libblade/src/include/blade_types.h
浏览文件 @ c6e60de3
......@@ -62,8 +62,8 @@ typedef struct blade_connectionmgr_s blade_connectionmgr_t;
typedef struct blade_sessionmgr_s blade_sessionmgr_t;
typedef struct blade_session_callback_data_s blade_session_callback_data_t;
typedef ks_bool_t (*blade_rpc_request_callback_t)(blade_rpc_request_t *brpcreq, void *data);
typedef ks_bool_t (*blade_rpc_response_callback_t)(blade_rpc_response_t *brpcres, void *data);
typedef ks_bool_t (*blade_rpc_request_callback_t)(blade_rpc_request_t *brpcreq, cJSON *data);
typedef ks_bool_t (*blade_rpc_response_callback_t)(blade_rpc_response_t *brpcres, cJSON *data);
typedef enum {
......
libs/libblade/src/include/blade_upstreammgr.h
浏览文件 @ c6e60de3
......@@ -47,6 +47,7 @@ KS_DECLARE(blade_session_t *) blade_upstreammgr_session_get(blade_upstreammgr_t
KS_DECLARE(ks_status_t) blade_upstreammgr_masterid_set(blade_upstreammgr_t *bumgr, const char *id);
KS_DECLARE(ks_bool_t) blade_upstreammgr_masterid_compare(blade_upstreammgr_t *bumgr, const char *id);
KS_DECLARE(ks_status_t) blade_upstreammgr_masterid_copy(blade_upstreammgr_t *bumgr, ks_pool_t *pool, const char **id);
KS_DECLARE(ks_bool_t) blade_upstreammgr_masterlocal(blade_upstreammgr_t *bumgr);
//KS_DECLARE(ks_hash_t *) blade_upstreammgr_realm_lookup(blade_handle_t *bh);
KS_DECLARE(ks_status_t) blade_upstreammgr_realm_add(blade_upstreammgr_t *bumgr, const char *realm);
KS_DECLARE(ks_status_t) blade_upstreammgr_realm_remove(blade_upstreammgr_t *bumgr, const char *realm);
......
libs/libblade/test/bladec.c
浏览文件 @ c6e60de3
......@@ -290,10 +290,15 @@ void command_execute(blade_handle_t *bh, char *args)
void command_subscribe(blade_handle_t *bh, char *args)
{
cJSON *channels = NULL;
ks_assert(bh);
ks_assert(args);
blade_handle_rpcsubscribe(bh, "test.event", "test", "mydomain.com", KS_FALSE, blade_subscribe_response_handler, NULL, test_event_request_handler, NULL);
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);
cJSON_Delete(channels);
}
/* For Emacs:
......
libs/libblade/test/blades.c
浏览文件 @ c6e60de3
......@@ -81,11 +81,12 @@ ks_bool_t test_echo_request_handler(blade_rpc_request_t *brpcreq, void *data)
cJSON_AddStringToObject(result, "text", text);
blade_rpcexecute_response_send(brpcreq, result);
cJSON_Delete(result);
return KS_FALSE;
}
ks_bool_t test_event_response_handler(blade_rpc_response_t *brpcres, void *data)
ks_bool_t test_broadcast_response_handler(blade_rpc_response_t *brpcres, void *data)
{
blade_handle_t *bh = NULL;
blade_session_t *bs = NULL;
......@@ -98,7 +99,7 @@ ks_bool_t test_event_response_handler(blade_rpc_response_t *brpcres, void *data)
bs = blade_sessionmgr_session_lookup(blade_handle_sessionmgr_get(bh), blade_rpc_response_sessionid_get(brpcres));
ks_assert(bs);
ks_log(KS_LOG_DEBUG, "Session (%s) test.event response processing\n", blade_session_id_get(bs));
ks_log(KS_LOG_DEBUG, "Session (%s) test broadcast response processing\n", blade_session_id_get(bs));
blade_session_read_unlock(bs);
......@@ -243,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", blade_publish_response_handler, NULL);
blade_handle_rpcpublish(bh, "test", "mydomain.com", NULL, blade_publish_response_handler, NULL);
}
void command_broadcast(blade_handle_t *bh, char *args)
......@@ -251,7 +252,7 @@ void command_broadcast(blade_handle_t *bh, char *args)
ks_assert(bh);
ks_assert(args);
blade_handle_rpcbroadcast(bh, NULL, "test.event", "test", "mydomain.com", NULL, test_event_response_handler, NULL);
blade_handle_rpcbroadcast(bh, "test", "mydomain.com", "channel", "event", NULL, test_broadcast_response_handler, NULL);
}
......
libs/libblade/test/testcli.c
浏览文件 @ c6e60de3
......@@ -33,7 +33,7 @@ static const struct command_def_s command_defs[] = {
const char *g_testcon_nodeid = NULL;
ks_bool_t test_locate_response_handler(blade_rpc_response_t *brpcres, void *data)
ks_bool_t test_locate_response_handler(blade_rpc_response_t *brpcres, cJSON *data)
{
blade_handle_t *bh = NULL;
blade_session_t *bs = NULL;
......@@ -87,7 +87,7 @@ ks_bool_t test_locate_response_handler(blade_rpc_response_t *brpcres, void *data
return KS_FALSE;
}
ks_bool_t test_join_response_handler(blade_rpc_response_t *brpcres, void *data)
ks_bool_t test_join_response_handler(blade_rpc_response_t *brpcres, cJSON *data)
{
blade_handle_t *bh = NULL;
blade_session_t *bs = NULL;
......@@ -111,7 +111,7 @@ ks_bool_t test_join_response_handler(blade_rpc_response_t *brpcres, void *data)
return KS_FALSE;
}
ks_bool_t test_leave_response_handler(blade_rpc_response_t *brpcres, void *data)
ks_bool_t test_leave_response_handler(blade_rpc_response_t *brpcres, cJSON *data)
{
blade_handle_t *bh = NULL;
blade_session_t *bs = NULL;
......@@ -135,7 +135,7 @@ ks_bool_t test_leave_response_handler(blade_rpc_response_t *brpcres, void *data)
return KS_FALSE;
}
ks_bool_t test_talk_response_handler(blade_rpc_response_t *brpcres, void *data)
ks_bool_t test_talk_response_handler(blade_rpc_response_t *brpcres, cJSON *data)
{
blade_handle_t *bh = NULL;
blade_session_t *bs = NULL;
......@@ -159,11 +159,10 @@ ks_bool_t test_talk_response_handler(blade_rpc_response_t *brpcres, void *data)
return KS_FALSE;
}
ks_bool_t test_join_broadcast_handler(blade_rpc_request_t *brpcreq, void *data)
ks_bool_t test_broadcast_handler(blade_rpc_request_t *brpcreq, cJSON *data)
{
blade_handle_t *bh = NULL;
blade_session_t *bs = NULL;
const char *broadcaster_nodeid = NULL;
cJSON *params = NULL;
//cJSON *result = NULL;
......@@ -178,70 +177,7 @@ ks_bool_t test_join_broadcast_handler(blade_rpc_request_t *brpcreq, void *data)
params = blade_rpcbroadcast_request_params_get(brpcreq);
ks_assert(params);
broadcaster_nodeid = blade_rpcbroadcast_request_broadcaster_nodeid_get(brpcreq);
ks_assert(broadcaster_nodeid);
ks_log(KS_LOG_DEBUG, "Session (%s) test.join (%s) broadcast processing\n", blade_session_id_get(bs), broadcaster_nodeid);
blade_session_read_unlock(bs);
return KS_FALSE;
}
ks_bool_t test_leave_broadcast_handler(blade_rpc_request_t *brpcreq, void *data)
{
blade_handle_t *bh = NULL;
blade_session_t *bs = NULL;
const char *broadcaster_nodeid = NULL;
cJSON *params = NULL;
//cJSON *result = NULL;
ks_assert(brpcreq);
bh = blade_rpc_request_handle_get(brpcreq);
ks_assert(bh);
bs = blade_sessionmgr_session_lookup(blade_handle_sessionmgr_get(bh), blade_rpc_request_sessionid_get(brpcreq));
ks_assert(bs);
params = blade_rpcbroadcast_request_params_get(brpcreq);
ks_assert(params);
broadcaster_nodeid = blade_rpcbroadcast_request_broadcaster_nodeid_get(brpcreq);
ks_assert(broadcaster_nodeid);
ks_log(KS_LOG_DEBUG, "Session (%s) test.leave (%s) broadcast processing\n", blade_session_id_get(bs), broadcaster_nodeid);
blade_session_read_unlock(bs);
return KS_FALSE;
}
ks_bool_t test_talk_broadcast_handler(blade_rpc_request_t *brpcreq, void *data)
{
blade_handle_t *bh = NULL;
blade_session_t *bs = NULL;
const char *broadcaster_nodeid = NULL;
cJSON *params = NULL;
//cJSON *result = NULL;
ks_assert(brpcreq);
bh = blade_rpc_request_handle_get(brpcreq);
ks_assert(bh);
bs = blade_sessionmgr_session_lookup(blade_handle_sessionmgr_get(bh), blade_rpc_request_sessionid_get(brpcreq));
ks_assert(bs);
broadcaster_nodeid = blade_rpcbroadcast_request_broadcaster_nodeid_get(brpcreq);
ks_assert(broadcaster_nodeid);
params = blade_rpcbroadcast_request_params_get(brpcreq);
ks_assert(params);
// @todo pull out text from params
ks_log(KS_LOG_DEBUG, "Session (%s) test.talk (%s) broadcast processing\n", blade_session_id_get(bs), broadcaster_nodeid);
ks_log(KS_LOG_DEBUG, "Session (%s) test broadcast processing\n", blade_session_id_get(bs));
blade_session_read_unlock(bs);
......@@ -391,6 +327,7 @@ void command_locate(blade_handle_t *bh, char *args)
void command_join(blade_handle_t *bh, char *args)
{
cJSON *params = NULL;
cJSON *channels = NULL;
ks_assert(bh);
ks_assert(args);
......@@ -402,17 +339,19 @@ void command_join(blade_handle_t *bh, char *args)
params = cJSON_CreateObject();
blade_handle_rpcexecute(bh, g_testcon_nodeid, "test.join", "test", "mydomain.com", params, test_join_response_handler, NULL);
cJSON_Delete(params);
blade_handle_rpcsubscribe(bh, "test.join", "test", "mydomain.com", KS_FALSE, NULL, NULL, test_join_broadcast_handler, NULL);
blade_handle_rpcsubscribe(bh, "test.leave", "test", "mydomain.com", KS_FALSE, NULL, NULL, test_leave_broadcast_handler, NULL);
blade_handle_rpcsubscribe(bh, "test.talk", "test", "mydomain.com", KS_FALSE, NULL, NULL, test_talk_broadcast_handler, NULL);
channels = cJSON_CreateArray();
cJSON_AddItemToArray(channels, cJSON_CreateString("test"));
blade_handle_rpcsubscribe(bh, "test", "mydomain.com", channels, NULL, NULL, NULL, test_broadcast_handler, NULL);
cJSON_Delete(channels);
}
void command_leave(blade_handle_t *bh, char *args)
{
cJSON *params = NULL;
cJSON *channels = NULL;
ks_assert(bh);
ks_assert(args);
......@@ -423,12 +362,13 @@ void command_leave(blade_handle_t *bh, char *args)
}
params = cJSON_CreateObject();
blade_handle_rpcexecute(bh, g_testcon_nodeid, "test.leave", "test", "mydomain.com", params, test_leave_response_handler, NULL);
cJSON_Delete(params);
blade_handle_rpcsubscribe(bh, "test.join", "test", "mydomain.com", KS_TRUE, NULL, NULL, NULL, NULL);
blade_handle_rpcsubscribe(bh, "test.leave", "test", "mydomain.com", KS_TRUE, NULL, NULL, NULL, NULL);
blade_handle_rpcsubscribe(bh, "test.talk", "test", "mydomain.com", KS_TRUE, NULL, NULL, NULL, NULL);
channels = cJSON_CreateArray();
cJSON_AddItemToArray(channels, cJSON_CreateString("test"));
blade_handle_rpcsubscribe(bh, "test", "mydomain.com", NULL, channels, NULL, NULL, NULL, NULL);
cJSON_Delete(channels);
}
void command_talk(blade_handle_t *bh, char *args)
......@@ -448,10 +388,9 @@ void command_talk(blade_handle_t *bh, char *args)
}
params = cJSON_CreateObject();
cJSON_AddStringToObject(params, "text", args);
blade_handle_rpcexecute(bh, g_testcon_nodeid, "test.talk", "test", "mydomain.com", params, test_talk_response_handler, NULL);
cJSON_Delete(params);
}
/* For Emacs:
......
libs/libblade/test/testcon.c
浏览文件 @ c6e60de3
......@@ -88,7 +88,7 @@ ks_status_t testproto_destroy(testproto_t **testP)
return KS_STATUS_SUCCESS;
}
ks_bool_t test_publish_response_handler(blade_rpc_response_t *brpcres, void *data)
ks_bool_t test_publish_response_handler(blade_rpc_response_t *brpcres, cJSON *data)
{
//testproto_t *test = NULL;
blade_handle_t *bh = NULL;
......@@ -97,7 +97,7 @@ ks_bool_t test_publish_response_handler(blade_rpc_response_t *brpcres, void *dat
ks_assert(brpcres);
ks_assert(data);
//test = (testproto_t *)data;
//test = (testproto_t *)cJSON_GetPtrValue(data);
bh = blade_rpc_response_handle_get(brpcres);
ks_assert(bh);
......@@ -112,7 +112,7 @@ ks_bool_t test_publish_response_handler(blade_rpc_response_t *brpcres, void *dat
return KS_FALSE;
}
ks_bool_t test_join_request_handler(blade_rpc_request_t *brpcreq, void *data)
ks_bool_t test_join_request_handler(blade_rpc_request_t *brpcreq, cJSON *data)
{
testproto_t *test = NULL;
blade_handle_t *bh = NULL;
......@@ -120,48 +120,70 @@ ks_bool_t test_join_request_handler(blade_rpc_request_t *brpcreq, void *data)
const char *requester_nodeid = NULL;
const char *key = NULL;
cJSON *params = NULL;
cJSON *channels = NULL;
cJSON *result = NULL;
ks_assert(brpcreq);
ks_assert(data);
test = (testproto_t *)data;
test = (testproto_t *)cJSON_GetPtrValue(data);
bh = blade_rpc_request_handle_get(brpcreq);
ks_assert(bh);
// session for execute response
bs = blade_sessionmgr_session_lookup(blade_handle_sessionmgr_get(bh), blade_rpc_request_sessionid_get(brpcreq));
ks_assert(bs);
requester_nodeid = blade_rpcexecute_request_requester_nodeid_get(brpcreq);
ks_assert(requester_nodeid);
// inner rpcexecute parameters
params = blade_rpcexecute_request_params_get(brpcreq);
ks_assert(params);
ks_log(KS_LOG_DEBUG, "Session (%s) test.join request processing\n", blade_session_id_get(bs));
// add to participants
key = ks_pstrdup(test->pool, requester_nodeid);
ks_assert(key);
// @todo to properly maintain protocol details tied to a specific node like this participants list requires a way to know if a specific node of interest goes offline to cleanup associated details
// refer back to work notes on ideas about this
ks_hash_write_lock(test->participants);
ks_hash_insert(test->participants, (void *)key, (void *)KS_TRUE);
ks_hash_write_unlock(test->participants);
blade_session_read_unlock(bs);
// authorize channels with the master for the requester
channels = cJSON_CreateArray();
cJSON_AddItemToArray(channels, cJSON_CreateString("channel"));
blade_handle_rpcauthorize(bh, requester_nodeid, KS_FALSE, "test", "mydomain.com", channels, NULL, NULL);
cJSON_Delete(channels);
// send rpcexecute response to the requester
result = cJSON_CreateObject();
blade_rpcexecute_response_send(brpcreq, result);
cJSON_Delete(result);
blade_session_read_unlock(bs);
// broadcast to authorized nodes that have subscribed, that the requester has joined
params = cJSON_CreateObject();
blade_handle_rpcbroadcast(bh, requester_nodeid, "test.join", "test", "mydomain.com", params, NULL, NULL);
cJSON_AddStringToObject(params, "joiner-nodeid", requester_nodeid);
blade_handle_rpcbroadcast(bh, "test", "mydomain.com", "channel", "join", params, NULL, NULL);
cJSON_Delete(params);
return KS_FALSE;
}
ks_bool_t test_leave_request_handler(blade_rpc_request_t *brpcreq, void *data)
ks_bool_t test_leave_request_handler(blade_rpc_request_t *brpcreq, cJSON *data)
{
testproto_t *test = NULL;
blade_handle_t *bh = NULL;
......@@ -174,7 +196,7 @@ ks_bool_t test_leave_request_handler(blade_rpc_request_t *brpcreq, void *data)
ks_assert(brpcreq);
ks_assert(data);
test = (testproto_t *)data;
test = (testproto_t *)cJSON_GetPtrValue(data);
bh = blade_rpc_request_handle_get(brpcreq);
ks_assert(bh);
......@@ -202,12 +224,14 @@ ks_bool_t test_leave_request_handler(blade_rpc_request_t *brpcreq, void *data)
params = cJSON_CreateObject();
blade_handle_rpcbroadcast(bh, requester_nodeid, "test.leave", "test", "mydomain.com", params, NULL, NULL);
cJSON_AddStringToObject(params, "leaver-nodeid", requester_nodeid);
blade_handle_rpcbroadcast(bh, "test", "mydomain.com", "channel", "leave", params, NULL, NULL);
return KS_FALSE;
}
ks_bool_t test_talk_request_handler(blade_rpc_request_t *brpcreq, void *data)
ks_bool_t test_talk_request_handler(blade_rpc_request_t *brpcreq, cJSON *data)
{
//testproto_t *test = NULL;
blade_handle_t *bh = NULL;
......@@ -220,7 +244,7 @@ ks_bool_t test_talk_request_handler(blade_rpc_request_t *brpcreq, void *data)
ks_assert(brpcreq);
ks_assert(data);
//test = (testproto_t *)data;
//test = (testproto_t *)cJSON_GetPtrValue(data);
bh = blade_rpc_request_handle_get(brpcreq);
ks_assert(bh);
......@@ -249,7 +273,9 @@ ks_bool_t test_talk_request_handler(blade_rpc_request_t *brpcreq, void *data)
cJSON_AddStringToObject(params, "text", text);
blade_handle_rpcbroadcast(bh, requester_nodeid, "test.talk", "test", "mydomain.com", params, NULL, NULL);
cJSON_AddStringToObject(params, "talker-nodeid", requester_nodeid);
blade_handle_rpcbroadcast(bh, "test", "mydomain.com", "channel", "talk", params, NULL, NULL);
return KS_FALSE;
}
......@@ -314,19 +340,24 @@ int main(int argc, char **argv)
blade_identity_destroy(&target);
if (connected) {
cJSON *channels = NULL;
// @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, test);
blade_rpc_create(&brpc, bh, "test.join", "test", "mydomain.com", test_join_request_handler, cJSON_CreatePtr((uintptr_t)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, test);
blade_rpc_create(&brpc, bh, "test.leave", "test", "mydomain.com", test_leave_request_handler, cJSON_CreatePtr((uintptr_t)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, test);
blade_rpc_create(&brpc, bh, "test.talk", "test", "mydomain.com", test_talk_request_handler, cJSON_CreatePtr((uintptr_t)test));
blade_rpcmgr_protocolrpc_add(blade_handle_rpcmgr_get(bh), brpc);
blade_handle_rpcpublish(bh, "test", "mydomain.com", test_publish_response_handler, test);
channels = cJSON_CreateArray();
cJSON_AddItemToArray(channels, cJSON_CreateString("channel"));
blade_handle_rpcpublish(bh, "test", "mydomain.com", channels, test_publish_response_handler, cJSON_CreatePtr((uintptr_t)test));
}
}
......
libs/libks/src/include/ks_json.h
浏览文件 @ c6e60de3
......@@ -45,6 +45,9 @@ extern "C"
KS_DECLARE(cJSON *) cJSON_CreateStringPrintf(const char *fmt, ...);
KS_DECLARE(const char *)cJSON_GetObjectCstr(const cJSON *object, const char *string);
KS_DECLARE(cJSON *) cJSON_CreatePtr(uintptr_t pointer);
KS_DECLARE(uintptr_t) cJSON_GetPtrValue(const cJSON *object);
KS_DECLARE(uintptr_t) cJSON_GetObjectPtr(const cJSON *object, const char *string);
static inline cJSON *ks_json_add_child_obj(cJSON *json, const char *name, cJSON *obj)
{
......
libs/libks/src/ks_json.c
浏览文件 @ c6e60de3
......@@ -19,7 +19,7 @@ KS_DECLARE(cJSON *) cJSON_CreateStringPrintf(const char *fmt, ...)
return item;
}
KS_DECLARE(const char *)cJSON_GetObjectCstr(const cJSON *object, const char *string)
KS_DECLARE(const char *) cJSON_GetObjectCstr(const cJSON *object, const char *string)
{
cJSON *cj = cJSON_GetObjectItem(object, string);
......@@ -28,6 +28,25 @@ KS_DECLARE(const char *)cJSON_GetObjectCstr(const cJSON *object, const char *str
return cj->valuestring;
}
KS_DECLARE(cJSON *) cJSON_CreatePtr(uintptr_t pointer)
{
// @todo check for 32bit and use integer storage instead
return cJSON_CreateStringPrintf("%p", (void *)pointer);
}
KS_DECLARE(uintptr_t) cJSON_GetPtrValue(const cJSON *object)
{
// @todo check for 32bit and use integer storage instead
void *pointer = NULL;
if (object && object->type == cJSON_String) sscanf_s(object->valuestring, "%p", &pointer);
return (uintptr_t)pointer;
}
KS_DECLARE(uintptr_t) cJSON_GetObjectPtr(const cJSON *object, const char *string)
{
return cJSON_GetPtrValue(cJSON_GetObjectItem(object, string));
}
/* For Emacs:
* Local Variables:
* mode:c
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论