diff --git a/libs/libblade/src/blade_connection.c b/libs/libblade/src/blade_connection.c index 7d5cfadfe0..6816d61db9 100644 --- a/libs/libblade/src/blade_connection.c +++ b/libs/libblade/src/blade_connection.c @@ -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 diff --git a/libs/libblade/src/blade_module_wss.c b/libs/libblade/src/blade_module_wss.c index 31d2235e30..ae5c5e1f96 100644 --- a/libs/libblade/src/blade_module_wss.c +++ b/libs/libblade/src/blade_module_wss.c @@ -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; } diff --git a/libs/libblade/src/blade_session.c b/libs/libblade/src/blade_session.c index 06f500a9d7..d5037bfcad 100644 --- a/libs/libblade/src/blade_session.c +++ b/libs/libblade/src/blade_session.c @@ -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