add realtime API support

Author: Septias

backend/message.ts

@@ -1,4 +1,5 @@
 import type {
+  RealtimeListener,
   ReceivedStatusUpdate,
   SendingStatusUpdate,
   Webxdc,
@@ -23,20 +24,28 @@ type Connect = (
 export type WebXdcMulti = {
   connect: Connect;
   sendUpdate: Webxdc<any>["sendUpdate"];
+  joinRealtimeChannel: Webxdc<any>["joinRealtimeChannel"]
 };
 
 export type UpdateDescr = [ReceivedStatusUpdate<any>, string];
 
 export type OnMessage = (message: Message) => void;
 
+export type OnRealtime = (message: Message) => void;
+
 export interface IProcessor {
   createClient(id: string): WebXdcMulti;
   clear(): void;
   removeClient(id: string): void;
 }
 
+class Realtime {
+  constructor(public listener: (data: Uint8Array) => void = () => {}) {};
+}
+
 class Client implements WebXdcMulti {
   updateListener: UpdateListenerMulti | null = null;
+  realtime: Realtime | null = null; 
   clearListener: ClearListener | null = null;
   updateSerial: number | null = null;
   deleteListener: DeleteListener | null = null;
@@ -50,6 +59,10 @@ class Client implements WebXdcMulti {
     this.processor.distribute(this.id, update, descr);
   }
 
+  sendRealtimeData(data: Uint8Array): void {
+    this.processor.distributeRealtime(this.id, data);
+  }
+
   connect(
     listener: UpdateListenerMulti,
     serial: number,
@@ -112,7 +125,29 @@ class Client implements WebXdcMulti {
     }
     this.updateListener([[update, descr]]);
   }
+  
+  receiveRealtime(data: Uint8Array) {
+    if (this.updateListener == null || this.updateSerial == null) {
+      return;
+    }
+    if (this.realtime && this.realtime.listener) 
+      this.realtime?.listener(data)
+  } 
 
+  joinRealtimeChannel(): RealtimeListener {
+    return {
+      setListener: (listener) =>  {
+        this.realtime = new Realtime(listener)
+      },
+      leave: () => {
+        this.realtime = null
+      },
+      send: (data) => {
+        this.sendRealtimeData(data)
+      }
+    }
+  }
+  
   clear() {
     if (
       this.clearListener == null ||
@@ -153,6 +188,23 @@ 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) {
+      client.receiveRealtime(data);
+    }
+  }
+
+
   distribute(
     instanceId: string,
     update: SendingStatusUpdate<any>,

types/message.ts

@@ -17,5 +17,6 @@ export type UpdateMessage =
 
 export type Message =
   | UpdateMessage
+  | ({ type: "realtime-sent", data: Uint8Array} & InstanceMessage)
   | ({ type: "clear" } & InstanceMessage)
   | ({ type: "connect" } & InstanceMessage);