ssb: Beginnings of a "sync now" mode for mobile.

src/ssb.c

@@ -365,6 +365,8 @@ typedef struct _tf_ssb_connection_t
 	int active_write_count;
 
 	uint64_t last_notified_active;
+
+	int flags;
 } tf_ssb_connection_t;
 
 static JSClassID _connection_class_id;
@@ -1148,11 +1150,11 @@ bool tf_ssb_verify_and_strip_signature(JSContext* context, JSValue val, char* ou
 	return false;
 }
 
-void tf_ssb_close_all(tf_ssb_t* ssb)
+void tf_ssb_close_all(tf_ssb_t* ssb, const char* reason)
 {
 	for (tf_ssb_connection_t* connection = ssb->connections; connection; connection = connection->next)
 	{
-		_tf_ssb_connection_close(connection, "tf_ssb_close_all");
+		_tf_ssb_connection_close(connection, reason);
 	}
 }
 
@@ -2755,7 +2757,7 @@ static void _tf_ssb_connection_tunnel_callback(
 	}
 }
 
-tf_ssb_connection_t* tf_ssb_connection_tunnel_create(tf_ssb_t* ssb, const char* portal_id, int32_t request_number, const char* target_id)
+tf_ssb_connection_t* tf_ssb_connection_tunnel_create(tf_ssb_t* ssb, const char* portal_id, int32_t request_number, const char* target_id, int connect_flags)
 {
 	tf_ssb_connection_t* connection = tf_ssb_connection_get(ssb, portal_id);
 
@@ -2765,6 +2767,7 @@ tf_ssb_connection_t* tf_ssb_connection_tunnel_create(tf_ssb_t* ssb, const char*
 	memset(tunnel, 0, sizeof(*tunnel));
 	snprintf(tunnel->name, sizeof(tunnel->name), "tun%d", s_tunnel_index++);
 	tunnel->ssb = ssb;
+	tunnel->flags = connect_flags;
 	tunnel->tunnel_connection = connection;
 	tunnel->tunnel_request_number = -request_number;
 	tunnel->send_request_number = 1;
@@ -2807,6 +2810,7 @@ typedef struct _connect_t
 	uv_getaddrinfo_t req;
 	char host[256];
 	int port;
+	int flags;
 	uint8_t key[k_id_bin_len];
 } connect_t;
 
@@ -2819,7 +2823,11 @@ static void _tf_on_connect_getaddrinfo(uv_getaddrinfo_t* addrinfo, int result, s
 		{
 			struct sockaddr_in addr = *(struct sockaddr_in*)info->ai_addr;
 			addr.sin_port = htons(connect->port);
-			tf_ssb_connection_create(connect->ssb, connect->host, &addr, connect->key);
+			tf_ssb_connection_t* connection = tf_ssb_connection_create(connect->ssb, connect->host, &addr, connect->key);
+			if (connection)
+			{
+				connection->flags = connect->flags;
+			}
 		}
 		else
 		{
@@ -2831,7 +2839,7 @@ static void _tf_on_connect_getaddrinfo(uv_getaddrinfo_t* addrinfo, int result, s
 	tf_free(connect);
 }
 
-void tf_ssb_connect(tf_ssb_t* ssb, const char* host, int port, const uint8_t* key)
+void tf_ssb_connect(tf_ssb_t* ssb, const char* host, int port, const uint8_t* key, int connect_flags)
 {
 	if (ssb->shutting_down)
 	{
@@ -2841,6 +2849,7 @@ void tf_ssb_connect(tf_ssb_t* ssb, const char* host, int port, const uint8_t* ke
 	*connect = (connect_t) {
 		.ssb = ssb,
 		.port = port,
+		.flags = connect_flags,
 		.req.data = connect,
 	};
 	char id[k_id_base64_len] = { 0 };
@@ -3131,12 +3140,12 @@ static bool _tf_ssb_parse_broadcast(const char* in_broadcast, tf_ssb_broadcast_t
 	return false;
 }
 
-void tf_ssb_connect_str(tf_ssb_t* ssb, const char* address)
+void tf_ssb_connect_str(tf_ssb_t* ssb, const char* address, int connect_flags)
 {
 	tf_ssb_broadcast_t broadcast = { 0 };
 	if (_tf_ssb_parse_broadcast(address, &broadcast))
 	{
-		tf_ssb_connect(ssb, broadcast.host, ntohs(broadcast.addr.sin_port), broadcast.pub);
+		tf_ssb_connect(ssb, broadcast.host, ntohs(broadcast.addr.sin_port), broadcast.pub, connect_flags);
 	}
 	else
 	{
@@ -4282,3 +4291,35 @@ void tf_ssb_connection_adjust_write_count(tf_ssb_connection_t* connection, int d
 	connection->active_write_count += delta;
 	_tf_ssb_connection_dispatch_scheduled(connection);
 }
+
+void tf_ssb_sync_start(tf_ssb_t* ssb)
+{
+	tf_ssb_connections_sync_start(ssb->connections_tracker);
+}
+
+bool tf_ssb_tunnel_create(tf_ssb_t* ssb, const char* portal_id, const char* target_id, int connect_flags)
+{
+	tf_ssb_connection_t* connection = tf_ssb_connection_get(ssb, portal_id);
+	if (connection)
+	{
+		JSContext* context = ssb->context;
+		int32_t request_number = tf_ssb_connection_next_request_number(connection);
+		JSValue message = JS_NewObject(context);
+		JSValue name = JS_NewArray(context);
+		JS_SetPropertyUint32(context, name, 0, JS_NewString(context, "tunnel"));
+		JS_SetPropertyUint32(context, name, 1, JS_NewString(context, "connect"));
+		JS_SetPropertyStr(context, message, "name", name);
+		JSValue arg = JS_NewObject(context);
+		JS_SetPropertyStr(context, arg, "portal", JS_NewString(context, portal_id));
+		JS_SetPropertyStr(context, arg, "target", JS_NewString(context, target_id));
+		JSValue args = JS_NewArray(context);
+		JS_SetPropertyUint32(context, args, 0, arg);
+		JS_SetPropertyStr(context, message, "args", args);
+		JS_SetPropertyStr(context, message, "type", JS_NewString(context, "duplex"));
+		tf_ssb_connection_rpc_send_json(connection, k_ssb_rpc_flag_stream | k_ssb_rpc_flag_new_request, request_number, "tunnel.connect", message, NULL, NULL, NULL);
+		JS_FreeValue(context, message);
+
+		tf_ssb_connection_tunnel_create(ssb, portal_id, request_number, target_id, connect_flags);
+	}
+	return connection != NULL;
+}