/*
* 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
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,
ks_rwl_create(&bc->lock, pool);
ks_assert(bc->lock);
-
+
ks_q_create(&bc->sending, pool, 0);
ks_assert(bc->sending);
// @todo error logging
return KS_STATUS_FAIL;
}
-
+
ks_log(KS_LOG_DEBUG, "Started\n");
return KS_STATUS_SUCCESS;
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) {
if (callback) hook = callback(bc, BLADE_CONNECTION_STATE_CONDITION_PRE);
bc->state = state;
-
+
if (hook == BLADE_CONNECTION_STATE_HOOK_DISCONNECT) blade_connection_disconnect(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);
{
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);
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);
+ 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;
}
- // @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_DISCONNECT) break;
+ }
+
+ 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;
}
+ }
- 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;
- }
- }
+ while (!done) {
+ if (bc->transport_callbacks->onreceive(bc, &json) != KS_STATUS_SUCCESS) {
+ blade_connection_disconnect(bc);
+ break;
}
-
- 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;
- }
+ 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 (state == BLADE_CONNECTION_STATE_DISCONNECT) break;
}
- return 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
/*
* 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
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,
{
blade_module_wss_t *bm_wss = NULL;
ks_pool_t *pool = NULL;
-
+
ks_assert(bm_wssP);
ks_assert(bh);
*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);
ks_assert(bm_wss);
*bmP = bm_wss->module;
-
+
ks_log(KS_LOG_DEBUG, "Loaded\n");
return KS_STATUS_SUCCESS;
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;
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;
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);
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;
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");
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);
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;
{
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);
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;
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);
}
// @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);
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) {
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);
bt_wss->sock = sock;
*bt_wssP = bt_wss;
-
+
ks_log(KS_LOG_DEBUG, "Created\n");
return KS_STATUS_SUCCESS;
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);
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;
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));
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;
{
ks_assert(bc);
ks_assert(target);
-
+
return BLADE_CONNECTION_RANK_POOR;
}
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())
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);
}
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
}
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;
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")) {
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
ks_assert(bc);
ks_log(KS_LOG_DEBUG, "State Callback: %d\n", (int32_t)condition);
-
+
return BLADE_CONNECTION_STATE_HOOK_SUCCESS;
}
/*
* 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
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);
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);
// @todo error logging
return KS_STATUS_FAIL;
}
-
+
ks_log(KS_LOG_DEBUG, "Started\n");
return KS_STATUS_SUCCESS;
}
list_iterator_stop(&bs->connections);
list_clear(&bs->connections);
-
+
ks_log(KS_LOG_DEBUG, "Stopped\n");
return KS_STATUS_SUCCESS;
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;
}
}
}
+ if (list_size(&bs->connections) == 0) bs->ttl = ks_time_now() + (5 * KS_USEC_PER_SEC);
+
return ret;
}
// no connections available
return KS_STATUS_FAIL;
}
-
+
bc = blade_handle_connections_get(bs->handle, cid);
if (!bc) {
// @todo error logging... this shouldn't happen
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;
bs = (blade_session_t *)data;
while (!bs->shutdown) {
-
+
state = bs->state;
if (!list_empty(&bs->connections)) {
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);
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
projects / thirdparty / freeswitch.git / commitdiff
