FS-9952: Initial implementation of a very basic text chat system which introduced a number of supporting subsystems

libs/libblade/test/Makefile.am

@@ -13,6 +13,11 @@ bladec_SOURCES = bladec.c tap.c
 bladec_CFLAGS = $(AM_CFLAGS)
 bladec_LDADD = $(TEST_LDADD)
 
+check_PROGRAMS += blades
+blades_SOURCES = blades.c tap.c
+blades_CFLAGS = $(AM_CFLAGS)
+blades_LDADD = $(TEST_LDADD)
+
 
 TESTS=$(check_PROGRAMS)
 

libs/libblade/test/bladec.c

@@ -26,22 +26,22 @@ struct command_def_s {
 	command_callback callback;
 };
 
-void command_test(blade_handle_t *bh, char *args);
 void command_quit(blade_handle_t *bh, char *args);
-void command_store(blade_handle_t *bh, char *args);
-void command_fetch(blade_handle_t *bh, char *args);
 void command_connect(blade_handle_t *bh, char *args);
+void command_chat(blade_handle_t *bh, char *args);
 
 static const struct command_def_s command_defs[] = {
-	{ "test", command_test },
 	{ "quit", command_quit },
-	{ "store", command_store },
-	{ "fetch", command_fetch },
 	{ "connect", command_connect },
-	
+	{ "chat", command_chat },
+
 	{ NULL, NULL }
 };
 
+ks_bool_t on_blade_chat_join_response(blade_response_t *bres);
+ks_bool_t on_blade_chat_message_event(blade_event_t *bev);
+void on_blade_session_state_callback(blade_session_t *bs, blade_session_state_condition_t condition, void *data);
+
 int main(int argc, char **argv)
 {
 	blade_handle_t *bh = NULL;
@@ -50,16 +50,16 @@ int main(int argc, char **argv)
 	blade_module_t *mod_wss = NULL;
 	//blade_identity_t *id = NULL;
 	const char *cfgpath = "bladec.cfg";
-	
+	const char *session_state_callback_id = NULL;
 
 	ks_global_set_default_logger(KS_LOG_LEVEL_DEBUG);
-	
+
 	blade_init();
 
 	blade_handle_create(&bh, NULL, NULL);
 
 	if (argc > 1) cfgpath = argv[1];
-	
+
 	config_init(&config);
 	if (!config_read_file(&config, cfgpath)) {
 		ks_log(KS_LOG_ERROR, "%s:%d - %s\n", config_error_file(&config), config_error_line(&config), config_error_text(&config));
@@ -76,7 +76,7 @@ int main(int argc, char **argv)
 		ks_log(KS_LOG_ERROR, "The 'blade' config setting is not a group\n");
 		return EXIT_FAILURE;
 	}
-	
+
 	if (blade_handle_startup(bh, config_blade) != KS_STATUS_SUCCESS) {
 		ks_log(KS_LOG_ERROR, "Blade startup failed\n");
 		return EXIT_FAILURE;
@@ -91,8 +91,13 @@ int main(int argc, char **argv)
 		return EXIT_FAILURE;
 	}
 
+	blade_handle_event_register(bh, "blade.chat.message", on_blade_chat_message_event);
+	blade_handle_session_state_callback_register(bh, NULL, on_blade_session_state_callback, &session_state_callback_id);
+
 	loop(bh);
 
+	blade_handle_session_state_callback_unregister(bh, session_state_callback_id);
+
 	blade_module_wss_on_shutdown(mod_wss);
 
 	blade_module_wss_on_unload(mod_wss);
@@ -104,7 +109,37 @@ int main(int argc, char **argv)
 	return 0;
 }
 
