From 2dc7f58c80c1697561933371b500e321f075398f Mon Sep 17 00:00:00 2001 From: Cory McWilliams Date: Sat, 12 Nov 2022 02:00:49 +0000 Subject: [PATCH] Fixed some plumbing so that I can actually stay connected to a go-ssb-room. git-svn-id: https://www.unprompted.com/svn/projects/tildefriends/trunk@4032 ed5197a5-7fde-0310-b194-c3ffbd925b24 --- core/ssb.js | 95 +++++++++++++++++++++++++++-------------------------- src/ssb.c | 26 ++++++++++++--- 2 files changed, 71 insertions(+), 50 deletions(-) diff --git a/core/ssb.js b/core/ssb.js index 03092ba7..3ecd53b8 100644 --- a/core/ssb.js +++ b/core/ssb.js @@ -94,8 +94,45 @@ function tunnel_attendants(request) { }); } +function send_blobs_create_wants(connection) { + connection.send_json({'name': ['blobs', 'createWants'], 'type': 'source', 'args': []}, function on_blob_create_wants(message) { + Object.keys(message.message).forEach(function(id) { + if (message.message[id] < 0) { + if (g_wants_requests[connection.id]) { + delete connection.active_blob_wants[id]; + var blob = ssb.blobGet(id); + if (blob) { + var out_message = {}; + out_message[id] = blob.byteLength; + g_wants_requests[connection.id].send_json(out_message); + } + } + } else { + var received_bytes = 0; + var expected_bytes = message.message[id]; + var buffer = new Uint8Array(expected_bytes); + connection.send_json({'name': ['blobs', 'get'], 'type': 'source', 'args': [id]}, function(message) { + if (message.flags & 0x4 /* end */) { + delete connection.active_blob_wants[id]; + } else { + buffer.set(new Uint8Array(message.message, 0, message.message.byteLength), received_bytes); + received_bytes += message.message.byteLength; + if (received_bytes == expected_bytes) { + ssb.blobStore(buffer); + } + } + }); + if (g_wants_requests[connection.id] && Object.keys(connection.active_blob_wants).length < k_blobs_concurrent_target) { + requestMoreBlobs(g_wants_requests[connection.id]); + } + } + }); + }); +} + ssb.addEventListener('connections', function on_connections_changed(change, connection) { if (change == 'add') { + connection.active_blob_wants = {}; var sequence = get_latest_sequence_for_author(connection.id); if (k_use_create_history_stream) { connection.send_json({'name': ['createHistoryStream'], 'type': 'source', 'args': [{'id': connection.id, 'seq': sequence, 'live': true, 'keys': false}]}, storeMessage); @@ -111,49 +148,16 @@ ssb.addEventListener('connections', function on_connections_changed(change, conn }); } else { if (connection.is_client) { - connection.send_json({'name': ['tunnel', 'isRoom'], 'args': []}, function tunnel_is_room(request) { + 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) { - connection.send_json({'name': ['room', 'attendants'], 'args': [], 'type': 'source'}, tunnel_attendants); + request.connection.send_json({'name': ['room', 'attendants'], 'args': [], 'type': 'source'}, tunnel_attendants); } }); - connection.send_json({"name": ["ebt", "replicate"], "args": [{"version": 3, "format": "classic"}], "type": "duplex"}, ebtReplicateClient); } + send_blobs_create_wants(connection); } - - connection.active_blob_wants = {}; - connection.send_json({'name': ['blobs', 'createWants'], 'type': 'source', 'args': []}, function on_blob_create_wants(message) { - Object.keys(message.message).forEach(function(id) { - if (message.message[id] < 0) { - if (g_wants_requests[connection.id]) { - delete connection.active_blob_wants[id]; - var blob = ssb.blobGet(id); - if (blob) { - var out_message = {}; - out_message[id] = blob.byteLength; - g_wants_requests[connection.id].send_json(out_message); - } - } - } else { - var received_bytes = 0; - var expected_bytes = message.message[id]; - var buffer = new Uint8Array(expected_bytes); - connection.send_json({'name': ['blobs', 'get'], 'type': 'source', 'args': [id]}, function(message) { - if (message.flags & 0x4 /* end */) { - delete connection.active_blob_wants[id]; - } else { - buffer.set(new Uint8Array(message.message, 0, message.message.byteLength), received_bytes); - received_bytes += message.message.byteLength; - if (received_bytes == expected_bytes) { - ssb.blobStore(buffer); - } - } - }); - if (g_wants_requests[connection.id] && Object.keys(connection.active_blob_wants).length < k_blobs_concurrent_target) { - requestMoreBlobs(g_wants_requests[connection.id]); - } - } - }); - }); } else if (change == 'remove') { print('REMOVE', connection.id); notify_attendant_changed(connection.id, 'left'); @@ -181,9 +185,6 @@ function requestMoreBlobs(request) { row => blob_want_discovered(request, row.id)); } -ssb.addRpc(['manifest'], function(request) { -}); - ssb.addRpc(['blobs', 'createWants'], function(request) { g_wants_requests[request.connection.id] = request; ssb.addEventListener('blob_want_added', id => blob_want_discovered(request, id)); @@ -229,7 +230,7 @@ ssb.addRpc(['gossip', 'ping'], function(request) { }); ssb.addRpc(['tunnel', 'isRoom'], function(request) { - request.send_json(true); + request.send_json({"name":"tilde friends tunnel","membership":false,"features":["tunnel","room1"]}); }); function notify_attendant_changed(id, type) { @@ -369,11 +370,13 @@ function ebtReplicateCommon(request) { } function ebtReplicateClient(request) { - if (!request.connection.message_registered) { - ebtReplicateRegisterMessageCallback(request); - request.connection.message_registered = true; + if (request.message?.name !== 'Error') { + if (!request.connection.message_registered) { + ebtReplicateRegisterMessageCallback(request); + request.connection.message_registered = true; + } + ebtReplicateCommon(request); } - ebtReplicateCommon(request); } function ebtReplicateServer(request) { diff --git a/src/ssb.c b/src/ssb.c index a81083f8..d220098e 100644 --- a/src/ssb.c +++ b/src/ssb.c @@ -1124,6 +1124,13 @@ static bool _tf_ssb_name_equals(JSContext* context, JSValue object, const char** result = false; } } + else if (JS_IsString(name)) + { + /* Manifest is traditionally sent as not an array for some reason. */ + const char* str = JS_ToCString(context, name); + result = str && match[0] && strcmp(str, match[0]) == 0 && !match[1]; + JS_FreeCString(context, str); + } else { result = false; @@ -1625,7 +1632,6 @@ static void _tf_ssb_connection_on_connect(uv_connect_t* connect, int status) } else { - printf("connect => %s\n", uv_strerror(status)); _tf_ssb_connection_close(connection, "uv_tcp_connect failed"); } } @@ -1986,7 +1992,7 @@ static void _tf_ssb_connection_cleanup_value(tf_ssb_t* ssb, void* user_data) } } -static JSValue _tf_ssb_connection_send_json(JSContext* context, JSValueConst this_val, int argc, JSValueConst* argv) +static JSValue _tf_ssb_connection_send_json_internal(JSContext* context, JSValueConst this_val, int argc, JSValueConst* argv, int flags) { tf_ssb_connection_t* connection = JS_GetOpaque(this_val, _connection_class_id); if (!connection) @@ -2003,7 +2009,7 @@ static JSValue _tf_ssb_connection_send_json(JSContext* context, JSValueConst thi tf_ssb_connection_rpc_send( connection, - k_ssb_rpc_flag_json | k_ssb_rpc_flag_stream, + flags, request_number, (const uint8_t*)message, size, @@ -2014,6 +2020,16 @@ static JSValue _tf_ssb_connection_send_json(JSContext* context, JSValueConst thi return JS_NewInt32(context, request_number); } +static JSValue _tf_ssb_connection_send_json(JSContext* context, JSValueConst this_val, int argc, JSValueConst* argv) +{ + return _tf_ssb_connection_send_json_internal(context, this_val, argc, argv, k_ssb_rpc_flag_json | k_ssb_rpc_flag_stream); +} + +static JSValue _tf_ssb_connection_send_json_async(JSContext* context, JSValueConst this_val, int argc, JSValueConst* argv) +{ + return _tf_ssb_connection_send_json_internal(context, this_val, argc, argv, k_ssb_rpc_flag_json); +} + static void _tf_ssb_connection_process_message_async(uv_async_t* async) { tf_ssb_connection_t* connection = async->data; @@ -2059,6 +2075,7 @@ tf_ssb_connection_t* tf_ssb_connection_create(tf_ssb_t* ssb, const char* host, c connection->object = JS_NewObjectClass(ssb->context, _connection_class_id); JS_SetOpaque(connection->object, connection); JS_SetPropertyStr(context, connection->object, "send_json", JS_NewCFunction(context, _tf_ssb_connection_send_json, "send_json", 2)); + JS_SetPropertyStr(context, connection->object, "send_json_async", JS_NewCFunction(context, _tf_ssb_connection_send_json_async, "send_json_async", 2)); char public_key_str[k_id_base64_len] = { 0 }; if (tf_ssb_id_bin_to_str(public_key_str, sizeof(public_key_str), public_key)) { @@ -2069,7 +2086,6 @@ tf_ssb_connection_t* tf_ssb_connection_create(tf_ssb_t* ssb, const char* host, c memcpy(connection->serverpub, public_key, sizeof(connection->serverpub)); uv_tcp_init(ssb->loop, &connection->tcp); - printf("uv_tcp_connect\n"); int result = uv_tcp_connect(&connection->connect, &connection->tcp, (const struct sockaddr*)addr, _tf_ssb_connection_on_connect); if (result) { @@ -2118,6 +2134,7 @@ tf_ssb_connection_t* tf_ssb_connection_tunnel_create(tf_ssb_connection_t* connec tunnel->object = JS_NewObjectClass(ssb->context, _connection_class_id); JS_SetOpaque(tunnel->object, tunnel); JS_SetPropertyStr(context, tunnel->object, "send_json", JS_NewCFunction(context, _tf_ssb_connection_send_json, "send_json", 2)); + JS_SetPropertyStr(context, tunnel->object, "send_json_async", JS_NewCFunction(context, _tf_ssb_connection_send_json_async, "send_json_async", 2)); JS_SetPropertyStr(context, tunnel->object, "id", JS_NewString(context, target_id)); JS_SetPropertyStr(context, tunnel->object, "is_client", JS_TRUE); @@ -2217,6 +2234,7 @@ static void _tf_ssb_on_connection(uv_stream_t* stream, int status) connection->object = JS_NewObjectClass(ssb->context, _connection_class_id); JS_SetPropertyStr(ssb->context, connection->object, "send_json", JS_NewCFunction(ssb->context, _tf_ssb_connection_send_json, "send_json", 2)); + JS_SetPropertyStr(ssb->context, connection->object, "send_json_async", JS_NewCFunction(ssb->context, _tf_ssb_connection_send_json_async, "send_json_async", 2)); JS_SetOpaque(connection->object, connection); if (uv_tcp_init(ssb->loop, &connection->tcp) != 0)