core: Implement websocket timeout in C.

src/http.c

@@ -955,6 +955,11 @@ void tf_http_request_websocket_send(tf_http_request_t* request, int op_code, con
 	tf_free(copy);
 }
 
+void tf_http_request_websocket_close(tf_http_request_t* request)
+{
+	_http_connection_destroy(request->connection, "websocket close");
+}
+
 void tf_http_respond(tf_http_request_t* request, int status, const char** headers, int headers_count, const void* body, size_t content_length)
 {
 	if (request->connection->is_response_sent)

src/http.h

@@ -210,6 +210,13 @@ const char* tf_http_get_cookie(const char* cookie_header, const char* name);
 */
 void tf_http_request_websocket_send(tf_http_request_t* request, int op_code, const void* data, size_t size);
 
+/**
+** Close a websocket.
+** @param request The HTTP request which was previously updated to a websocket
+** session with tf_http_request_websocket_upgrade().
+*/
+void tf_http_request_websocket_close(tf_http_request_t* request);
+
 /**
 ** Upgrade an HTTP request to a websocket session.
 ** @param request The HTTP request.

src/httpd.app.c

@@ -218,11 +218,14 @@ void tf_httpd_endpoint_app(tf_http_request_t* request)
 typedef struct _app_t
 {
 	tf_http_request_t* request;
+	uv_timer_t timer;
 	const char* settings;
 	JSValue opaque;
 	JSValue credentials;
-	tf_taskstub_t* taskstub;
 	JSValue process;
+	uint64_t last_ping_ms;
+	uint64_t last_active_ms;
+	bool got_hello;
 } app_t;
 
 static void _httpd_auth_query_work(tf_ssb_t* ssb, void* user_data)
@@ -233,13 +236,19 @@ static void _httpd_auth_query_work(tf_ssb_t* ssb, void* user_data)
 
 static void _httpd_app_kill_task(app_t* work)
 {
-	if (work->taskstub)
-	{
 	JSContext* context = work->request->context;
-		JSValue result = tf_taskstub_kill(work->taskstub);
+	if (!JS_IsUndefined(work->process))
+	{
+		JSValue task = JS_GetPropertyStr(context, work->process, "task");
+		if (!JS_IsUndefined(task))
+		{
+			JSValue kill = JS_GetPropertyStr(context, task, "kill");
+			JSValue result = JS_Call(context, kill, task, 0, NULL);
 			tf_util_report_error(context, result);
 			JS_FreeValue(context, result);
-		work->taskstub = NULL;
+			JS_FreeValue(context, kill);
+		}
+		JS_FreeValue(context, task);
 	}
 }
 
@@ -501,6 +510,7 @@ static void _httpd_app_message_hello(app_t* work, JSValue message)
 	tf_task_t* task = tf_task_get(context);
 	tf_ssb_t* ssb = tf_task_get_ssb(task);
 	tf_http_request_ref(work->request);
+	work->got_hello = true;
 
 	JSValue session = JS_IsObject(work->credentials) ? JS_GetPropertyStr(context, work->credentials, "session") : JS_UNDEFINED;
 	const char* user = tf_util_get_property_as_string(context, session, "name");
@@ -561,6 +571,8 @@ static void _httpd_app_on_message(tf_http_request_t* request, int op_code, const
 {
 	app_t* work = request->user_data;
 	JSContext* context = request->context;
+	tf_task_t* task = tf_task_get(context);
+	work->last_active_ms = uv_now(tf_task_get_loop(task));
 	switch (op_code)
 	{
 	/* TEXT */
@@ -572,14 +584,13 @@ static void _httpd_app_on_message(tf_http_request_t* request, int op_code, const
 			if (JS_IsException(message) || !JS_IsObject(message))
 			{
 				tf_util_report_error(context, message);
-				_httpd_app_kill_task(work);
-				/* http close? */
+				tf_http_request_websocket_close(request);
 			}
 			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)
+				if (action_string && !work->got_hello && strcmp(action_string, "hello") == 0)
 				{
 					_httpd_app_message_hello(work, message);
 				}
@@ -604,6 +615,13 @@ static void _httpd_app_on_message(tf_http_request_t* request, int op_code, const
 	}
 }
 
+static void _httpd_app_on_timer_close(uv_handle_t* handle)
+{
+	app_t* work = handle->data;
+	handle->data = NULL;
+	tf_free(work);
+}
+
 static void _httpd_app_on_close(tf_http_request_t* request)
 {
 	JSContext* context = request->context;
@@ -614,11 +632,31 @@ static void _httpd_app_on_close(tf_http_request_t* request)
 	JS_FreeValue(context, work->process);
 	JS_FreeValue(context, work->opaque);
 	work->process = JS_UNDEFINED;
-	tf_free(work);
-
+	uv_close((uv_handle_t*)&work->timer, _httpd_app_on_timer_close);
 	tf_http_request_unref(request);
 }
 
+static void _httpd_app_on_timer(uv_timer_t* timer)
+{
+	app_t* app = timer->data;
+	uint64_t now_ms = uv_now(timer->loop);
+	uint64_t repeat_ms = uv_timer_get_repeat(timer);
+	if (now_ms - app->last_active_ms < repeat_ms)
+	{
+		/* Active. */
+	}
+	else if (app->last_ping_ms > app->last_active_ms)
+	{
+		/* Timed out. */
+		tf_http_request_websocket_close(app->request);
+	}
+	else
+	{
+		tf_http_request_websocket_send(app->request, 0x9, NULL, 0);
+		app->last_ping_ms = now_ms;
+	}
+}
+
 static void _httpd_auth_query_after_work(tf_ssb_t* ssb, int status, void* user_data)
 {
 	app_t* work = user_data;
@@ -701,6 +739,8 @@ static void _httpd_auth_query_after_work(tf_ssb_t* ssb, int status, void* user_d
 	tf_http_request_websocket_upgrade(request);
 	tf_http_respond(request, 101, headers, headers_count, NULL, 0);
 
+	uv_timer_start(&work->timer, _httpd_app_on_timer, 6 * 1000, 6 * 1000);
+
 	/* What now? */
 	tf_free((void*)cookie);
 	JS_FreeCString(context, name_string);
@@ -740,7 +780,9 @@ static void _tf_httpd_endpoint_app_socket_c(tf_http_request_t* request)
 		*work = (app_t) {
 			.request = request,
 			.credentials = credentials,
+			.timer = { .data = work },
 		};
+		uv_timer_init(tf_ssb_get_loop(ssb), &work->timer);
 		tf_ssb_run_work(ssb, _httpd_auth_query_work, _httpd_auth_query_after_work, work);
 	}
 }