+ks_bool_t on_blade_chat_message_event(blade_event_t *bev)
+{
+	cJSON *res = NULL;
+	const char *from = NULL;
+	const char *message = NULL;
 
+	ks_assert(bev);
+
+	res = cJSON_GetObjectItem(bev->message, "result");
+	from = cJSON_GetObjectCstr(res, "from");
+	message = cJSON_GetObjectCstr(res, "message");
+
+	ks_log(KS_LOG_DEBUG, "Received Chat Message Event: (%s) %s\n", from, message);
+
+	return KS_FALSE;
+}
+
+void on_blade_session_state_callback(blade_session_t *bs, blade_session_state_condition_t condition, void *data)
+{
+	blade_session_state_t state = blade_session_state_get(bs);
+
+	if (condition == BLADE_SESSION_STATE_CONDITION_PRE) {
+		ks_log(KS_LOG_DEBUG, "Blade Session State Changed: %s, %d\n", blade_session_id_get(bs), state);
+		if (state == BLADE_SESSION_STATE_READY) {
+			cJSON *req = NULL;
+			blade_rpc_request_create(blade_session_pool_get(bs), &req, NULL, NULL, "blade.chat.join");
+			blade_session_send(bs, req, on_blade_chat_join_response);
+			cJSON_Delete(req);
+		}
+	}
+}
 
 void buffer_console_input(void)
 {
@@ -162,7 +197,7 @@ void loop(blade_handle_t *bh)
 void parse_argument(char **input, char **arg, char terminator)
 {
 	char *tmp;
-	
+
 	ks_assert(input);
 	ks_assert(*input);
 	ks_assert(arg);
@@ -183,11 +218,11 @@ void process_console_input(blade_handle_t *bh, char *line)
 	char *args = line;
 	char *cmd = NULL;
 	ks_bool_t found = KS_FALSE;
-	
+
 	ks_log(KS_LOG_DEBUG, "Output: %s\n", line);
 
 	parse_argument(&args, &cmd, ' ');
-	
+
 	ks_log(KS_LOG_DEBUG, "Command: %s, Args: %s\n", cmd, args);
 
 	for (int32_t index = 0; command_defs[index].cmd; ++index) {
@@ -199,61 +234,78 @@ void process_console_input(blade_handle_t *bh, char *line)
 	if (!found) ks_log(KS_LOG_INFO, "Command '%s' unknown.\n", cmd);
 }
 
-void command_test(blade_handle_t *bh, char *args)
-{
-	ks_log(KS_LOG_DEBUG, "Hello World!\n");
-}
-
 void command_quit(blade_handle_t *bh, char *args)
 {
 	ks_assert(bh);
 	ks_assert(args);
-	
+
 	ks_log(KS_LOG_DEBUG, "Shutting down\n");
 	g_shutdown = KS_TRUE;
 }
 
-void command_store(blade_handle_t *bh, char *args)
-{
-	char *key;
-	char *data;
-
-	ks_assert(args);
-
-	parse_argument(&args, &key, ' ');
-	parse_argument(&args, &data, ' ');
-
-	blade_handle_datastore_store(bh, key, strlen(key), data, strlen(data) + 1);
-}
-
-ks_bool_t blade_datastore_fetch_callback(blade_datastore_t *bds, const void *data, uint32_t data_length, void *userdata)
-{
-	ks_log(KS_LOG_INFO, "%s\n", data);
-	return KS_TRUE;
-}
-
-void command_fetch(blade_handle_t *bh, char *args)
-{
-	char *key;
-
-	ks_assert(args);
-
-	parse_argument(&args, &key, ' ');
-
-	blade_handle_datastore_fetch(bh, blade_datastore_fetch_callback, key, strlen(key), bh);
-}
-
 void command_connect(blade_handle_t *bh, char *args)
 {
 	blade_connection_t *bc = NULL;
 	blade_identity_t *target = NULL;
-	
+
 	ks_assert(bh);
 	ks_assert(args);
 
 	blade_identity_create(&target, blade_handle_pool_get(bh));
-	
+
 	if (blade_identity_parse(target, args) == KS_STATUS_SUCCESS) blade_handle_connect(bh, &bc, target, NULL);
 
 	blade_identity_destroy(&target);
 }
+
+ks_bool_t on_blade_chat_join_response(blade_response_t *bres) // @todo this should get userdata passed in from when the callback is registered
+{
+	ks_log(KS_LOG_DEBUG, "Received Chat Join Response!\n");
+	return KS_FALSE;
+}
+
+ks_bool_t on_blade_chat_send_response(blade_response_t *bres) // @todo this should get userdata passed in from when the callback is registered
+{
+	ks_log(KS_LOG_DEBUG, "Received Chat Send Response!\n");
+	return KS_FALSE;
+}
+
+void command_chat(blade_handle_t *bh, char *args)
+{
+	char *cmd = NULL;
+
+	ks_assert(bh);
+	ks_assert(args);
+
+	parse_argument(&args, &cmd, ' ');
+	ks_log(KS_LOG_DEBUG, "Chat Command: %s, Args: %s\n", cmd, args);
+
+	if (!strcmp(cmd, "leave")) {
+	} else if (!strcmp(cmd, "send")) {
+		char *sid = NULL;
+		blade_session_t *bs = NULL;
+		cJSON *req = NULL;
+		cJSON *params = NULL;
+
+		parse_argument(&args, &sid, ' ');
+
+		bs = blade_handle_sessions_get(bh, sid);
+		if (!bs) {
+			ks_log(KS_LOG_DEBUG, "Unknown Session: %s\n", sid);
+			return;
+		}
+		blade_rpc_request_create(blade_handle_pool_get(bh), &req, &params, NULL, "blade.chat.send");
+		ks_assert(req);
+		ks_assert(params);
+
+		cJSON_AddStringToObject(params, "message", args);
+
+		blade_session_send(bs, req, on_blade_chat_send_response);
+
+		blade_session_read_unlock(bs);
+
+		cJSON_Delete(req);
+	} else {
+		ks_log(KS_LOG_DEBUG, "Unknown Chat Command: %s\n", cmd);
+	}
+}

libs/libblade/test/bladec.cfg

@@ -1,9 +1,6 @@
 blade:
 {
-    identity = "directory@domain";
-    directory:
-    {
-    };
+    identity = "peer@domain";
     datastore:
     {
         database:
@@ -11,18 +8,4 @@ blade:
             path = ":mem:";
         };
     };
-    wss:
-    {
-        endpoints:
-        {
-            ipv4 = ( { address = "0.0.0.0", port = 2100 } );
-            ipv6 = ( { address = "::", port = 2100 } );
-            backlog = 128;
-        };
-        # SSL group is optional, disabled when absent
-        ssl:
-        {
-            # todo: server SSL stuffs here
-        };
-    };
 };

libs/libblade/test/bladec2.cfg

@@ -1,11 +0,0 @@
-blade:
-{
-    identity = "peer@domain";
-    datastore:
-    {
-        database:
-        {
-            path = ":mem:";
-        };
-    };
-};

libs/libblade/test/blades.c

@@ -0,0 +1,207 @@
+#include "blade.h"
+#include "tap.h"
+
+#ifdef _WIN32
+#define STDIO_FD(_fs) _fileno(_fs)
+#define READ(_fd, _buffer, _count) _read(_fd, _buffer, _count)
+#else
+#define STDIO_FD(_fs) fileno(_fs)
+#define READ(_fd, _buffer, _count) read(_fd, _buffer, _count)
+#endif
+
+#define CONSOLE_INPUT_MAX 512
+
+ks_bool_t g_shutdown = KS_FALSE;
+char g_console_input[CONSOLE_INPUT_MAX];
+size_t g_console_input_length = 0;
+size_t g_console_input_eol = 0;
+
+void loop(blade_handle_t *bh);
+void process_console_input(blade_handle_t *bh, char *line);
+
+typedef void (*command_callback)(blade_handle_t *bh, char *args);
+
+struct command_def_s {
+	const char *cmd;
+	command_callback callback;
+};
+
+void command_quit(blade_handle_t *bh, char *args);
+
+static const struct command_def_s command_defs[] = {
+	{ "quit", command_quit },
+
+	{ NULL, NULL }
+};
+
+int main(int argc, char **argv)
+{
+	blade_handle_t *bh = NULL;
+	config_t config;
+	config_setting_t *config_blade = NULL;
+	blade_module_t *mod_wss = NULL;
+	blade_module_t *mod_chat = NULL;
+	//blade_identity_t *id = NULL;
+	const char *cfgpath = "blades.cfg";
+
+	ks_global_set_default_logger(KS_LOG_LEVEL_DEBUG);
+
+	blade_init();
+
+	blade_handle_create(&bh, NULL, NULL);
+
+	if (argc > 1) cfgpath = argv[1];
+
+	config_init(&config);
+	if (!config_read_file(&config, cfgpath)) {
+		ks_log(KS_LOG_ERROR, "%s:%d - %s\n", config_error_file(&config), config_error_line(&config), config_error_text(&config));
+		config_destroy(&config);
+		return EXIT_FAILURE;
+	}
+	config_blade = config_lookup(&config, "blade");
+	if (!config_blade) {
+		ks_log(KS_LOG_ERROR, "Missing 'blade' config group\n");
+		config_destroy(&config);
+		return EXIT_FAILURE;
+	}
+	if (config_setting_type(config_blade) != CONFIG_TYPE_GROUP) {
+		ks_log(KS_LOG_ERROR, "The 'blade' config setting is not a group\n");
+		return EXIT_FAILURE;
+	}
+
+	if (blade_handle_startup(bh, config_blade) != KS_STATUS_SUCCESS) {
+		ks_log(KS_LOG_ERROR, "Blade startup failed\n");
+		return EXIT_FAILURE;
+	}
+
+	if (blade_module_wss_on_load(&mod_wss, bh) != KS_STATUS_SUCCESS) {
+		ks_log(KS_LOG_ERROR, "Blade WSS module load failed\n");
+		return EXIT_FAILURE;
+	}
+	if (blade_module_wss_on_startup(mod_wss, config_blade) != KS_STATUS_SUCCESS) {
+		ks_log(KS_LOG_ERROR, "Blade WSS module startup failed\n");
+		return EXIT_FAILURE;
+	}
+
+	blade_module_chat_on_load(&mod_chat, bh);
+	blade_module_chat_on_startup(mod_chat, config_blade);
+
+	loop(bh);
+
+	blade_module_chat_on_shutdown(mod_chat);
+	blade_module_chat_on_unload(mod_chat);
+
+	blade_module_wss_on_shutdown(mod_wss);
+
+	blade_module_wss_on_unload(mod_wss);
+
+	blade_handle_destroy(&bh);
+
+	blade_shutdown();
+
+	return 0;
+}
+
+
+
+void buffer_console_input(void)
+{
+	ssize_t bytes = 0;
+	struct pollfd poll[1];
+	poll[0].fd = STDIO_FD(stdin);
+	poll[0].events = POLLIN | POLLERR;
+
+	if (ks_poll(poll, 1, 1) > 0) {
+		if (poll[0].revents & POLLIN) {
+			if ((bytes = READ(poll[0].fd, g_console_input + g_console_input_length, CONSOLE_INPUT_MAX - g_console_input_length)) <= 0) {
+				// @todo error
+				return;
+			}
+			g_console_input_length += bytes;
+		}
+	}
+}
+
+void loop(blade_handle_t *bh)
+{
+	while (!g_shutdown) {
+		ks_bool_t eol = KS_FALSE;
+		buffer_console_input();
+
+		for (; g_console_input_eol < g_console_input_length; ++g_console_input_eol) {
+			char c = g_console_input[g_console_input_eol];
+			if (c == '\r' || c == '\n') {
+				eol = KS_TRUE;
+				break;
+			}
+		}
+		if (eol) {
+			g_console_input[g_console_input_eol] = '\0';
+			process_console_input(bh, g_console_input);
+			g_console_input_eol++;
+			for (; g_console_input_eol < g_console_input_length; ++g_console_input_eol) {
+				char c = g_console_input[g_console_input_eol];
+				if (c != '\r' && c != '\n') break;
+			}
+			if (g_console_input_eol == g_console_input_length) g_console_input_eol = g_console_input_length = 0;
+			else {
+				memcpy(g_console_input, g_console_input + g_console_input_eol, g_console_input_length - g_console_input_eol);
+				g_console_input_length -= g_console_input_eol;
+				g_console_input_eol = 0;
+			}
+		}
+		if (g_console_input_length == CONSOLE_INPUT_MAX) {
+			// @todo lines must not exceed 512 bytes, treat as error and ignore buffer until next new line?
+			ks_assert(0);
+		}
+	}
+}
+
+void parse_argument(char **input, char **arg, char terminator)
+{
+	char *tmp;
+
+	ks_assert(input);
+	ks_assert(*input);
+	ks_assert(arg);
+
+	tmp = *input;
+	*arg = tmp;
+
+	while (*tmp && *tmp != terminator) ++tmp;
+	if (*tmp == terminator) {
+		*tmp = '\0';
+		++tmp;
+	}
+	*input = tmp;
+}
+
+void process_console_input(blade_handle_t *bh, char *line)
+{
+	char *args = line;
+	char *cmd = NULL;
+	ks_bool_t found = KS_FALSE;
+
+	ks_log(KS_LOG_DEBUG, "Output: %s\n", line);
+
+	parse_argument(&args, &cmd, ' ');
+
+	ks_log(KS_LOG_DEBUG, "Command: %s, Args: %s\n", cmd, args);
+
+	for (int32_t index = 0; command_defs[index].cmd; ++index) {
+		if (!strcmp(command_defs[index].cmd, cmd)) {
+			found = KS_TRUE;
+			command_defs[index].callback(bh, args);
+		}
+	}
+	if (!found) ks_log(KS_LOG_INFO, "Command '%s' unknown.\n", cmd);
+}
+
+void command_quit(blade_handle_t *bh, char *args)
+{
+	ks_assert(bh);
+	ks_assert(args);
+
+	ks_log(KS_LOG_DEBUG, "Shutting down\n");
+	g_shutdown = KS_TRUE;
+}

libs/libblade/test/blades.cfg

@@ -0,0 +1,28 @@
+blade:
+{
+    identity = "service@domain";
+    directory:
+    {
+    };
+    datastore:
+    {
+        database:
+        {
+            path = ":mem:";
+        };
+    };
+    wss:
+    {
+        endpoints:
+        {
+            ipv4 = ( { address = "0.0.0.0", port = 2100 } );
+            ipv6 = ( { address = "::", port = 2100 } );
+            backlog = 128;
+        };
+        # SSL group is optional, disabled when absent
+        ssl:
+        {
+            # todo: server SSL stuffs here
+        };
+    };
+};