diff --git a/core/core.js b/core/core.js index 4f7ce728..c1f7a04d 100644 --- a/core/core.js +++ b/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, diff --git a/core/ssb.js b/core/ssb.js index d4996dd5..a835784b 100644 --- a/core/ssb.js +++ b/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 = {}; diff --git a/src/ssb.c b/src/ssb.c index 34cb093d..9994aee7 100644 --- a/src/ssb.c +++ b/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(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); + tf_ssb_connection_rpc_send_error_method_not_allowed(connection, flags, -request_number); } } } @@ -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; diff --git a/src/ssb.h b/src/ssb.h index 04303183..fc7a707a 100644 --- a/src/ssb.h +++ b/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); diff --git a/src/ssb.js.c b/src/ssb.js.c index 9f1b53ab..8091f629 100644 --- a/src/ssb.js.c +++ b/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); diff --git a/src/ssb.rpc.c b/src/ssb.rpc.c index 905d01af..c94636d5 100644 --- a/src/ssb.rpc.c +++ b/src/ssb.rpc.c @@ -11,6 +11,10 @@ #include #include +#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); - tf_ssb_connection_add_request(target_connection, tunnel_request_number, _tf_ssb_rpc_tunnel_callback, _tf_ssb_rpc_tunnel_cleanup, data1, target_connection); + printf("MAKE TUNNEL %p %p\n", connection, 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); - size_t size; - const char* message_str = JS_ToCStringLen(context, &size, message_val); - - tf_ssb_connection_rpc_send( + tf_ssb_connection_rpc_send_json( 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, - size, + message, _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); } diff --git a/src/ssb.tests.c b/src/ssb.tests.c index a90eca24..4db08cd4 100644 --- a/src/ssb.tests.c +++ b/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);