FS-10167: Implemented identity registration and routing, removed subrealms to be revisited later, implemented public event channels and node presence, restructured some code and fixed a number of bugs.

This commit is contained in:
Shane Bryldt committed 2017-09-20 22:07:34 -06:00
1 parent dacca35a18
commit 43cb4e3468
38 files changed
+1396 -1365

No files matched your search

+4 -8
View File
@@ -66,7 +66,6 @@ ks_bool_t blade_locate_response_handler(blade_rpc_response_t *brpcres, void *dat
cJSON *res_result = NULL;
cJSON *res_result_controllers = NULL;
const char *res_result_protocol = NULL;
const char *res_result_realm = NULL;
cJSON *params = NULL;
ks_assert(brpcres);
@@ -86,9 +85,6 @@ ks_bool_t blade_locate_response_handler(blade_rpc_response_t *brpcres, void *dat
res_result_protocol = cJSON_GetObjectCstr(res_result, "protocol");
ks_assert(res_result_protocol);
res_result_realm = cJSON_GetObjectCstr(res_result, "realm");
ks_assert(res_result_realm);
res_result_controllers = cJSON_GetObjectItem(res_result, "controllers");
ks_assert(res_result_controllers);
@@ -97,7 +93,7 @@ ks_bool_t blade_locate_response_handler(blade_rpc_response_t *brpcres, void *dat
for (int index = 0; index < cJSON_GetArraySize(res_result_controllers); ++index) {
cJSON *elem = cJSON_GetArrayItem(res_result_controllers, index);
if (elem->type == cJSON_String) {
ks_log(KS_LOG_DEBUG, "Session (%s) blade.locate (%s@%s) provider (%s)\n", blade_session_id_get(bs), res_result_protocol, res_result_realm, elem->valuestring);
ks_log(KS_LOG_DEBUG, "Session (%s) blade.locate (%s) provider (%s)\n", blade_session_id_get(bs), res_result_protocol, elem->valuestring);
nodeid = elem->valuestring;
}
}
@@ -106,7 +102,7 @@ ks_bool_t blade_locate_response_handler(blade_rpc_response_t *brpcres, void *dat
params = cJSON_CreateObject();
cJSON_AddStringToObject(params, "text", "hello world!");
blade_handle_rpcexecute(bh, nodeid, "test.echo", res_result_protocol, res_result_realm, params, test_echo_response_handler, NULL);
blade_handle_rpcexecute(bh, nodeid, "test.echo", res_result_protocol, params, test_echo_response_handler, NULL);
return KS_FALSE;
}
@@ -285,7 +281,7 @@ void command_execute(blade_handle_t *bh, char *args)
ks_assert(bh);
ks_assert(args);
blade_handle_rpclocate(bh, "test", "mydomain.com", blade_locate_response_handler, NULL);
blade_handle_rpclocate(bh, "test", blade_locate_response_handler, NULL);
}
void command_subscribe(blade_handle_t *bh, char *args)
@@ -297,7 +293,7 @@ void command_subscribe(blade_handle_t *bh, char *args)
channels = cJSON_CreateArray();
cJSON_AddItemToArray(channels, cJSON_CreateString("test"));
blade_handle_rpcsubscribe(bh, BLADE_RPCSUBSCRIBE_COMMAND_SUBSCRIBER_ADD, "test", "mydomain.com", channels, blade_subscribe_response_handler, NULL, test_event_request_handler, NULL);
blade_handle_rpcsubscribe(bh, BLADE_RPCSUBSCRIBE_COMMAND_SUBSCRIBER_ADD, "test", channels, blade_subscribe_response_handler, NULL, test_event_request_handler, NULL);
cJSON_Delete(channels);
}
+3 -3
View File
@@ -240,11 +240,11 @@ void command_publish(blade_handle_t *bh, char *args)
ks_assert(bh);
ks_assert(args);
blade_rpc_create(&brpc, bh, "test.echo", "test", "mydomain.com", test_echo_request_handler, NULL);
blade_rpc_create(&brpc, bh, "test.echo", "test", test_echo_request_handler, NULL);
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, BLADE_RPCPUBLISH_COMMAND_CONTROLLER_ADD, "test", "mydomain.com", NULL, blade_publish_response_handler, NULL);
blade_handle_rpcpublish(bh, BLADE_RPCPUBLISH_COMMAND_CONTROLLER_ADD, "test", NULL, blade_publish_response_handler, NULL);
}
void command_broadcast(blade_handle_t *bh, char *args)
@@ -252,7 +252,7 @@ void command_broadcast(blade_handle_t *bh, char *args)
ks_assert(bh);
ks_assert(args);
blade_handle_rpcbroadcast(bh, "test", "mydomain.com", "channel", "event", NULL, test_broadcast_response_handler, NULL);
blade_handle_rpcbroadcast(bh, "test", "channel", "event", NULL, test_broadcast_response_handler, NULL);
}
+10 -13
View File
@@ -46,7 +46,6 @@ ks_bool_t test_locate_response_handler(blade_rpc_response_t *brpcres, void *data
cJSON *res_result = NULL;
cJSON *res_result_controllers = NULL;
const char *res_result_protocol = NULL;
const char *res_result_realm = NULL;
//cJSON *params = NULL;
ks_assert(brpcres);
@@ -66,12 +65,9 @@ ks_bool_t test_locate_response_handler(blade_rpc_response_t *brpcres, void *data
res_result_protocol = cJSON_GetObjectCstr(res_result, "protocol");
ks_assert(res_result_protocol);
res_result_realm = cJSON_GetObjectCstr(res_result, "realm");
ks_assert(res_result_realm);
res_result_controllers = cJSON_GetObjectItem(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);
ks_log(KS_LOG_DEBUG, "Session (%s) locate (%s) response processing\n", blade_session_id_get(bs), res_result_protocol);
if (res_result_controllers) {
for (int index = 0; index < cJSON_GetArraySize(res_result_controllers); ++index) {
@@ -89,7 +85,7 @@ ks_bool_t test_locate_response_handler(blade_rpc_response_t *brpcres, void *data
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);
ks_log(KS_LOG_DEBUG, "Session (%s) locate (%s) provider (%s)\n", blade_session_id_get(bs), res_result_protocol, g_testcon_nodeid);
return KS_FALSE;
}
@@ -273,7 +269,7 @@ int main(int argc, char **argv)
if (connected) {
// @todo use session state change callback to know when the session is ready and the realm(s) available from blade.connect
// @todo use session state change callback to know when the session is ready after blade.connect
ks_sleep_ms(3000);
}
}
@@ -356,7 +352,7 @@ void command_locate(blade_handle_t *bh, char *args)
ks_assert(bh);
ks_assert(args);
blade_handle_rpclocate(bh, "test", "mydomain.com", test_locate_response_handler, NULL);
blade_handle_rpclocate(bh, "test", test_locate_response_handler, NULL);
}
void command_join(blade_handle_t *bh, char *args)
@@ -372,7 +368,7 @@ 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);
blade_handle_rpcexecute(bh, g_testcon_nodeid, "test.join", "test", params, test_join_response_handler, NULL);
cJSON_Delete(params);
}
@@ -388,7 +384,7 @@ void command_subscribe(blade_handle_t *bh, char *args)
channels = cJSON_CreateArray();
cJSON_AddItemToArray(channels, cJSON_CreateString("channel"));
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);
blade_handle_rpcsubscribe(bh, BLADE_RPCSUBSCRIBE_COMMAND_SUBSCRIBER_ADD, "test", channels, NULL, NULL, test_channel_handler, NULL);
cJSON_Delete(channels);
}
@@ -404,7 +400,7 @@ void command_unsubscribe(blade_handle_t *bh, char *args)
channels = cJSON_CreateArray();
cJSON_AddItemToArray(channels, cJSON_CreateString("channel"));
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);
blade_handle_rpcsubscribe(bh, BLADE_RPCSUBSCRIBE_COMMAND_SUBSCRIBER_REMOVE, "test", channels, test_subscribe_response_handler, NULL, test_channel_handler, NULL);
cJSON_Delete(channels);
}
@@ -421,7 +417,7 @@ 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);
blade_handle_rpcexecute(bh, g_testcon_nodeid, "test.leave", "test", params, test_leave_response_handler, NULL);
cJSON_Delete(params);
}
@@ -443,7 +439,8 @@ 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);
//blade_handle_rpcexecute(bh, g_testcon_nodeid, "test.talk", "test", params, test_talk_response_handler, NULL);
blade_handle_rpcexecute(bh, "blade:testcon@freeswitch.com", "test.talk", "test", params, test_talk_response_handler, NULL);
cJSON_Delete(params);
}
+47 -64
View File
@@ -16,15 +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);
void command_presence(blade_handle_t *bh, char *args);
void command_identity(blade_handle_t *bh, char *args);
static const struct command_def_s command_defs[] = {
{ "quit", command_quit },
{ "channeladd", command_channeladd },
{ "channelremove", command_channelremove },
{ "presence", command_presence },
{ "identity", command_identity },
{ NULL, NULL }
};
@@ -175,7 +173,7 @@ ks_bool_t test_join_request_handler(blade_rpc_request_t *brpcreq, void *data)
cJSON_AddItemToArray(channels, cJSON_CreateString((const char *)key));
}
blade_handle_rpcauthorize(bh, requester_nodeid, KS_FALSE, "test", "mydomain.com", channels, NULL, NULL);
blade_handle_rpcauthorize(bh, requester_nodeid, KS_FALSE, "test", channels, NULL, NULL);
cJSON_Delete(channels);
@@ -193,7 +191,7 @@ ks_bool_t test_join_request_handler(blade_rpc_request_t *brpcreq, void *data)
cJSON_AddStringToObject(params, "joiner-nodeid", requester_nodeid);
blade_handle_rpcbroadcast(bh, "test", "mydomain.com", "channel", "join", params, NULL, NULL);
blade_handle_rpcbroadcast(bh, "test", "channel", "join", params, NULL, NULL);
cJSON_Delete(params);
@@ -245,7 +243,7 @@ ks_bool_t test_leave_request_handler(blade_rpc_request_t *brpcreq, void *data)
cJSON_AddItemToArray(channels, cJSON_CreateString((const char *)key));
}
blade_handle_rpcauthorize(bh, requester_nodeid, KS_TRUE, "test", "mydomain.com", channels, NULL, NULL);
blade_handle_rpcauthorize(bh, requester_nodeid, KS_TRUE, "test", channels, NULL, NULL);
cJSON_Delete(channels);
@@ -263,7 +261,7 @@ ks_bool_t test_leave_request_handler(blade_rpc_request_t *brpcreq, void *data)
cJSON_AddStringToObject(params, "leaver-nodeid", requester_nodeid);
blade_handle_rpcbroadcast(bh, "test", "mydomain.com", "channel", "leave", params, NULL, NULL);
blade_handle_rpcbroadcast(bh, "test", "channel", "leave", params, NULL, NULL);
cJSON_Delete(params);
@@ -318,7 +316,7 @@ ks_bool_t test_talk_request_handler(blade_rpc_request_t *brpcreq, void *data)
cJSON_AddStringToObject(params, "talker-nodeid", requester_nodeid);
blade_handle_rpcbroadcast(bh, "test", "mydomain.com", "channel", "talk", params, NULL, NULL);
blade_handle_rpcbroadcast(bh, "test", "channel", "talk", params, NULL, NULL);
cJSON_Delete(params);
@@ -329,7 +327,6 @@ ks_bool_t test_presence_request_handler(blade_rpc_request_t *brpcreq, void *data
{
blade_handle_t *bh = NULL;
blade_session_t *bs = NULL;
const char *realm = NULL;
const char *protocol = NULL;
const char *channel = NULL;
const char *event = NULL;
@@ -345,7 +342,6 @@ ks_bool_t test_presence_request_handler(blade_rpc_request_t *brpcreq, void *data
bs = blade_sessionmgr_session_lookup(blade_handle_sessionmgr_get(bh), blade_rpc_request_sessionid_get(brpcreq));
ks_assert(bs);
realm = blade_rpcbroadcast_request_realm_get(brpcreq);
protocol = blade_rpcbroadcast_request_protocol_get(brpcreq);
channel = blade_rpcbroadcast_request_channel_get(brpcreq);
event = blade_rpcbroadcast_request_event_get(brpcreq);
@@ -353,7 +349,27 @@ ks_bool_t test_presence_request_handler(blade_rpc_request_t *brpcreq, void *data
params = blade_rpcbroadcast_request_params_get(brpcreq);
nodeid = cJSON_GetObjectCstr(params, "nodeid");
ks_log(KS_LOG_DEBUG, "Session (%s) presence (%s@%s/%s/%s for %s) request processing\n", blade_session_id_get(bs), protocol, realm, channel, event, nodeid);
ks_log(KS_LOG_DEBUG, "Session (%s) presence (protocol %s, channel %s, event %s for %s) request processing\n", blade_session_id_get(bs), protocol, channel, event, nodeid);
return KS_FALSE;
}
ks_bool_t test_register_response_handler(blade_rpc_response_t *brpcres, void *data)
{
blade_handle_t *bh = NULL;
blade_session_t *bs = NULL;
ks_assert(brpcres);
bh = blade_rpc_response_handle_get(brpcres);
ks_assert(bh);
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) register response processing\n", blade_session_id_get(bs));
blade_session_read_unlock(bs);
return KS_FALSE;
}
@@ -418,23 +434,27 @@ int main(int argc, char **argv)
if (connected) {
cJSON *channels = NULL;
cJSON *entry = 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
// @todo use session state change callback to know when the session is ready after 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 *)g_test);
blade_rpc_create(&brpc, bh, "test.join", "test", 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 *)g_test);
blade_rpc_create(&brpc, bh, "test.leave", "test", 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 *)g_test);
blade_rpc_create(&brpc, bh, "test.talk", "test", 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"));
entry = cJSON_CreateObject();
cJSON_AddStringToObject(entry, "name", "channel");
cJSON_AddNumberToObject(entry, "flags", BLADE_CHANNEL_FLAGS_NONE);
cJSON_AddItemToArray(channels, entry);
blade_handle_rpcpublish(bh, BLADE_RPCPUBLISH_COMMAND_CONTROLLER_ADD, "test", "mydomain.com", channels, test_publish_response_handler, (void *)g_test);
blade_handle_rpcpublish(bh, BLADE_RPCPUBLISH_COMMAND_CONTROLLER_ADD, "test", channels, test_publish_response_handler, (void *)g_test);
cJSON_Delete(channels);
}
@@ -515,51 +535,6 @@ 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);
}
}
void command_presence(blade_handle_t *bh, char *args)
{
cJSON *channels = NULL;
@@ -571,7 +546,15 @@ void command_presence(blade_handle_t *bh, char *args)
cJSON_AddItemToArray(channels, cJSON_CreateString("join"));
cJSON_AddItemToArray(channels, cJSON_CreateString("leave"));
blade_handle_rpcsubscribe(bh, BLADE_RPCSUBSCRIBE_COMMAND_SUBSCRIBER_ADD, "presence", "blade", channels, NULL, NULL, test_presence_request_handler, (void *)g_test);
blade_handle_rpcsubscribe(bh, BLADE_RPCSUBSCRIBE_COMMAND_SUBSCRIBER_ADD, "blade.presence", channels, NULL, NULL, test_presence_request_handler, (void *)g_test);
}
void command_identity(blade_handle_t *bh, char *args)
{
ks_assert(bh);
ks_assert(args);
blade_handle_rpcregister(bh, "blade:testcon@freeswitch.com", test_register_response_handler, NULL);
}
/* For Emacs: