cory/tildefriends
5
6
Fork 2
Code Issues 28 Pull Requests Actions Releases 39 Wiki Activity

Simplify and fix ebt.replicate.

Browse Source 
git-svn-id: https://www.unprompted.com/svn/projects/tildefriends/trunk@4112 ed5197a5-7fde-0310-b194-c3ffbd925b24
This commit is contained in:
Cory McWilliams
2023-01-09 22:37:34 +00:00
parent 69253432b8
commit 80a1e94da4
2 changed files with 41 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/ssb.c
View File

@@ -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);
}

93
src/ssb.rpc.c
View File

@@ -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)