Merge pull request #67 from webxdc/sk/realtime2

Feat: Add realtime API

Author: Septias

backend/instance.ts

@@ -21,6 +21,11 @@ type SendUpdateMessage = {
   update: ReceivedStatusUpdate<any>;
 };
 
+type SendRealtimeMessage = {
+  type: "sendRealtime";
+  data: Uint8Array;
+};
+
 type SetUpdateListenerMessage = {
   type: "setUpdateListener";
   serial: number;
@@ -125,6 +130,12 @@ export class Instances {
         // XXX should validate parsed
         if (isSendUpdateMessage(parsed)) {
           instance.webXdc.sendUpdate(parsed.update, "");
+        } else if (isSendRealtimeMessage(parsed)) {
+          instance.webXdc.sendRealtimeData(parsed.data);
+        } else if (isSetRealtimeListenerMessage(parsed)) {
+          instance.webXdc.connectRealtime((data) => {
+            return broadcast(wss, JSON.stringify({ type: "realtime", data }));
+          });
         } else if (isSetUpdateListenerMessage(parsed)) {
           instance.webXdc.connect(
             (updates) => {
@@ -216,6 +227,16 @@ function isSendUpdateMessage(value: any): value is SendUpdateMessage {
   return value.type === "sendUpdate";
 }
 
+function isSendRealtimeMessage(value: any): value is SendRealtimeMessage {
+  return value.type === "sendRealtime";
+}
+
+function isSetRealtimeListenerMessage(
+  value: any,
+): value is { type: "setRealtimeListener" } {
+  return value.type === "setRealtimeListener";
+}
+
 function isSetUpdateListenerMessage(
   value: any,
 ): value is SetUpdateListenerMessage {

backend/message.test.ts

@@ -53,6 +53,33 @@ test("distribute to self", () => {
   ]);
 });
 
+test("Send realtime", () => {
+  const [getMessages, onMessage] = track();
+  const processor = createProcessor(onMessage);
+  const client0 = processor.createClient("3001");
+  const client1 = processor.createClient("3002");
+
+  const client0Heard: string[] = [];
+  const client1Heard: string[] = [];
+
+  const decoder = new TextDecoder();
+  client0.connectRealtime((data) => {
+    client0Heard.push(decoder.decode(data));
+    return true;
+  });
+  client1.connectRealtime((data) => {
+    client1Heard.push(decoder.decode(data));
+    return true;
+  });
+
+  const encoder = new TextEncoder();
+
+  client0.sendRealtimeData(new Uint8Array(encoder.encode("hi")));
+
+  expect(client1Heard).toMatchObject(["hi"]);
+  expect(client0Heard).toMatchObject([]);
+});
+
 test("distribute to self and other", () => {
   const [getMessages, onMessage] = track();
   const processor = createProcessor(onMessage);

backend/message.ts

@@ -1,4 +1,5 @@
-import type {
+import {
+  RealtimeListener as WebxdcRealtimeListener,
   ReceivedStatusUpdate,
   SendingStatusUpdate,
   Webxdc,
@@ -10,6 +11,7 @@ type UpdateListenerMulti = (updates: ReceivedStatusUpdate<any>[]) => boolean;
 
 type ClearListener = () => boolean;
 type DeleteListener = () => boolean;
+type RTListener = (data: Uint8Array) => boolean;
 
 type Connect = (
   updateListener: UpdateListenerMulti,
@@ -20,7 +22,9 @@ type Connect = (
 
 export type WebXdcMulti = {
   connect: Connect;
+  connectRealtime: (listener: RTListener) => void;
   sendUpdate: Webxdc<any>["sendUpdate"];
+  sendRealtimeData: (data: Uint8Array) => void;
 };
 
 export type OnMessage = (message: Message) => void;
@@ -33,6 +37,7 @@ export interface IProcessor {
 
 class Client implements WebXdcMulti {
   updateListener: UpdateListenerMulti | null = null;
+  realtimeListener: RTListener | null = null;
   clearListener: ClearListener | null = null;
   updateSerial: number | null = null;
   deleteListener: DeleteListener | null = null;
@@ -46,6 +51,35 @@ class Client implements WebXdcMulti {
     this.processor.distribute(this.id, update);
   }
 
+  sendRealtimeData(data: Uint8Array) {
+    this.processor.distributeRealtime(this.id, data);
+  }
+
+  connectRealtime(listener: RTListener) {
+    this.processor.onMessage({
+      type: "connect-realtime",
+      instanceId: this.id,
+      instanceColor: getColorForId(this.id),
+      timestamp: Date.now(),
+    });
+
+    const realtimeListener = (data: Uint8Array) => {
+      const hasReceived = listener(data);
+      if (hasReceived) {
+        this.processor.onMessage({
+          type: "realtime-received",
+          data,
+          instanceId: this.id,
+          instanceColor: getColorForId(this.id),
+          timestamp: Date.now(),
+        });
+      }
+      return hasReceived;
+    };
+
+    this.realtimeListener = realtimeListener;
+  }
+
   connect(
     listener: UpdateListenerMulti,
     serial: number,
@@ -108,6 +142,13 @@ class Client implements WebXdcMulti {
     this.updateListener([update]);
   }
 
+  receiveRealtime(data: Uint8Array) {
+    if (this.realtimeListener == null) {
+      return;
+    }
+    this.realtimeListener(data);
+  }
+
   clear() {
     if (
       this.clearListener == null ||
@@ -148,6 +189,21 @@ class Processor implements IProcessor {
     this.clients.splice(client_index, 1);
   }
 
+  distributeRealtime(instanceId: string, data: Uint8Array) {
+    this.onMessage({
+      type: "realtime-sent",
+      instanceId: instanceId,
+      instanceColor: getColorForId(instanceId),
+      data,
+      timestamp: Date.now(),
+    });
+    for (const client of this.clients) {
+      if (client.id != instanceId) {
+        client.receiveRealtime(data);
+      }
+    }
+  }
+
   distribute(instanceId: string, update: SendingStatusUpdate<any>) {
     this.currentSerial++;
     const receivedUpdate: ReceivedStatusUpdate<any> = {