Rooms JS => C.

git-svn-id: https://www.unprompted.com/svn/projects/tildefriends/trunk@4107 ed5197a5-7fde-0310-b194-c3ffbd925b24

core/core.js

@@ -17,6 +17,11 @@ const k_global_settings = {
 		default_value: true,
 		description: 'Whether this instance should behave as a room.',
 	},
+	room_name: {
+		type: 'string',
+		default_value: 'tilde friends tunnel',
+		description: 'Name of the room.',
+	},
 	code_of_conduct: {
 		type: 'textarea',
 		default_value: undefined,

core/ssb.js

@@ -1,7 +1,6 @@
 "use strict";
 var g_wants_requests = {};
 var g_database = new Database('core');
-let g_attendants = {};
 const k_use_create_history_stream = false;
 
 function get_latest_sequence_for_author(author) {
@@ -24,27 +23,6 @@ function storeMessage(message) {
 	}
 }
 
-function tunnel_attendants(request) {
-	if (request.message.type !== 'state') {
-		throw Error('Unexpected type: ' + request.message.type);
-	}
-	let state = new Set(request.message.ids);
-	for (let id of state) {
-		request.add_room_attendant(id);
-	}
-	request.more(function attendants(message) {
-		if (message.message.type === 'joined') {
-			request.add_room_attendant(message.message.id);
-			state.add(message.message.id);
-		} else if (message.message.type === 'left') {
-			request.remove_room_attendant(message.message.id);
-			state.delete(message.message.id);
-		} else {
-			throw Error('Unexpected type: ' + message.type);
-		}
-	});
-}
-
 ssb.addEventListener('connections', function on_connections_changed(change, connection) {
 	if (change == 'add') {
 		var sequence = get_latest_sequence_for_author(connection.id);
@@ -63,52 +41,16 @@ ssb.addEventListener('connections', function on_connections_changed(change, conn
 		} else {
 			if (connection.is_client) {
 				connection.send_json({"name": ["ebt", "replicate"], "args": [{"version": 3, "format": "classic"}], "type": "duplex"}, ebtReplicateClient);
-
-				connection.send_json_async({'name': ['tunnel', 'isRoom'], 'args': []}, function tunnel_is_room(request) {
-					if (request.message) {
-						request.connection.send_json({'name': ['room', 'attendants'], 'args': [], 'type': 'source'}, tunnel_attendants);
-					}
-				});
 			}
 		}
 	} else if (change == 'remove') {
 		print('REMOVE', connection.id);
-		notify_attendant_changed(connection.id, 'left');
-		delete g_attendants[connection.id];
 		delete g_wants_requests[connection.id];
 	} else {
 		print('CHANGE', change);
 	}
 });
 
-function notify_attendant_changed(id, type) {
-	if (!id) {
-		print(`notify_attendant_changed called with id=${id}`);
-		return;
-	}
-	for (let r of Object.values(g_attendants)) {
-		try {
-			r.send_json({
-				type: type,
-				id: id,
-			});
-		} catch (e) {
-			print(`Removing ${id} from g_attendants in ${type}.`, e);
-			delete g_attendants[id];
-		}
-	}
-}
-
-ssb.addRpc(['room', 'attendants'], function(request) {
-	let ids = Object.keys(g_attendants).sort();
-	request.send_json({
-		type: 'state',
-		ids: ids,
-	});
-	notify_attendant_changed(request.connection.id, 'joined');
-	g_attendants[request.connection.id] = request;
-});
-
 function ebtReplicateSendClock(request, have) {
 	var identities = ssb.getAllIdentities();
 	var message = {};

src/ssb.c

@@ -208,6 +208,8 @@ typedef struct _tf_ssb_connection_t
 	int port;
 
 	tf_ssb_state_t state;
+	bool is_attendant;
+	int32_t attendant_request_number;
 
 	uint8_t epub[crypto_box_PUBLICKEYBYTES];
 	uint8_t epriv[crypto_box_SECRETKEYBYTES];
@@ -516,6 +518,7 @@ static void _tf_ssb_connection_remove_request(tf_ssb_connection_t* connection, i
 
 void tf_ssb_connection_rpc_send(tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number, const uint8_t* message, size_t size, tf_ssb_rpc_callback_t* callback, tf_ssb_callback_cleanup_t* cleanup, void* user_data)
 {
+	printf("SEND %p\n", connection);
 	if (!connection)
 	{
 		return;
@@ -540,6 +543,50 @@ void tf_ssb_connection_rpc_send(tf_ssb_connection_t* connection, uint8_t flags,
 	connection->ssb->rpc_out++;
 }
 
+void tf_ssb_connection_rpc_send_json(tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number, JSValue message, tf_ssb_rpc_callback_t* callback, tf_ssb_callback_cleanup_t* cleanup, void* user_data)
+{
+	JSContext* context = connection->ssb->context;
+	JSValue json = JS_JSONStringify(context, message, JS_NULL, JS_NULL);
+	size_t size = 0;
+	const char* json_string = JS_ToCStringLen(context, &size, json);
+	tf_ssb_connection_rpc_send(
+		connection,
+		k_ssb_rpc_flag_json | (flags & k_ssb_rpc_flag_stream),
+		request_number,
+		(const uint8_t*)json_string,
+		size,
+		callback,
+		cleanup,
+		user_data);
+	JS_FreeCString(context, json_string);
+	JS_FreeValue(context, json);
+}
+
+void tf_ssb_connection_rpc_send_error(tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number, const char* error)
+{
+	JSContext* context = connection->ssb->context;
+	JSValue message = JS_NewObject(context);
+	JS_SetPropertyStr(context, message, "name", JS_NewString(context, "Error"));
+	JS_SetPropertyStr(context, message, "stack", JS_NewString(context, "none"));
+	JS_SetPropertyStr(context, message, "message", JS_NewString(context, error));
+	tf_ssb_connection_rpc_send_json(connection, flags, request_number, message, NULL, NULL, NULL);
+	JS_FreeValue(context, message);
+}
+
+void tf_ssb_connection_rpc_send_error_method_not_allowed(tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number)
+{
+	const char* k_unsupported = "{\"message\": \"method: is not in list of allowed methods\", \"name\": \"Error\", \"stack\": \"none\"}";
+	tf_ssb_connection_rpc_send(
+		connection,
+		k_ssb_rpc_flag_json | k_ssb_rpc_flag_end_error | (flags & k_ssb_rpc_flag_stream),
+		request_number,
+		(const uint8_t*)k_unsupported,
+		strlen(k_unsupported),
+		NULL,
+		NULL,
+		NULL);
+}
+
 static int _utf8_len(uint8_t ch)
 {
 	static const uint8_t k_length[] =
@@ -875,6 +922,11 @@ static void _tf_ssb_connection_verify_identity(tf_ssb_connection_t* connection,
 	_tf_ssb_notify_connections_changed(connection->ssb, k_tf_ssb_change_connect, connection);
 }
 
+bool tf_ssb_connection_is_client(tf_ssb_connection_t* connection)
+{
+	return connection->state == k_tf_ssb_state_verified;
+}
+
 const char* tf_ssb_connection_get_host(tf_ssb_connection_t* connection)
 {
 	return connection->host;
@@ -1201,11 +1253,9 @@ static void _tf_ssb_connection_rpc_recv(tf_ssb_connection_t* connection, uint8_t
 						callback(connection, flags, request_number, val, NULL, 0, user_data);
 					}
 				}
-				else
+				else if (!_tf_ssb_name_equals(context, val, (const char*[]) { "Error", NULL }))
 				{
-					const char* k_unsupported = "{\"message\": \"method: is not in list of allowed methods\", \"name\": \"Error\", \"stack\": \"none\"}";
+					tf_ssb_connection_rpc_send_error_method_not_allowed(connection, flags, -request_number);
-					tf_ssb_connection_rpc_send(connection, k_ssb_rpc_flag_json | k_ssb_rpc_flag_end_error | (flags & k_ssb_rpc_flag_stream), -request_number,
-						(const uint8_t*)k_unsupported, strlen(k_unsupported), NULL, NULL, NULL);
 				}
 			}
 		}
@@ -1417,6 +1467,7 @@ void tf_ssb_append_message(tf_ssb_t* ssb, JSValue message)
 
 void _tf_ssb_connection_destroy(tf_ssb_connection_t* connection, const char* reason)
 {
+	printf("DESTROY %p\n", connection);
 	tf_ssb_t* ssb = connection->ssb;
 	if (!connection->destroy_reason)
 	{
@@ -1437,7 +1488,7 @@ void _tf_ssb_connection_destroy(tf_ssb_connection_t* connection, const char* rea
 	}
 	for (tf_ssb_connection_t** it = &connection->ssb->connections; *it; it = &(*it)->next)
 	{
-		for (int i = 0; i < (*it)->requests_count; i++)
+		for (int i = (*it)->requests_count - 1; i >= 0; i--)
 		{
 			if ((*it)->requests[i].dependent_connection == connection)
 			{
@@ -2929,6 +2980,48 @@ void tf_ssb_connection_remove_room_attendant(tf_ssb_connection_t* connection, co
 	}
 }
 
+bool tf_ssb_connection_is_attendant(tf_ssb_connection_t* connection)
+{
+	return connection->is_attendant;
+}
+
+int32_t tf_ssb_connection_get_attendant_request_number(tf_ssb_connection_t* connection)
+{
+	return connection->attendant_request_number;
+}
+
+void tf_ssb_connection_set_attendant(tf_ssb_connection_t* connection, bool attendant, int request_number)
+{
+	connection->is_attendant = attendant;
+	connection->attendant_request_number = request_number;
+	_tf_ssb_notify_broadcasts_changed(connection->ssb);
+}
+
+void tf_ssb_connection_clear_room_attendants(tf_ssb_connection_t* connection)
+{
+	int modified = 0;
+	for (tf_ssb_broadcast_t** it = &connection->ssb->broadcasts; *it;)
+	{
+		if ((*it)->tunnel_connection == connection)
+		{
+			tf_ssb_broadcast_t* node = *it;
+			*it = node->next;
+			tf_free(node);
+			connection->ssb->broadcasts_count--;
+			modified++;
+		}
+		else
+		{
+			it = &(*it)->next;
+		}
+	}
+
+	if (modified)
+	{
+		_tf_ssb_notify_broadcasts_changed(connection->ssb);
+	}
+}
+
 tf_ssb_blob_wants_t* tf_ssb_connection_get_blob_wants_state(tf_ssb_connection_t* connection)
 {
 	return connection ? &connection->blob_wants : NULL;

src/ssb.h

@@ -105,6 +105,7 @@ bool tf_ssb_id_bin_to_str(char* str, size_t str_size, const uint8_t* bin);
 bool tf_ssb_verify_and_strip_signature(JSContext* context, JSValue val, char* out_id, size_t out_id_size, char* out_signature, size_t out_signature_size, bool* out_sequence_before_author);
 void tf_ssb_calculate_message_id(JSContext* context, JSValue message, char* out_id, size_t out_id_size);
 
+bool tf_ssb_connection_is_client(tf_ssb_connection_t* connection);
 const char* tf_ssb_connection_get_host(tf_ssb_connection_t* connection);
 int tf_ssb_connection_get_port(tf_ssb_connection_t* connection);
 tf_ssb_t* tf_ssb_connection_get_ssb(tf_ssb_connection_t* connection);
@@ -143,8 +144,15 @@ void tf_ssb_add_rpc_callback(tf_ssb_t* ssb, const char** name, tf_ssb_rpc_callba
 void tf_ssb_remove_rpc_callback(tf_ssb_t* ssb, const char** name, tf_ssb_rpc_callback_t* callback, void* user_data);
 
 void tf_ssb_connection_rpc_send(tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number, const uint8_t* message, size_t size, tf_ssb_rpc_callback_t* callback, tf_ssb_callback_cleanup_t* cleanup, void* user_data);
+void tf_ssb_connection_rpc_send_json(tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number, JSValue message, tf_ssb_rpc_callback_t* callback, tf_ssb_callback_cleanup_t* cleanup, void* user_data);
+void tf_ssb_connection_rpc_send_error(tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number, const char* error);
+void tf_ssb_connection_rpc_send_error_method_not_allowed(tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number);
 void tf_ssb_connection_add_request(tf_ssb_connection_t* connection, int32_t request_number, tf_ssb_rpc_callback_t* callback, tf_ssb_callback_cleanup_t* cleanup, void* user_data, tf_ssb_connection_t* dependent_connection);
 
+bool tf_ssb_connection_is_attendant(tf_ssb_connection_t* connection);
+int32_t tf_ssb_connection_get_attendant_request_number(tf_ssb_connection_t* connection);
+void tf_ssb_connection_set_attendant(tf_ssb_connection_t* connection, bool attendant, int request_number);
+void tf_ssb_connection_clear_room_attendants(tf_ssb_connection_t* connection);
 void tf_ssb_connection_add_room_attendant(tf_ssb_connection_t* connection, const char* id);
 void tf_ssb_connection_remove_room_attendant(tf_ssb_connection_t* connection, const char* id);
 

src/ssb.js.c

@@ -554,28 +554,6 @@ static JSValue _tf_ssb_rpc_send_binary(JSContext* context, JSValueConst this_val
 	return JS_UNDEFINED;
 }
 
-static JSValue _tf_ssb_rpc_add_room_attendant(JSContext* context, JSValueConst this_val, int argc, JSValueConst* argv)
-{
-	JSValue connection_val = JS_GetPropertyStr(context, this_val, "connection");
-	tf_ssb_connection_t* connection = JS_GetOpaque(connection_val, tf_ssb_get_connection_class_id());
-	const char* id = JS_ToCString(context, argv[0]);
-	tf_ssb_connection_add_room_attendant(connection, id);
-	JS_FreeCString(context, id);
-	JS_FreeValue(context, connection_val);
-	return JS_UNDEFINED;
-}
-
-static JSValue _tf_ssb_rpc_remove_room_attendant(JSContext* context, JSValueConst this_val, int argc, JSValueConst* argv)
-{
-	JSValue connection_val = JS_GetPropertyStr(context, this_val, "connection");
-	tf_ssb_connection_t* connection = JS_GetOpaque(connection_val, tf_ssb_get_connection_class_id());
-	const char* id = JS_ToCString(context, argv[0]);
-	tf_ssb_connection_remove_room_attendant(connection, id);
-	JS_FreeCString(context, id);
-	JS_FreeValue(context, connection_val);
-	return JS_UNDEFINED;
-}
-
 void _tf_ssb_on_rpc(tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number, JSValue args, const uint8_t* message, size_t size, void* user_data)
 {
 	tf_ssb_t* ssb = tf_ssb_connection_get_ssb(connection);
@@ -592,8 +570,6 @@ void _tf_ssb_on_rpc(tf_ssb_connection_t* connection, uint8_t flags, int32_t requ
 	JS_SetPropertyStr(context, object, "send_binary", JS_NewCFunction(context, _tf_ssb_rpc_send_binary, "send_binary", 1));
 	JS_SetPropertyStr(context, object, "send_json_end", JS_NewCFunction(context, _tf_ssb_rpc_send_json_end, "send_json_end", 1));
 	JS_SetPropertyStr(context, object, "more", JS_NewCFunction(context, _tf_ssb_rpc_more, "more", 1));
-	JS_SetPropertyStr(context, object, "add_room_attendant", JS_NewCFunction(context, _tf_ssb_rpc_add_room_attendant, "add_room_attendant", 1));
-	JS_SetPropertyStr(context, object, "remove_room_attendant", JS_NewCFunction(context, _tf_ssb_rpc_remove_room_attendant, "remove_room_attendant", 1));
 
 	JSValue result = JS_Call(context, callback, JS_UNDEFINED, 1, &object);
 	tf_util_report_error(context, result);

src/ssb.rpc.c

@@ -11,6 +11,10 @@
 #include <string.h>
 #include <time.h>
 
+#if !defined(_countof)
+#define _countof(a) ((int)(sizeof((a)) / sizeof(*(a))))
+#endif
+
 static void _tf_ssb_rpc_gossip_ping(tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number, JSValue args, const uint8_t* message, size_t size, void* user_data)
 {
 	char buffer[256];
@@ -52,7 +56,6 @@ static void _tf_ssb_rpc_blobs_get(tf_ssb_connection_t* connection, uint8_t flags
 		const int k_send_max = 8192;
 		if (tf_ssb_db_blob_get(ssb, id, &blob, &size))
 		{
-			printf("sending %s (%zd)\n", id, size);
 			for (size_t offset = 0; offset < size; offset += k_send_max)
 			{
 				tf_ssb_connection_rpc_send(
@@ -178,9 +181,65 @@ void _tf_ssb_rpc_tunnel_cleanup(tf_ssb_t* ssb, void* user_data)
 	tf_free(user_data);
 }
 
+static bool _get_global_setting_bool(tf_ssb_t* ssb, const char* name, bool default_value)
+{
+	bool result = default_value;
+	JSContext* context = tf_ssb_get_context(ssb);
+	sqlite3* db = tf_ssb_get_db(ssb);
+	sqlite3_stmt* statement;
+	if (sqlite3_prepare(db, "SELECT value FROM properties WHERE id = 'core' AND key = 'settings'", -1, &statement, NULL) == SQLITE_OK)
+	{
+		if (sqlite3_step(statement) == SQLITE_ROW)
+		{
+			JSValue value = JS_ParseJSON(context, (const char*)sqlite3_column_text(statement, 0), sqlite3_column_bytes(statement, 0), NULL);
+			JSValue property = JS_GetPropertyStr(context, value,  name);
+			if (JS_IsBool(property))
+			{
+				result = JS_ToBool(context, property);
+			}
+			JS_FreeValue(context, property);
+			JS_FreeValue(context, value);
+		}
+		sqlite3_finalize(statement);
+	}
+	return result;
+}
+
+static bool _get_global_setting_string(tf_ssb_t* ssb, const char* name, char* out_value, size_t size)
+{
+	bool result = false;
+	JSContext* context = tf_ssb_get_context(ssb);
+	sqlite3* db = tf_ssb_get_db(ssb);
+	sqlite3_stmt* statement;
+	if (sqlite3_prepare(db, "SELECT value FROM properties WHERE id = 'core' AND key = 'settings'", -1, &statement, NULL) == SQLITE_OK)
+	{
+		if (sqlite3_step(statement) == SQLITE_ROW)
+		{
+			JSValue value = JS_ParseJSON(context, (const char*)sqlite3_column_text(statement, 0), sqlite3_column_bytes(statement, 0), NULL);
+			JSValue property = JS_GetPropertyStr(context, value, name);
+			const char* value_string = JS_ToCString(context, property);
+			if (value_string)
+			{
+				snprintf(out_value, size, "%s", value_string);
+				result = true;
+			}
+			JS_FreeValue(context, property);
+			JS_FreeValue(context, value);
+		}
+		sqlite3_finalize(statement);
+	}
+	return result;
+}
+
 static void _tf_ssb_rpc_tunnel_connect(tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number, JSValue args, const uint8_t* message, size_t size, void* user_data)
 {
 	tf_ssb_t* ssb = tf_ssb_connection_get_ssb(connection);
+	if (!_get_global_setting_bool(ssb, "room", true))
+	{
+		tf_ssb_connection_rpc_send_error_method_not_allowed(connection, flags, -request_number);
+		return;
+	}
+
 	JSContext* context = tf_ssb_connection_get_context(connection);
 	JSValue arg_array = JS_GetPropertyStr(context, args, "args");
 	JSValue arg = JS_GetPropertyUint32(context, arg_array, 0);
@@ -241,8 +300,9 @@ static void _tf_ssb_rpc_tunnel_connect(tf_ssb_connection_t* connection, uint8_t
 			.connection = connection,
 			.request_number = -request_number,
 		};
-		tf_ssb_connection_add_request(connection, -request_number, _tf_ssb_rpc_tunnel_callback, _tf_ssb_rpc_tunnel_cleanup, data0, connection);
+		printf("MAKE TUNNEL %p %p\n", connection, target_connection);
-		tf_ssb_connection_add_request(target_connection, tunnel_request_number, _tf_ssb_rpc_tunnel_callback, _tf_ssb_rpc_tunnel_cleanup, data1, target_connection);
+		tf_ssb_connection_add_request(connection, -request_number, _tf_ssb_rpc_tunnel_callback, _tf_ssb_rpc_tunnel_cleanup, data0, target_connection);
+		tf_ssb_connection_add_request(target_connection, tunnel_request_number, _tf_ssb_rpc_tunnel_callback, _tf_ssb_rpc_tunnel_cleanup, data1, connection);
 
 		JS_FreeValue(context, message_val);
 		JS_FreeValue(context, message);
@@ -256,6 +316,7 @@ static void _tf_ssb_rpc_tunnel_connect(tf_ssb_connection_t* connection, uint8_t
 		const char* origin_str = JS_ToCString(context, origin);
 		const char* portal_str = JS_ToCString(context, portal);
 		const char* target_str = JS_ToCString(context, target);
+		printf("TUNNEL CREATE\n");
 		tf_ssb_connection_tunnel_create(ssb, portal_str, -request_number, origin_str);
 		JS_FreeCString(context, origin_str);
 		JS_FreeCString(context, portal_str);
@@ -269,30 +330,6 @@ static void _tf_ssb_rpc_tunnel_connect(tf_ssb_connection_t* connection, uint8_t
 	JS_FreeValue(context, arg_array);
 }
 
-static bool _get_global_setting_bool(tf_ssb_t* ssb, const char* name, bool default_value)
-{
-	bool result = default_value;
-	JSContext* context = tf_ssb_get_context(ssb);
-	sqlite3* db = tf_ssb_get_db(ssb);
-	sqlite3_stmt* statement;
-	if (sqlite3_prepare(db, "SELECT value FROM properties WHERE id = 'core' AND key = 'settings'", -1, &statement, NULL) == SQLITE_OK)
-	{
-		if (sqlite3_step(statement) == SQLITE_ROW)
-		{
-			JSValue value = JS_ParseJSON(context, (const char*)sqlite3_column_text(statement, 0), sqlite3_column_bytes(statement, 0), NULL);
-			JSValue property = JS_GetPropertyStr(context, value,  name);
-			if (JS_IsBool(property))
-			{
-				result = JS_ToBool(context, property);
-			}
-			JS_FreeValue(context, property);
-			JS_FreeValue(context, value);
-		}
-		sqlite3_finalize(statement);
-	}
-	return result;
-}
-
 static void _tf_ssb_rpc_tunnel_is_room(tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number, JSValue args, const uint8_t* message, size_t size, void* user_data)
 {
 	tf_ssb_t* ssb = tf_ssb_connection_get_ssb(connection);
@@ -300,8 +337,10 @@ static void _tf_ssb_rpc_tunnel_is_room(tf_ssb_connection_t* connection, uint8_t
 	JSValue response = JS_FALSE;
 	if (_get_global_setting_bool(ssb, "room", true))
 	{
+		char room_name[1024] = "tilde friends tunnel";
+		_get_global_setting_string(ssb, "room_name", room_name, sizeof(room_name));
 		response = JS_NewObject(context);
-		JS_SetPropertyStr(context, response, "name", JS_NewString(context, "tilde friends tunnel"));
+		JS_SetPropertyStr(context, response, "name", JS_NewString(context, room_name));
 		JS_SetPropertyStr(context, response, "membership", JS_FALSE);
 		JSValue features = JS_NewArray(context);
 		JS_SetPropertyUint32(context, features, 0, JS_NewString(context, "tunnel"));
@@ -326,6 +365,56 @@ static void _tf_ssb_rpc_tunnel_is_room(tf_ssb_connection_t* connection, uint8_t
 	JS_FreeValue(context, response);
 }
 
+static void _tf_ssb_rpc_room_attendants(tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number, JSValue args, const uint8_t* message, size_t size, void* user_data)
+{
+	tf_ssb_t* ssb = tf_ssb_connection_get_ssb(connection);
+	if (!_get_global_setting_bool(ssb, "room", true))
+	{
+		tf_ssb_connection_rpc_send_error_method_not_allowed(connection, flags, -request_number);
+		return;
+	}
+
+	JSContext* context = tf_ssb_get_context(ssb);
+	JSValue joined = JS_NewObject(context);
+	JS_SetPropertyStr(context, joined, "type", JS_NewString(context, "joined"));
+	char my_id[k_id_base64_len] = "";
+	if (tf_ssb_connection_get_id(connection, my_id, sizeof(my_id)))
+	{
+		JS_SetPropertyStr(context, joined, "id", JS_NewString(context, my_id));
+	}
+
+	JSValue state = JS_NewObject(context);
+	JS_SetPropertyStr(context, state, "type", JS_NewString(context, "state"));
+	JSValue ids = JS_NewArray(context);
+	int id_count = 0;
+	tf_ssb_connection_t* connections[1024];
+	int count = tf_ssb_get_connections(ssb, connections, _countof(connections));
+	for (int i = 0; i < count; i++)
+	{
+		char id[k_id_base64_len] = { 0 };
+		if (tf_ssb_connection_is_attendant(connections[i]) &&
+			tf_ssb_connection_get_id(connections[i], id, sizeof(id)))
+		{
+			JS_SetPropertyUint32(context, ids, id_count++, JS_NewString(context, id));
+
+			tf_ssb_connection_rpc_send_json(
+				connections[i],
+				flags,
+				-tf_ssb_connection_get_attendant_request_number(connections[i]),
+				joined,
+				NULL,
+				NULL,
+				NULL);
+		}
+	}
+	JS_SetPropertyStr(context, state, "ids", ids);
+	tf_ssb_connection_rpc_send_json(connection, flags, -request_number, state, NULL, NULL, NULL);
+	JS_FreeValue(context, joined);
+	JS_FreeValue(context, state);
+
+	tf_ssb_connection_set_attendant(connection, true, request_number);
+}
+
 typedef struct _blobs_get_t
 {
 	char id[k_blob_id_len];
@@ -506,11 +595,94 @@ static void _tf_ssb_rpc_connection_blobs_createWants_callback(tf_ssb_connection_
 	js_free(context, ptab);
 }
 
+static void _tf_ssb_rpc_connection_room_attendants_callback(tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number, JSValue args, const uint8_t* message, size_t size, void* user_data)
+{
+	tf_ssb_t* ssb = tf_ssb_connection_get_ssb(connection);
+	JSContext* context = tf_ssb_get_context(ssb);
+	JSValue type = JS_GetPropertyStr(context, args, "type");
+	const char* type_string = JS_ToCString(context, type);
+	if (!type_string)
+	{
+		tf_ssb_connection_rpc_send_error(connection, flags, -request_number, "Missing type.");
+	}
+	else if (strcmp(type_string, "state") == 0)
+	{
+		tf_ssb_connection_clear_room_attendants(connection);
+		JSValue ids = JS_GetPropertyStr(context, args, "ids");
+		int length = tf_util_get_length(context, ids);
+		for (int i = 0; i < length; i++)
+		{
+			JSValue id = JS_GetPropertyUint32(context, ids, i);
+			const char* id_string = JS_ToCString(context, id);
+			if (id_string)
+			{
+				tf_ssb_connection_add_room_attendant(connection, id_string);
+			}
+			JS_FreeCString(context, id_string);
+			JS_FreeValue(context, id);
+		}
+		JS_FreeValue(context, ids);
+	}
+	else if (strcmp(type_string, "joined") == 0)
+	{
+		JSValue id = JS_GetPropertyStr(context, args, "id");
+		const char* id_string = JS_ToCString(context, id);
+		if (id_string)
+		{
+			tf_ssb_connection_add_room_attendant(connection, id_string);
+		}
+		JS_FreeCString(context, id_string);
+		JS_FreeValue(context, id);
+	}
+	else if (strcmp(type_string, "left") == 0)
+	{
+		JSValue id = JS_GetPropertyStr(context, args, "id");
+		const char* id_string = JS_ToCString(context, id);
+		if (id_string)
+		{
+			tf_ssb_connection_remove_room_attendant(connection, id_string);
+		}
+		JS_FreeCString(context, id_string);
+		JS_FreeValue(context, id);
+	}
+	else
+	{
+		tf_ssb_connection_rpc_send_error(connection, flags, -request_number, "Unexpected room.attendants response type.");
+	}
+	JS_FreeCString(context, type_string);
+	JS_FreeValue(context, type);
+}
+
+static void _tf_ssb_rpc_connection_tunnel_isRoom_callback(tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number, JSValue args, const uint8_t* message, size_t size, void* user_data)
+{
+	tf_ssb_t* ssb = tf_ssb_connection_get_ssb(connection);
+	JSContext* context = tf_ssb_get_context(ssb);
+	if (JS_IsObject(args))
+	{
+		JSValue message = JS_NewObject(context);
+		JSValue name = JS_NewArray(context);
+		JS_SetPropertyUint32(context, name, 0, JS_NewString(context, "room"));
+		JS_SetPropertyUint32(context, name, 1, JS_NewString(context, "attendants"));
+		JS_SetPropertyStr(context, message, "name", name);
+		JS_SetPropertyStr(context, message, "type", JS_NewString(context, "source"));
+		JS_SetPropertyStr(context, message, "args", JS_NewArray(context));
+		tf_ssb_connection_rpc_send_json(
+			connection,
+			k_ssb_rpc_flag_stream,
+			tf_ssb_connection_next_request_number(connection),
+			message,
+			_tf_ssb_rpc_connection_room_attendants_callback,
+			NULL,
+			NULL);
+		JS_FreeValue(context, message);
+	}
+}
+
 static void _tf_ssb_rpc_connections_changed_callback(tf_ssb_t* ssb, tf_ssb_change_t change, tf_ssb_connection_t* connection, void* user_data)
 {
+	JSContext* context = tf_ssb_get_context(ssb);
 	if (change == k_tf_ssb_change_connect)
 	{
-		JSContext* context = tf_ssb_get_context(ssb);
 		JSValue message = JS_NewObject(context);
 		JSValue name = JS_NewArray(context);
 		JS_SetPropertyUint32(context, name, 0, JS_NewString(context, "blobs"));
@@ -518,27 +690,63 @@ static void _tf_ssb_rpc_connections_changed_callback(tf_ssb_t* ssb, tf_ssb_chang
 		JS_SetPropertyStr(context, message, "name", name);
 		JS_SetPropertyStr(context, message, "type", JS_NewString(context, "source"));
 		JS_SetPropertyStr(context, message, "args", JS_NewArray(context));
-		JSValue message_val = JS_JSONStringify(context, message, JS_NULL, JS_NULL);
+		tf_ssb_connection_rpc_send_json(
-		size_t size;
-		const char* message_str = JS_ToCStringLen(context, &size, message_val);
-
-		tf_ssb_connection_rpc_send(
 			connection,
-			k_ssb_rpc_flag_json | k_ssb_rpc_flag_stream,
+			k_ssb_rpc_flag_stream,
 			tf_ssb_connection_next_request_number(connection),
-			(const uint8_t*)message_str,
+			message,
-			size,
 			_tf_ssb_rpc_connection_blobs_createWants_callback,
 			NULL,
 			NULL);
-
-		JS_FreeCString(context, message_str);
-		JS_FreeValue(context, message_val);
 		JS_FreeValue(context, message);
+
+		if (tf_ssb_connection_is_client(connection))
+		{
+			message = JS_NewObject(context);
+			name = JS_NewArray(context);
+			JS_SetPropertyUint32(context, name, 0, JS_NewString(context, "tunnel"));
+			JS_SetPropertyUint32(context, name, 1, JS_NewString(context, "isRoom"));
+			JS_SetPropertyStr(context, message, "name", name);
+			JS_SetPropertyStr(context, message, "args", JS_NewArray(context));
+			tf_ssb_connection_rpc_send_json(
+				connection,
+				0,
+				tf_ssb_connection_next_request_number(connection),
+				message,
+				_tf_ssb_rpc_connection_tunnel_isRoom_callback,
+				NULL,
+				NULL);
+			JS_FreeValue(context, message);
+		}
 	}
 	else if (change == k_tf_ssb_change_remove)
 	{
 		tf_ssb_remove_blob_want_added_callback(ssb, _tf_ssb_rpc_blob_wants_added_callback, connection);
+
+		JSValue left = JS_NewObject(context);
+		JS_SetPropertyStr(context, left, "type", JS_NewString(context, "left"));
+		char id[k_id_base64_len] = "";
+		if (tf_ssb_connection_get_id(connection, id, sizeof(id)))
+		{
+			JS_SetPropertyStr(context, left, "id", JS_NewString(context, id));
+		}
+		tf_ssb_connection_t* connections[1024];
+		int count = tf_ssb_get_connections(ssb, connections, _countof(connections));
+		for (int i = 0; i < count; i++)
+		{
+			if (tf_ssb_connection_is_attendant(connections[i]))
+			{
+				tf_ssb_connection_rpc_send_json(
+					connections[i],
+					k_ssb_rpc_flag_stream,
+					-tf_ssb_connection_get_attendant_request_number(connections[i]),
+					left,
+					NULL,
+					NULL,
+					NULL);
+			}
+		}
+		JS_FreeValue(context, left);
 	}
 }
 
@@ -551,4 +759,5 @@ void tf_ssb_rpc_register(tf_ssb_t* ssb)
 	tf_ssb_add_rpc_callback(ssb, (const char*[]) { "blobs", "createWants", NULL }, _tf_ssb_rpc_blobs_createWants, NULL, NULL);
 	tf_ssb_add_rpc_callback(ssb, (const char*[]) { "tunnel", "connect", NULL }, _tf_ssb_rpc_tunnel_connect, NULL, NULL);
 	tf_ssb_add_rpc_callback(ssb, (const char*[]) { "tunnel", "isRoom", NULL }, _tf_ssb_rpc_tunnel_is_room, NULL, NULL);
+	tf_ssb_add_rpc_callback(ssb, (const char*[]) { "room", "attendants", NULL }, _tf_ssb_rpc_room_attendants, NULL, NULL);
 }

src/ssb.tests.c

@@ -383,6 +383,7 @@ void tf_ssb_test_rooms(const tf_test_options_t* options)
 	}
 	tf_ssb_server_close(ssb0);
 
+	printf("Waiting for broadcasts.\n");
 	while (test.broadcast_count1 != 1 ||
 		test.broadcast_count2 != 1)
 	{
@@ -449,12 +450,12 @@ void tf_ssb_test_rooms(const tf_test_options_t* options)
 	uv_close((uv_handle_t*)&idle1, NULL);
 	uv_close((uv_handle_t*)&idle2, NULL);
 
-	uv_run(&loop, UV_RUN_DEFAULT);
-
 	tf_ssb_destroy(ssb0);
 	tf_ssb_destroy(ssb1);
 	tf_ssb_destroy(ssb2);
 
+	uv_run(&loop, UV_RUN_DEFAULT);
+
 	uv_loop_close(&loop);
 
 	sqlite3_close(db0);