From 80a1e94da4125f9cf07436ea7ccf8d976228c17d Mon Sep 17 00:00:00 2001 From: Cory McWilliams Date: Mon, 9 Jan 2023 22:37:34 +0000 Subject: [PATCH] Simplify and fix ebt.replicate. git-svn-id: https://www.unprompted.com/svn/projects/tildefriends/trunk@4112 ed5197a5-7fde-0310-b194-c3ffbd925b24 --- src/ssb.c | 4 +-- src/ssb.rpc.c | 93 +++++++++++++++++++++------------------------------ 2 files changed, 41 insertions(+), 56 deletions(-) diff --git a/src/ssb.c b/src/ssb.c index 843b14e9..bb903489 100644 --- a/src/ssb.c +++ b/src/ssb.c @@ -603,7 +603,7 @@ void tf_ssb_connection_rpc_send_json(tf_ssb_connection_t* connection, uint8_t fl 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), + k_ssb_rpc_flag_json | (flags & k_ssb_rpc_flag_stream) | (flags & k_ssb_rpc_flag_end_error), request_number, (const uint8_t*)json_string, size, @@ -621,7 +621,7 @@ void tf_ssb_connection_rpc_send_error(tf_ssb_connection_t* connection, uint8_t f 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); + tf_ssb_connection_rpc_send_json(connection, flags | k_ssb_rpc_flag_end_error, request_number, message, NULL, NULL, NULL); JS_FreeValue(context, message); } diff --git a/src/ssb.rpc.c b/src/ssb.rpc.c index ec0ba6b7..f6fd1c36 100644 --- a/src/ssb.rpc.c +++ b/src/ssb.rpc.c @@ -348,20 +348,14 @@ static void _tf_ssb_rpc_tunnel_is_room(tf_ssb_connection_t* connection, uint8_t JS_SetPropertyStr(context, response, "features", features); } - JSValue message_val = JS_JSONStringify(context, response, JS_NULL, JS_NULL); - size_t json_size = 0; - const char* message_str = JS_ToCStringLen(context, &json_size, message_val); - tf_ssb_connection_rpc_send( + tf_ssb_connection_rpc_send_json( connection, flags | k_ssb_rpc_flag_end_error, -request_number, - (const uint8_t*)message_str, - json_size, + response, NULL, NULL, NULL); - JS_FreeCString(context, message_str); - JS_FreeValue(context, message_val); JS_FreeValue(context, response); } @@ -653,10 +647,29 @@ static void _tf_ssb_rpc_connection_room_attendants_callback(tf_ssb_connection_t* JS_FreeValue(context, type); } +static bool _is_error(JSContext* context, JSValue message) +{ + JSValue name = JS_GetPropertyStr(context, message, "name"); + const char* name_string = JS_ToCString(context, name); + bool is_error = false; + if (name_string && strcmp(name_string, "Error") == 0) + { + is_error = true; + } + JS_FreeCString(context, name_string); + JS_FreeValue(context, name); + return is_error; +} + 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 (_is_error(context, args)) + { + return; + } + if (JS_IsObject(args)) { JSValue message = JS_NewObject(context); @@ -757,20 +770,6 @@ static void _tf_ssb_rpc_createHistoryStream(tf_ssb_connection_t* connection, uin JS_FreeValue(context, arg_array); } -static bool _is_error(JSContext* context, JSValue message) -{ - JSValue name = JS_GetPropertyStr(context, message, "name"); - const char* name_string = JS_ToCString(context, name); - bool is_error = false; - if (name_string && strcmp(name_string, "Error") == 0) - { - is_error = true; - } - JS_FreeCString(context, name_string); - JS_FreeValue(context, name); - return is_error; -} - static void _tf_ssb_rpc_ebt_replicate_send_clock(tf_ssb_connection_t* connection, int32_t request_number, JSValue message) { tf_ssb_t* ssb = tf_ssb_connection_get_ssb(connection); @@ -783,7 +782,7 @@ static void _tf_ssb_rpc_ebt_replicate_send_clock(tf_ssb_connection_t* connection { int64_t sequence = 0; tf_ssb_db_get_latest_message_by_author(ssb, visible[i], &sequence, NULL, 0); - JS_SetPropertyStr(context, full_clock, visible[i], JS_NewInt64(context, sequence)); + JS_SetPropertyStr(context, full_clock, visible[i], JS_NewInt64(context, sequence == -1 ? -1 : (sequence << 1))); } /* Ask about the incoming connection, too. */ @@ -795,7 +794,7 @@ static void _tf_ssb_rpc_ebt_replicate_send_clock(tf_ssb_connection_t* connection { int64_t sequence = 0; tf_ssb_db_get_latest_message_by_author(ssb, id, &sequence, NULL, 0); - JS_SetPropertyStr(context, full_clock, id, JS_NewInt64(context, sequence)); + JS_SetPropertyStr(context, full_clock, id, JS_NewInt64(context, sequence == -1 ? -1 : (sequence << 1))); } JS_FreeValue(context, in_clock); } @@ -817,7 +816,7 @@ static void _tf_ssb_rpc_ebt_replicate_send_clock(tf_ssb_connection_t* connection { int64_t sequence = -1; tf_ssb_db_get_latest_message_by_author(ssb, key_string, &sequence, NULL, 0); - JS_SetPropertyStr(context, full_clock, key_string, JS_NewInt64(context, sequence)); + JS_SetPropertyStr(context, full_clock, key_string, JS_NewInt64(context, sequence == -1 ? -1 : (sequence << 1))); } JS_FreeCString(context, key_string); JS_FreeValue(context, key); @@ -860,8 +859,8 @@ static void _tf_ssb_rpc_ebt_replicate_send_messages(tf_ssb_connection_t* connect const char* author = JS_ToCString(context, key); if (sequence >= 0 && (sequence & 1) == 0) { - int32_t request_number = -tf_ssb_connection_get_ebt_request_number(connection); - _tf_ssb_connection_send_history_stream(connection, request_number, author, sequence, false); + int32_t request_number = tf_ssb_connection_get_ebt_request_number(connection); + _tf_ssb_connection_send_history_stream(connection, request_number, author, sequence >> 1, false); tf_ssb_connection_add_new_message_request(connection, author, request_number, false); } else @@ -890,6 +889,11 @@ static void _tf_ssb_rpc_ebt_replicate(tf_ssb_connection_t* connection, uint8_t f return; } + if (!tf_ssb_connection_get_ebt_request_number(connection)) + { + tf_ssb_connection_set_ebt_request_number(connection, -request_number); + } + JSValue author = JS_GetPropertyStr(context, args, "author"); JSValue name = JS_GetPropertyStr(context, args, "name"); JSValue in_clock = JS_IsUndefined(name) ? args : JS_UNDEFINED; @@ -909,7 +913,6 @@ static void _tf_ssb_rpc_ebt_replicate(tf_ssb_connection_t* connection, uint8_t f tf_ssb_connection_set_sent_clock(connection, true); } _tf_ssb_rpc_ebt_replicate_send_messages(connection, in_clock); - } JS_FreeValue(context, name); JS_FreeValue(context, author); @@ -936,44 +939,26 @@ static void _tf_ssb_rpc_send_ebt_replicate(tf_ssb_connection_t* connection) JS_SetPropertyUint32(context, args, 0, arg); JS_SetPropertyStr(context, message, "args", args); JS_SetPropertyStr(context, message, "type", JS_NewString(context, "duplex")); + int32_t request_number = tf_ssb_connection_next_request_number(connection); tf_ssb_connection_rpc_send_json( connection, - 0, - tf_ssb_connection_next_request_number(connection), + k_ssb_rpc_flag_stream, + request_number, message, _tf_ssb_rpc_ebt_replicate_client, NULL, NULL); + if (!tf_ssb_connection_get_ebt_request_number(connection)) + { + tf_ssb_connection_set_ebt_request_number(connection, request_number); + } JS_FreeValue(context, message); } static void _tf_ssb_rpc_ebt_replicate_server(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 (_is_error(context, args)) - { - /* TODO: Send createHistoryStream. */ - return; - } - - if (!tf_ssb_connection_get_ebt_request_number(connection)) - { - tf_ssb_connection_set_ebt_request_number(connection, request_number); - } - - JSValue in_name = JS_GetPropertyStr(context, args, "name"); - if (!JS_IsUndefined(in_name)) - { - /* This is the server receiving the initial ebt.replicate message. Respond. */ - if (!tf_ssb_connection_is_client(connection)) - { - _tf_ssb_rpc_send_ebt_replicate(connection); - } - } - JS_FreeValue(context, in_name); - _tf_ssb_rpc_ebt_replicate(connection, flags, request_number, args, message, size, user_data); + tf_ssb_connection_add_request(connection, -request_number, _tf_ssb_rpc_ebt_replicate, NULL, NULL, NULL); } 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)