kenmirrors
/
freeswitch
mirror of https://github.com/signalwire/freeswitch.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

FS-9952: Some code refactoring and added TTL for sessions, currently harcoded at 5 seconds for testing purposes only and should come from config

This commit is contained in:
Shane Bryldt 2017-02-27 18:39:40 +00:00 committed by Mike Jerris
parent 14a99987bb
commit 5d7e40c811
3 changed files with 356 additions and 204 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

284
libs/libblade/src/blade_connection.c
View File

@ -1,23 +1,23 @@
/*
* Copyright (c) 2017, Shane Bryldt
* All rights reserved.
*
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions
* are met:
*
*
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
*
*
* * Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
*
*
* * Neither the name of the original author; nor the names of any contributors
* may be used to endorse or promote products derived from this software
* without specific prior written permission.
*
*
*
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
@ -48,13 +48,19 @@ struct blade_connection_s {
const char *id;
ks_rwl_t *lock;
ks_q_t *sending;
const char *session;
};
void *blade_connection_state_thread(ks_thread_t *thread, void *data);
ks_status_t blade_connection_state_on_disconnect(blade_connection_t *bc);
ks_status_t blade_connection_state_on_new(blade_connection_t *bc);
ks_status_t blade_connection_state_on_connect(blade_connection_t *bc);
ks_status_t blade_connection_state_on_attach(blade_connection_t *bc);
ks_status_t blade_connection_state_on_detach(blade_connection_t *bc);
ks_status_t blade_connection_state_on_ready(blade_connection_t *bc);
KS_DECLARE(ks_status_t) blade_connection_create(blade_connection_t **bcP,
@ -84,7 +90,7 @@ KS_DECLARE(ks_status_t) blade_connection_create(blade_connection_t **bcP,
ks_rwl_create(&bc->lock, pool);
ks_assert(bc->lock);
ks_q_create(&bc->sending, pool, 0);
ks_assert(bc->sending);
@ -136,7 +142,7 @@ KS_DECLARE(ks_status_t) blade_connection_startup(blade_connection_t *bc, blade_c
// @todo error logging
return KS_STATUS_FAIL;
}
ks_log(KS_LOG_DEBUG, "Started\n");
return KS_STATUS_SUCCESS;
@ -246,7 +252,7 @@ KS_DECLARE(void) blade_connection_transport_set(blade_connection_t *bc, void *tr
blade_transport_state_callback_t blade_connection_state_callback_lookup(blade_connection_t *bc, blade_connection_state_t state)
{
blade_transport_state_callback_t callback = NULL;
ks_assert(bc);
switch (state) {
@ -292,7 +298,7 @@ KS_DECLARE(void) blade_connection_state_set(blade_connection_t *bc, blade_connec
if (callback) hook = callback(bc, BLADE_CONNECTION_STATE_CONDITION_PRE);
bc->state = state;
if (hook == BLADE_CONNECTION_STATE_HOOK_DISCONNECT) blade_connection_disconnect(bc);
}
@ -309,7 +315,7 @@ KS_DECLARE(void) blade_connection_disconnect(blade_connection_t *bc)
KS_DECLARE(ks_status_t) blade_connection_sending_push(blade_connection_t *bc, cJSON *json)
{
cJSON *json_copy = NULL;
ks_assert(bc);
ks_assert(json);
@ -344,9 +350,6 @@ void *blade_connection_state_thread(ks_thread_t *thread, void *data)
{
blade_connection_t *bc = NULL;
blade_connection_state_t state;
blade_transport_state_callback_t callback = NULL;
blade_connection_state_hook_t hook = BLADE_CONNECTION_STATE_HOOK_SUCCESS;
cJSON *json = NULL;
ks_assert(thread);
ks_assert(data);
@ -354,92 +357,28 @@ void *blade_connection_state_thread(ks_thread_t *thread, void *data)
bc = (blade_connection_t *)data;
while (!bc->shutdown) {
state = bc->state;
hook = BLADE_CONNECTION_STATE_HOOK_SUCCESS;
callback = blade_connection_state_callback_lookup(bc, state);
if (state == BLADE_CONNECTION_STATE_DISCONNECT) {
blade_handle_connections_remove(bc);
}
// @todo only READY state?
if (state != BLADE_CONNECTION_STATE_DETACH && state != BLADE_CONNECTION_STATE_DISCONNECT) {
while (blade_connection_sending_pop(bc, &json) == KS_STATUS_SUCCESS && json) {
ks_status_t ret = bc->transport_callbacks->onsend(bc, json);
cJSON_Delete(json);
if (ret != KS_STATUS_SUCCESS) {
blade_connection_disconnect(bc);
break;
}
}
}
if (state == BLADE_CONNECTION_STATE_READY) {
ks_bool_t done = KS_FALSE;
while (!done) {
if (bc->transport_callbacks->onreceive(bc, &json) != KS_STATUS_SUCCESS) {
blade_connection_disconnect(bc);
break;
}
if (!(done = (json == NULL))) {
blade_session_t *bs = blade_handle_sessions_get(bc->handle, bc->session);
ks_assert(bs);
blade_session_receiving_push(bs, json);
cJSON_Delete(json);
json = NULL;
}
}
}
if (callback) hook = callback(bc, BLADE_CONNECTION_STATE_CONDITION_POST);
if (hook == BLADE_CONNECTION_STATE_HOOK_DISCONNECT && (state == BLADE_CONNECTION_STATE_DETACH || state == BLADE_CONNECTION_STATE_DISCONNECT))
hook = BLADE_CONNECTION_STATE_HOOK_SUCCESS;
if (hook == BLADE_CONNECTION_STATE_HOOK_DISCONNECT) blade_connection_disconnect(bc);
else if (hook == BLADE_CONNECTION_STATE_HOOK_SUCCESS) {
switch (state) {
case BLADE_CONNECTION_STATE_DISCONNECT:
blade_connection_destroy(&bc);
break;
case BLADE_CONNECTION_STATE_NEW:
blade_connection_state_set(bc, BLADE_CONNECTION_STATE_CONNECT);
break;
case BLADE_CONNECTION_STATE_CONNECT:
blade_connection_state_set(bc, BLADE_CONNECTION_STATE_ATTACH);
break;
case BLADE_CONNECTION_STATE_ATTACH:
{
// @todo this is adding a second lock, since we keep it locked in the callback to allow finishing, we don't want get locking here...
// or just try unlocking twice to confirm...
blade_session_t *bs = blade_handle_sessions_get(bc->handle, bc->session);
ks_assert(bs); // should not happen because bs should still be locked
blade_session_connections_add(bs, bc->id);
blade_connection_state_set(bc, BLADE_CONNECTION_STATE_READY);
blade_session_state_set(bs, BLADE_SESSION_STATE_READY); // @todo only set this if it's not already in the READY state from prior connection
blade_session_read_unlock(bs); // unlock the session we locked obtaining it above
blade_session_read_unlock(bs); // unlock the session we expect to be locked during the callback to ensure we can finish attaching
break;
}
case BLADE_CONNECTION_STATE_DETACH:
{
if (bc->session) {
blade_session_t *bs = blade_handle_sessions_get(bc->handle, bc->session);
ks_assert(bs);
blade_session_connections_remove(bs, bc->id);
blade_session_read_unlock(bs);
// keep bc->session for later in case something triggers a reconnect later and needs the old session id for a hint
}
blade_connection_state_set(bc, BLADE_CONNECTION_STATE_DISCONNECT);
break;
}
default: break;
}
switch (state) {
case BLADE_CONNECTION_STATE_DISCONNECT:
blade_connection_state_on_disconnect(bc);
break;
case BLADE_CONNECTION_STATE_NEW:
blade_connection_state_on_new(bc);
break;
case BLADE_CONNECTION_STATE_CONNECT:
blade_connection_state_on_connect(bc);
break;
case BLADE_CONNECTION_STATE_ATTACH:
blade_connection_state_on_attach(bc);
break;
case BLADE_CONNECTION_STATE_DETACH:
blade_connection_state_on_detach(bc);
break;
case BLADE_CONNECTION_STATE_READY:
blade_connection_state_on_ready(bc);
break;
default: break;
}
if (state == BLADE_CONNECTION_STATE_DISCONNECT) break;
@ -447,7 +386,152 @@ void *blade_connection_state_thread(ks_thread_t *thread, void *data)
return NULL;
}
ks_status_t blade_connection_state_on_disconnect(blade_connection_t *bc)
{
blade_transport_state_callback_t callback = NULL;
ks_assert(bc);
blade_handle_connections_remove(bc);
callback = blade_connection_state_callback_lookup(bc, BLADE_CONNECTION_STATE_DISCONNECT);
if (callback) callback(bc, BLADE_CONNECTION_STATE_CONDITION_POST);
blade_connection_destroy(&bc);
return KS_STATUS_SUCCESS;
}
ks_status_t blade_connection_state_on_new(blade_connection_t *bc)
{
blade_transport_state_callback_t callback = NULL;
blade_connection_state_hook_t hook = BLADE_CONNECTION_STATE_HOOK_SUCCESS;
ks_assert(bc);
callback = blade_connection_state_callback_lookup(bc, BLADE_CONNECTION_STATE_NEW);
if (callback) hook = callback(bc, BLADE_CONNECTION_STATE_CONDITION_POST);
if (hook == BLADE_CONNECTION_STATE_HOOK_DISCONNECT) blade_connection_disconnect(bc);
else if (hook == BLADE_CONNECTION_STATE_HOOK_SUCCESS) {
blade_connection_state_set(bc, BLADE_CONNECTION_STATE_CONNECT);
}
return KS_STATUS_SUCCESS;
}
ks_status_t blade_connection_state_on_connect(blade_connection_t *bc)
{
blade_transport_state_callback_t callback = NULL;
blade_connection_state_hook_t hook = BLADE_CONNECTION_STATE_HOOK_SUCCESS;
ks_assert(bc);
callback = blade_connection_state_callback_lookup(bc, BLADE_CONNECTION_STATE_CONNECT);
if (callback) hook = callback(bc, BLADE_CONNECTION_STATE_CONDITION_POST);
if (hook == BLADE_CONNECTION_STATE_HOOK_DISCONNECT) blade_connection_disconnect(bc);
else if (hook == BLADE_CONNECTION_STATE_HOOK_SUCCESS) {
blade_connection_state_set(bc, BLADE_CONNECTION_STATE_ATTACH);
}
return KS_STATUS_SUCCESS;
}
ks_status_t blade_connection_state_on_attach(blade_connection_t *bc)
{
blade_transport_state_callback_t callback = NULL;
blade_connection_state_hook_t hook = BLADE_CONNECTION_STATE_HOOK_SUCCESS;
ks_assert(bc);
callback = blade_connection_state_callback_lookup(bc, BLADE_CONNECTION_STATE_ATTACH);
if (callback) hook = callback(bc, BLADE_CONNECTION_STATE_CONDITION_POST);
if (hook == BLADE_CONNECTION_STATE_HOOK_DISCONNECT) blade_connection_disconnect(bc);
else if (hook == BLADE_CONNECTION_STATE_HOOK_SUCCESS) {
// @todo this is adding a second lock, since we keep it locked in the callback to allow finishing, we don't want get locking here...
// or just try unlocking twice to confirm...
blade_session_t *bs = blade_handle_sessions_get(bc->handle, bc->session);
ks_assert(bs); // should not happen because bs should still be locked
blade_session_connections_add(bs, bc->id);
blade_connection_state_set(bc, BLADE_CONNECTION_STATE_READY);
blade_session_state_set(bs, BLADE_SESSION_STATE_READY); // @todo only set this if it's not already in the READY state from prior connection
blade_session_read_unlock(bs); // unlock the session we locked obtaining it above
blade_session_read_unlock(bs); // unlock the session we expect to be locked during the callback to ensure we can finish attaching
}
return KS_STATUS_SUCCESS;
}
ks_status_t blade_connection_state_on_detach(blade_connection_t *bc)
{
blade_transport_state_callback_t callback = NULL;
ks_assert(bc);
callback = blade_connection_state_callback_lookup(bc, BLADE_CONNECTION_STATE_DETACH);
if (callback) callback(bc, BLADE_CONNECTION_STATE_CONDITION_POST);
if (bc->session) {
blade_session_t *bs = blade_handle_sessions_get(bc->handle, bc->session);
ks_assert(bs);
blade_session_connections_remove(bs, bc->id);
blade_session_read_unlock(bs);
// keep bc->session for later in case something triggers a reconnect later and needs the old session id for a hint
}
blade_connection_state_set(bc, BLADE_CONNECTION_STATE_DISCONNECT);
return KS_STATUS_SUCCESS;
}
ks_status_t blade_connection_state_on_ready(blade_connection_t *bc)
{
blade_transport_state_callback_t callback = NULL;
blade_connection_state_hook_t hook = BLADE_CONNECTION_STATE_HOOK_SUCCESS;
cJSON *json = NULL;
ks_bool_t done = KS_FALSE;
ks_assert(bc);
while (blade_connection_sending_pop(bc, &json) == KS_STATUS_SUCCESS && json) {
ks_status_t ret = bc->transport_callbacks->onsend(bc, json);
cJSON_Delete(json);
if (ret != KS_STATUS_SUCCESS) {
blade_connection_disconnect(bc);
break;
}
}
while (!done) {
if (bc->transport_callbacks->onreceive(bc, &json) != KS_STATUS_SUCCESS) {
blade_connection_disconnect(bc);
break;
}
if (!(done = (json == NULL))) {
blade_session_t *bs = blade_handle_sessions_get(bc->handle, bc->session);
ks_assert(bs);
blade_session_receiving_push(bs, json);
cJSON_Delete(json);
json = NULL;
}
}
callback = blade_connection_state_callback_lookup(bc, BLADE_CONNECTION_STATE_READY);
if (callback) hook = callback(bc, BLADE_CONNECTION_STATE_CONDITION_POST);
if (hook == BLADE_CONNECTION_STATE_HOOK_DISCONNECT) blade_connection_disconnect(bc);
return KS_STATUS_SUCCESS;
}
/* For Emacs:
* Local Variables:
* mode:c

82
libs/libblade/src/blade_module_wss.c
View File

@ -1,23 +1,23 @@
/*
* Copyright (c) 2017, Shane Bryldt
* All rights reserved.
*
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions
* are met:
*
*
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
*
*
* * Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
*
*
* * Neither the name of the original author; nor the names of any contributors
* may be used to endorse or promote products derived from this software
* without specific prior written permission.
*
*
*
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
@ -135,7 +135,7 @@ static blade_transport_callbacks_t g_transport_wss_callbacks =
blade_transport_wss_on_rank,
blade_transport_wss_on_send,
blade_transport_wss_on_receive,
blade_transport_wss_on_state_disconnect,
blade_transport_wss_on_state_disconnect,
blade_transport_wss_on_state_new_inbound,
@ -156,7 +156,7 @@ ks_status_t blade_module_wss_create(blade_module_wss_t **bm_wssP, blade_handle_t
{
blade_module_wss_t *bm_wss = NULL;
ks_pool_t *pool = NULL;
ks_assert(bm_wssP);
ks_assert(bh);
@ -176,21 +176,21 @@ ks_status_t blade_module_wss_create(blade_module_wss_t **bm_wssP, blade_handle_t
*bm_wssP = bm_wss;
ks_log(KS_LOG_DEBUG, "Created\n");
return KS_STATUS_SUCCESS;
}
ks_status_t blade_module_wss_destroy(blade_module_wss_t **bm_wssP)
{
blade_module_wss_t *bm_wss = NULL;
ks_assert(bm_wssP);
ks_assert(*bm_wssP);
bm_wss = *bm_wssP;
blade_module_wss_on_shutdown(bm_wss->module);
blade_module_destroy(&bm_wss->module);
list_destroy(&bm_wss->connected);
@ -213,7 +213,7 @@ KS_DECLARE(ks_status_t) blade_module_wss_on_load(blade_module_t **bmP, blade_han
ks_assert(bm_wss);
*bmP = bm_wss->module;
ks_log(KS_LOG_DEBUG, "Loaded\n");
return KS_STATUS_SUCCESS;
@ -226,9 +226,9 @@ KS_DECLARE(ks_status_t) blade_module_wss_on_unload(blade_module_t *bm)
ks_assert(bm);
bm_wss = blade_module_data_get(bm);
blade_module_wss_destroy(&bm_wss);
ks_log(KS_LOG_DEBUG, "Unloaded\n");
return KS_STATUS_SUCCESS;
@ -249,7 +249,7 @@ ks_status_t blade_transport_wss_init_create(blade_transport_wss_init_t **bt_wssi
if (session_id) bt_wssi->session_id = ks_pstrdup(bt_wssi->pool, session_id);
*bt_wssiP = bt_wssi;
ks_log(KS_LOG_DEBUG, "Created\n");
return KS_STATUS_SUCCESS;
@ -258,7 +258,7 @@ ks_status_t blade_transport_wss_init_create(blade_transport_wss_init_t **bt_wssi
ks_status_t blade_transport_wss_init_destroy(blade_transport_wss_init_t **bt_wssiP)
{
blade_transport_wss_init_t *bt_wssi = NULL;
ks_assert(bt_wssiP);
ks_assert(*bt_wssiP);
@ -267,7 +267,7 @@ ks_status_t blade_transport_wss_init_destroy(blade_transport_wss_init_t **bt_wss
if (bt_wssi->session_id) ks_pool_free(bt_wssi->pool, &bt_wssi->session_id);
ks_pool_free(bt_wssi->pool, bt_wssiP);
ks_log(KS_LOG_DEBUG, "Destroyed\n");
return KS_STATUS_SUCCESS;
@ -310,7 +310,7 @@ ks_status_t blade_module_wss_config(blade_module_wss_t *bm_wss, config_setting_t
if (config_setting_type(wss_endpoints_ipv4) != CONFIG_TYPE_LIST) return KS_STATUS_FAIL;
if ((config_wss_endpoints_ipv4_length = config_setting_length(wss_endpoints_ipv4)) > BLADE_MODULE_WSS_ENDPOINTS_MULTIHOME_MAX)
return KS_STATUS_FAIL;
for (int32_t index = 0; index < config_wss_endpoints_ipv4_length; ++index) {
element = config_setting_get_elem(wss_endpoints_ipv4, index);
tmp1 = config_lookup_from(element, "address");
@ -385,7 +385,7 @@ ks_status_t blade_module_wss_config(blade_module_wss_t *bm_wss, config_setting_t
KS_DECLARE(ks_status_t) blade_module_wss_on_startup(blade_module_t *bm, config_setting_t *config)
{
blade_module_wss_t *bm_wss = NULL;
ks_assert(bm);
ks_assert(config);
@ -416,9 +416,9 @@ KS_DECLARE(ks_status_t) blade_module_wss_on_startup(blade_module_t *bm, config_s
KS_THREAD_DEFAULT_STACK,
KS_PRI_NORMAL,
bm_wss->pool) != KS_STATUS_SUCCESS) return KS_STATUS_FAIL;
blade_handle_transport_register(bm_wss->handle, bm, BLADE_MODULE_WSS_TRANSPORT_NAME, bm_wss->transport_callbacks);
ks_log(KS_LOG_DEBUG, "Started\n");
return KS_STATUS_SUCCESS;
@ -428,7 +428,7 @@ KS_DECLARE(ks_status_t) blade_module_wss_on_shutdown(blade_module_t *bm)
{
blade_module_wss_t *bm_wss = NULL;
blade_connection_t *bc = NULL;
ks_assert(bm);
bm_wss = (blade_module_wss_t *)blade_module_data_get(bm);
@ -460,7 +460,7 @@ KS_DECLARE(ks_status_t) blade_module_wss_on_shutdown(blade_module_t *bm)
list_iterator_stop(&bm_wss->connected);
while (list_size(&bm_wss->connected) > 0) ks_sleep_ms(100);
}
ks_log(KS_LOG_DEBUG, "Stopped\n");
return KS_STATUS_SUCCESS;
@ -471,7 +471,7 @@ ks_status_t blade_module_wss_listen(blade_module_wss_t *bm_wss, ks_sockaddr_t *a
ks_socket_t listener = KS_SOCK_INVALID;
int32_t listener_index = -1;
ks_status_t ret = KS_STATUS_SUCCESS;
ks_assert(bm_wss);
ks_assert(addr);
@ -549,7 +549,7 @@ void *blade_module_wss_listeners_thread(ks_thread_t *thread, void *data)
}
// @todo getsockname and getpeername (getpeername can be skipped if passing to accept instead)
ks_log(KS_LOG_DEBUG, "Socket accepted\n", index);
blade_transport_wss_init_create(&bt_wss_init, bm_wss, sock, NULL);
@ -557,7 +557,7 @@ void *blade_module_wss_listeners_thread(ks_thread_t *thread, void *data)
blade_connection_create(&bc, bm_wss->handle, bt_wss_init, bm_wss->transport_callbacks);
ks_assert(bc);
blade_connection_read_lock(bc, KS_TRUE);
if (blade_connection_startup(bc, BLADE_CONNECTION_DIRECTION_INBOUND) != KS_STATUS_SUCCESS) {
@ -568,7 +568,7 @@ void *blade_module_wss_listeners_thread(ks_thread_t *thread, void *data)
continue;
}
ks_log(KS_LOG_DEBUG, "Connection (%s) started\n", blade_connection_id_get(bc));
blade_handle_connections_add(bc);
list_append(&bm_wss->connected, bc);
blade_connection_state_set(bc, BLADE_CONNECTION_STATE_NEW);
@ -598,7 +598,7 @@ ks_status_t blade_transport_wss_create(blade_transport_wss_t **bt_wssP, blade_mo
bt_wss->sock = sock;
*bt_wssP = bt_wss;
ks_log(KS_LOG_DEBUG, "Created\n");
return KS_STATUS_SUCCESS;
@ -607,7 +607,7 @@ ks_status_t blade_transport_wss_create(blade_transport_wss_t **bt_wssP, blade_mo
ks_status_t blade_transport_wss_destroy(blade_transport_wss_t **bt_wssP)
{
blade_transport_wss_t *bt_wss = NULL;
ks_assert(bt_wssP);
ks_assert(*bt_wssP);
@ -615,9 +615,9 @@ ks_status_t blade_transport_wss_destroy(blade_transport_wss_t **bt_wssP)
if (bt_wss->kws) kws_destroy(&bt_wss->kws);
else ks_socket_close(&bt_wss->sock);
ks_pool_free(bt_wss->pool, bt_wssP);
ks_log(KS_LOG_DEBUG, "Destroyed\n");
return KS_STATUS_SUCCESS;
@ -641,7 +641,7 @@ ks_status_t blade_transport_wss_on_connect(blade_connection_t **bcP, blade_modul
ks_assert(target);
bm_wss = (blade_module_wss_t *)blade_module_data_get(bm);
*bcP = NULL;
ks_log(KS_LOG_DEBUG, "Connect Callback: %s\n", blade_identity_uri(target));
@ -670,7 +670,7 @@ ks_status_t blade_transport_wss_on_connect(blade_connection_t **bcP, blade_modul
if (ip[1] == '.' || ip[2] == '.' || (len > 3 && ip[3] == '.')) family = AF_INET;
else family = AF_INET6;
}
if (portstr) {
int p = atoi(portstr);
if (p > 0 && p <= UINT16_MAX) port = p;
@ -720,7 +720,7 @@ blade_connection_rank_t blade_transport_wss_on_rank(blade_connection_t *bc, blad
{
ks_assert(bc);
ks_assert(target);
return BLADE_CONNECTION_RANK_POOR;
}
@ -780,7 +780,7 @@ ks_status_t blade_transport_wss_read(blade_transport_wss_t *bt_wss, cJSON **json
if (poll_flags & KS_POLL_READ) {
kws_opcode_t opcode;
uint8_t *frame_data = NULL;
ks_size_t frame_data_len = kws_read_frame(bt_wss->kws, &opcode, &frame_data);
ks_ssize_t frame_data_len = kws_read_frame(bt_wss->kws, &opcode, &frame_data);
if (frame_data_len <= 0) {
// @todo error logging, strerror(ks_errno())
@ -1033,7 +1033,7 @@ blade_connection_state_hook_t blade_transport_wss_on_state_attach_inbound(blade_
json_res = cJSON_CreateObject();
cJSON_AddStringToObject(json_res, "jsonrpc", "2.0");
cJSON_AddStringToObject(json_res, "id", id);
result = cJSON_CreateObject();
cJSON_AddStringToObject(result, "session-id", blade_session_id_get(bs));
cJSON_AddItemToObject(json_res, "result", result);
@ -1046,7 +1046,7 @@ blade_connection_state_hook_t blade_transport_wss_on_state_attach_inbound(blade_
}
blade_connection_session_set(bc, blade_session_id_get(bs));
done:
// @note the state machine expects if we return SUCCESS, that the session assigned to the connection will be read locked to ensure that the state
// machine can finish attaching the session, if you BYPASS then you can handle everything here in the callback, but this should be fairly standard
@ -1103,7 +1103,7 @@ blade_connection_state_hook_t blade_transport_wss_on_state_attach_outbound(blade
}
ks_log(KS_LOG_DEBUG, "Session (%s) requested\n", (bt_wss_init->session_id ? bt_wss_init->session_id : "none"));
if (blade_transport_wss_write(bt_wss, json_req) != KS_STATUS_SUCCESS) {
ks_log(KS_LOG_DEBUG, "Failed to write message\n");
ret = BLADE_CONNECTION_STATE_HOOK_DISCONNECT;
@ -1123,7 +1123,7 @@ blade_connection_state_hook_t blade_transport_wss_on_state_attach_outbound(blade
ret = BLADE_CONNECTION_STATE_HOOK_DISCONNECT;
goto done;
}
// @todo validation wrapper for request and response/error to confirm jsonrpc and provide enum for output as to which it is
jsonrpc = cJSON_GetObjectCstr(json_res, "jsonrpc"); // @todo check for definitions of these keys and fixed values
if (!jsonrpc || strcmp(jsonrpc, "2.0")) {
@ -1159,7 +1159,7 @@ blade_connection_state_hook_t blade_transport_wss_on_state_attach_outbound(blade
ret = BLADE_CONNECTION_STATE_HOOK_DISCONNECT;
goto done;
}
if (sid) {
// @todo validate uuid format by parsing, not currently available in uuid functions
bs = blade_handle_sessions_get(bh, sid); // bs comes out read locked if not null to prevent it being cleaned up before we are done
@ -1203,7 +1203,7 @@ blade_connection_state_hook_t blade_transport_wss_on_state_detach(blade_connecti
ks_assert(bc);
ks_log(KS_LOG_DEBUG, "State Callback: %d\n", (int32_t)condition);
return BLADE_CONNECTION_STATE_HOOK_SUCCESS;
}

194
libs/libblade/src/blade_session.c
View File

@ -1,23 +1,23 @@
/*
* Copyright (c) 2017, Shane Bryldt
* All rights reserved.
*
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions
* are met:
*
*
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
*
*
* * Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
*
*
* * Neither the name of the original author; nor the names of any contributors
* may be used to endorse or promote products derived from this software
* without specific prior written permission.
*
*
*
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
@ -44,20 +44,24 @@ struct blade_session_s {
const char *id;
ks_rwl_t *lock;
list_t connections;
ks_time_t ttl;
ks_q_t *sending;
ks_q_t *receiving;
};
void *blade_session_state_thread(ks_thread_t *thread, void *data);
ks_status_t blade_session_state_on_destroy(blade_session_t *bs);
ks_status_t blade_session_state_on_hangup(blade_session_t *bs);
ks_status_t blade_session_state_on_ready(blade_session_t *bs);
ks_status_t blade_session_process(blade_session_t *bs, cJSON *json);
KS_DECLARE(ks_status_t) blade_session_create(blade_session_t **bsP, blade_handle_t *bh)
{
blade_session_t *bs = NULL;
ks_pool_t *pool = NULL;
uuid_t id;
ks_assert(bsP);
ks_assert(bh);
@ -72,7 +76,7 @@ KS_DECLARE(ks_status_t) blade_session_create(blade_session_t **bsP, blade_handle
ks_rwl_create(&bs->lock, pool);
ks_assert(bs->lock);
list_init(&bs->connections);
ks_q_create(&bs->sending, pool, 0);
ks_assert(bs->sending);
@ -128,7 +132,7 @@ KS_DECLARE(ks_status_t) blade_session_startup(blade_session_t *bs)
// @todo error logging
return KS_STATUS_FAIL;
}
ks_log(KS_LOG_DEBUG, "Started\n");
return KS_STATUS_SUCCESS;
@ -157,7 +161,7 @@ KS_DECLARE(ks_status_t) blade_session_shutdown(blade_session_t *bs)
}
list_iterator_stop(&bs->connections);
list_clear(&bs->connections);
ks_log(KS_LOG_DEBUG, "Stopped\n");
return KS_STATUS_SUCCESS;
@ -256,11 +260,13 @@ KS_DECLARE(ks_status_t) blade_session_connections_add(blade_session_t *bs, const
cid = ks_pstrdup(bs->pool, id);
ks_assert(cid);
list_append(&bs->connections, cid);
ks_log(KS_LOG_DEBUG, "Session (%s) connection added (%s)\n", bs->id, id);
bs->ttl = 0;
return ret;
}
@ -282,6 +288,8 @@ KS_DECLARE(ks_status_t) blade_session_connections_remove(blade_session_t *bs, co
}
}
if (list_size(&bs->connections) == 0) bs->ttl = ks_time_now() + (5 * KS_USEC_PER_SEC);
return ret;
}
@ -301,7 +309,7 @@ ks_status_t blade_session_connections_choose(blade_session_t *bs, cJSON *json, b
// no connections available
return KS_STATUS_FAIL;
}
bc = blade_handle_connections_get(bs->handle, cid);
if (!bc) {
// @todo error logging... this shouldn't happen
@ -314,28 +322,6 @@ ks_status_t blade_session_connections_choose(blade_session_t *bs, cJSON *json, b
return KS_STATUS_SUCCESS;
}
KS_DECLARE(ks_status_t) blade_session_send(blade_session_t *bs, cJSON *json)
{
ks_assert(bs);
ks_assert(json);
// @todo check json for "method", if this is an outgoing request then build up the data for a response to lookup the message id and get back to the request
// this can reuse blade_request_t so that when the blade_response_t is passed up the blade_request_t within it is familiar from inbound requests
if (list_empty(&bs->connections)) {
// @todo cache the blade_request_t here if it exists to gaurentee it's cached before a response could be received
blade_session_sending_push(bs, json);
} else {
blade_connection_t *bc = NULL;
if (blade_session_connections_choose(bs, json, &bc) != KS_STATUS_SUCCESS) return KS_STATUS_FAIL;
// @todo cache the blade_request_t here if it exists to gaurentee it's cached before a response could be received
blade_connection_sending_push(bc, json);
blade_connection_read_unlock(bc);
}
return KS_STATUS_SUCCESS;
}
KS_DECLARE(ks_status_t) blade_session_sending_push(blade_session_t *bs, cJSON *json)
{
cJSON *json_copy = NULL;
@ -387,7 +373,7 @@ void *blade_session_state_thread(ks_thread_t *thread, void *data)
bs = (blade_session_t *)data;
while (!bs->shutdown) {
state = bs->state;
if (!list_empty(&bs->connections)) {
@ -403,30 +389,11 @@ void *blade_session_state_thread(ks_thread_t *thread, void *data)
switch (state) {
case BLADE_SESSION_STATE_DESTROY:
ks_log(KS_LOG_DEBUG, "Session (%s) state destroy\n", bs->id);
blade_handle_sessions_remove(bs);
blade_session_destroy(&bs);
blade_session_state_on_destroy(bs);
return NULL;
case BLADE_SESSION_STATE_HANGUP:
{
ks_log(KS_LOG_DEBUG, "Session (%s) state hangup\n", bs->id);
list_iterator_start(&bs->connections);
while (list_iterator_hasnext(&bs->connections)) {
const char *cid = (const char *)list_iterator_next(&bs->connections);
blade_connection_t *bc = blade_handle_connections_get(bs->handle, cid);
ks_assert(bc);
blade_connection_disconnect(bc);
blade_connection_read_unlock(bc);
}
list_iterator_stop(&bs->connections);
while (!list_empty(&bs->connections)) ks_sleep(100);
blade_session_state_set(bs, BLADE_SESSION_STATE_DESTROY);
break;
}
blade_session_state_on_hangup(bs);
break;
case BLADE_SESSION_STATE_CONNECT:
ks_log(KS_LOG_DEBUG, "Session (%s) state connect\n", bs->id);
ks_sleep_ms(1000);
@ -440,17 +407,118 @@ void *blade_session_state_thread(ks_thread_t *thread, void *data)
ks_sleep_ms(1000);
break;
case BLADE_SESSION_STATE_READY:
ks_log(KS_LOG_DEBUG, "Session (%s) state ready\n", bs->id);
// @todo pop from session receiving queue and pass into protocol layer through something like blade_protocol_process()
ks_sleep_ms(1000);
blade_session_state_on_ready(bs);
break;
default: break;
}
if (list_empty(&bs->connections) &&
bs->ttl > 0 &&
bs->state != BLADE_SESSION_STATE_HANGUP &&
bs->state != BLADE_SESSION_STATE_DESTROY &&
ks_time_now() >= bs->ttl) {
ks_log(KS_LOG_DEBUG, "Session (%s) TTL timeout\n", bs->id);
blade_session_hangup(bs);
}
}
return NULL;
}
ks_status_t blade_session_state_on_destroy(blade_session_t *bs)
{
ks_assert(bs);
ks_log(KS_LOG_DEBUG, "Session (%s) state destroy\n", bs->id);
blade_handle_sessions_remove(bs);
blade_session_destroy(&bs);
// @todo ignoring returns for now, see what makes sense later
return KS_STATUS_SUCCESS;
}
ks_status_t blade_session_state_on_hangup(blade_session_t *bs)
{
ks_assert(bs);
ks_log(KS_LOG_DEBUG, "Session (%s) state hangup\n", bs->id);
list_iterator_start(&bs->connections);
while (list_iterator_hasnext(&bs->connections)) {
const char *cid = (const char *)list_iterator_next(&bs->connections);
blade_connection_t *bc = blade_handle_connections_get(bs->handle, cid);
ks_assert(bc);
blade_connection_disconnect(bc);
blade_connection_read_unlock(bc);
}
list_iterator_stop(&bs->connections);
while (!list_empty(&bs->connections)) ks_sleep(100);
blade_session_state_set(bs, BLADE_SESSION_STATE_DESTROY);
return KS_STATUS_SUCCESS;
}
ks_status_t blade_session_state_on_ready(blade_session_t *bs)
{
cJSON *json = NULL;
ks_assert(bs);
ks_log(KS_LOG_DEBUG, "Session (%s) state ready\n", bs->id);
// @todo for now only process messages if there is a connection available
if (list_size(&bs->connections) > 0) {
// @todo may only want to pop once per call to give sending a chance to keep up
while (blade_session_receiving_pop(bs, &json) == KS_STATUS_SUCCESS && json) {
blade_session_process(bs, json);
cJSON_Delete(json);
}
}
ks_sleep_ms(1000);
return KS_STATUS_SUCCESS;
}
KS_DECLARE(ks_status_t) blade_session_send(blade_session_t *bs, cJSON *json)
{
ks_assert(bs);
ks_assert(json);
// @todo check json for "method", if this is an outgoing request then build up the data for a response to lookup the message id and get back to the request
// this can reuse blade_request_t so that when the blade_response_t is passed up the blade_request_t within it is familiar from inbound requests
if (list_empty(&bs->connections)) {
// @todo cache the blade_request_t here if it exists to gaurentee it's cached before a response could be received
blade_session_sending_push(bs, json);
} else {
blade_connection_t *bc = NULL;
if (blade_session_connections_choose(bs, json, &bc) != KS_STATUS_SUCCESS) return KS_STATUS_FAIL;
// @todo cache the blade_request_t here if it exists to gaurentee it's cached before a response could be received
blade_connection_sending_push(bc, json);
blade_connection_read_unlock(bc);
}
return KS_STATUS_SUCCESS;
}
ks_status_t blade_session_process(blade_session_t *bs, cJSON *json)
{
ks_status_t ret = KS_STATUS_SUCCESS;
ks_assert(bs);
ks_assert(json);
ks_log(KS_LOG_DEBUG, "Session (%s) processing\n", bs->id);
// @todo teardown the message, convert into a blade_request_t or blade_response_t
return ret;
}
/* For Emacs:
* Local Variables:
* mode:c