core: More slight C/websocket progress.

core/core.js

@@ -641,6 +641,7 @@ async function getProcessBlob(blobId, key, options) {
 	}
 	return process;
 }
+exports.getProcessBlob = getProcessBlob;
 
 /**
  * Send any changed account information.

src/http.c

@@ -331,7 +331,7 @@ static void _http_add_body_bytes(tf_http_connection_t* connection, const void* d
 	{
 		if (size)
 		{
-			connection->body = tf_resize_vec(connection->body, connection->body_length + size);
+			connection->body = tf_resize_vec(connection->body, connection->body_length + size + 1);
 			memcpy((char*)connection->body + connection->body_length, data, size);
 			connection->body_length += size;
 		}
@@ -385,7 +385,7 @@ static void _http_add_body_bytes(tf_http_connection_t* connection, const void* d
 
 				if (!fin || connection->fragment_length)
 				{
-					connection->fragment = tf_resize_vec(connection->fragment, connection->fragment_length + length);
+					connection->fragment = tf_resize_vec(connection->fragment, connection->fragment_length + length + 1);
 					memcpy((uint8_t*)connection->fragment + connection->fragment_length, message, length);
 					connection->fragment_length += length;
 				}
@@ -394,10 +394,14 @@ static void _http_add_body_bytes(tf_http_connection_t* connection, const void* d
 				{
 					if (connection->request && connection->request->on_message)
 					{
+						uint8_t* payload = connection->fragment_length ? connection->fragment : message;
+						size_t payload_length = connection->fragment_length ? connection->fragment_length : length;
+						uint8_t backup = payload[payload_length];
+						payload[payload_length] = '\0';
 						tf_trace_begin(connection->http->trace, connection->trace_name ? connection->trace_name : "websocket");
-						connection->request->on_message(connection->request, connection->fragment_length ? connection->fragment_op_code : op_code,
+						connection->request->on_message(connection->request, connection->fragment_length ? connection->fragment_op_code : op_code, payload, payload_length);
-							connection->fragment_length ? connection->fragment : message, connection->fragment_length ? connection->fragment_length : length);
 						tf_trace_end(connection->http->trace);
+						payload[payload_length] = backup;
 					}
 					connection->fragment_length = 0;
 				}

src/httpd.app.c

@@ -222,6 +222,7 @@ typedef struct _app_t
 	JSValue credentials;
 
 	tf_taskstub_t* taskstub;
+	JSValue process;
 } app_t;
 
 static void _httpd_auth_query_work(tf_ssb_t* ssb, void* user_data)
@@ -230,6 +231,64 @@ static void _httpd_auth_query_work(tf_ssb_t* ssb, void* user_data)
 	work->settings = tf_ssb_db_get_property(ssb, "core", "settings");
 }
 
+static void _httpd_app_kill_task(app_t* work)
+{
+	if (work->taskstub)
+	{
+		JSContext* context = work->request->context;
+		JSValue result = tf_taskstub_kill(work->taskstub);
+		tf_util_report_error(context, result);
+		JS_FreeValue(context, result);
+		work->taskstub = NULL;
+	}
+}
+
+static void _httpd_app_message_hello(app_t* work, JSValue message)
+{
+	/* TODO */
+}
+
+static bool _httpd_app_message_call_client_api(app_t* work, JSValue message, const char* action_string)
+{
+	bool called = false;
+	JSContext* context = work->request->context;
+	JSValue client_api = JS_GetPropertyStr(context, work->process, "client_api");
+	JSValue callback = JS_GetPropertyStr(context, client_api, action_string);
+	if (!JS_IsUndefined(callback))
+	{
+		JSValue result = JS_Call(context, callback, JS_NULL, 1, &message);
+		tf_util_report_error(context, result);
+		JS_FreeValue(context, result);
+		called = true;
+	}
+	JS_FreeValue(context, callback);
+	JS_FreeValue(context, client_api);
+	return called;
+}
+
+static bool _httpd_app_message_call_message_handler(app_t* work, JSValue message)
+{
+	bool called = false;
+	JSContext* context = work->request->context;
+	JSValue event_handlers = JS_GetPropertyStr(context, work->process, "eventHandlers");
+	JSValue handler_array = JS_GetPropertyStr(context, event_handlers, "message");
+	if (!JS_IsUndefined(handler_array))
+	{
+		for (int i = 0; i < tf_util_get_length(context, handler_array); i++)
+		{
+			JSValue handler = JS_GetPropertyUint32(context, handler_array, i);
+			JSValue result = JS_Call(context, handler, JS_NULL, 1, &message);
+			tf_util_report_error(context, result);
+			JS_FreeValue(context, result);
+			JS_FreeValue(context, handler);
+			called = true;
+		}
+	}
+	JS_FreeValue(context, handler_array);
+	JS_FreeValue(context, event_handlers);
+	return called;
+}
+
 static void _httpd_app_on_message(tf_http_request_t* request, int op_code, const void* data, size_t size)
 {
 	tf_printf("REQUEST MESSAGE %.*s\n", (int)size, (const char*)data);
@@ -237,18 +296,42 @@ static void _httpd_app_on_message(tf_http_request_t* request, int op_code, const
 	JSContext* context = request->context;
 	switch (op_code)
 	{
+	/* TEXT */
 	case 0x1:
+	/* BINARY */
 	case 0x2:
+		{
+			char* copy = tf_malloc(size + 1);
+			memcpy(copy, data, size);
+			JSValue message = JS_ParseJSON(context, data, size, NULL);
+			if (JS_IsException(message) || !JS_IsObject(message))
+			{
+				tf_util_report_error(context, message);
+				_httpd_app_kill_task(work);
+				/* http close? */
+			}
+			else
+			{
+				JSValue action = JS_GetPropertyStr(context, message, "action");
+				const char* action_string = JS_ToCString(context, action);
+				if (action_string && !work->taskstub && strcmp(action_string, "hello") == 0)
+				{
+					_httpd_app_message_hello(work, message);
+				}
+				else if (!_httpd_app_message_call_client_api(work, message, action_string))
+				{
+					_httpd_app_message_call_message_handler(work, message);
+				}
+				JS_FreeCString(context, action_string);
+				JS_FreeValue(context, action);
+			}
+			JS_FreeValue(context, message);
+		}
 		break;
 	/* CLOSE */
 	case 0x8:
-		if (work->taskstub)
+		_httpd_app_kill_task(work);
-		{
+		tf_http_request_websocket_send(request, 0x8, data, tf_min(size, sizeof(uint16_t)));
-			JSValue result = tf_taskstub_kill(work->taskstub);
-			tf_util_report_error(context, result);
-			JS_FreeValue(context, result);
-			work->taskstub = NULL;
-		}
 		break;
 	/* PONG */
 	case 0xa:
@@ -258,10 +341,10 @@ static void _httpd_app_on_message(tf_http_request_t* request, int op_code, const
 
 static void _httpd_app_on_close(tf_http_request_t* request)
 {
-	tf_printf("REQUEST CLOSE\n");
 	JSContext* context = request->context;
 	app_t* work = request->user_data;
 	JS_FreeValue(context, work->credentials);
+	_httpd_app_kill_task(work);
 	tf_free(work);
 
 	tf_http_request_unref(request);