[core] Synclets can have multiple transport

Author: jamesgpearce

src/@types/index.d.ts

@@ -57,7 +57,7 @@ export type SyncletOptions = {
 
 export function createSynclet(
   connector: Connector,
-  transport: Transport,
+  transport: Transport | Transport[],
   implementations?: SyncletImplementations,
   options?: SyncletOptions,
 ): Promise<Synclet>;

src/common/queue.ts

@@ -1,6 +1,6 @@
 import {arrayPush, arrayShift} from './array.ts';
 
-type Task = () => Promise<any>;
+export type Task = () => Promise<any>;
 
 export const getQueueFunctions = (): [
   queue: (...tasks: Task[]) => Promise<void>,

src/core/connector.ts

@@ -9,7 +9,6 @@ import type {
   Hash,
   LogLevel,
   Meta,
-  Synclet,
   Timestamp,
 } from '@synclets/@types';
 import {Tomb} from '@synclets/@types/utils';
@@ -27,8 +26,11 @@ import {
   size,
 } from '../common/other.ts';
 import {getQueueFunctions} from '../common/queue.ts';
-import {setEvery, setNew} from '../common/set.ts';
-import {ProtectedConnector} from './types.js';
+import {
+  ProtectedConnector,
+  ProtectedSynclet,
+  ProtectedTransport,
+} from './types.js';
 
 export const createConnector: typeof createConnectorDecl = async (
   depth,
@@ -51,10 +53,9 @@ export const createConnector: typeof createConnectorDecl = async (
   }
 
   let connected = false;
+  let boundSynclet: ProtectedSynclet | undefined;
   let id = options.id ?? getUniqueId();
 
-  const boundSynclets: Set<Synclet> = setNew();
-
   const logger = options.logger ?? {};
 
   const [getNextTimestamp, seenTimestamp] = getHlcFunctions(id);
@@ -105,18 +106,19 @@ export const createConnector: typeof createConnectorDecl = async (
   const log = (string: string, level: LogLevel = 'info') =>
     logger?.[level]?.(`[${id}/C] ${string}`);
 
-  const bind = (synclet: Synclet, syncletId: string) => {
-    boundSynclets.add(synclet);
-    if (boundSynclets.size == 1) {
-      id = syncletId;
+  const bind = (synclet: ProtectedSynclet, syncletId: string) => {
+    if (boundSynclet) {
+      errorNew('Connector is already attached to Synclet');
     }
+    boundSynclet = synclet;
+    id = syncletId;
   };
 
   const setOrDelAtom = async (
     address: Address,
     atomOrUndefined: Atom | undefined,
     context: Context = {},
-    syncOrFromSynclet: boolean | Synclet = true,
+    syncOrFromTransport: boolean | ProtectedTransport = true,
     newTimestamp?: Timestamp,
     oldTimestamp?: Timestamp,
   ) => {
@@ -154,13 +156,12 @@ export const createConnector: typeof createConnectorDecl = async (
         });
       }
     }
-    if (syncOrFromSynclet) {
-      boundSynclets.forEach((boundSynclet) => {
-        if (boundSynclet !== syncOrFromSynclet) {
-          arrayPush(tasks, () => boundSynclet.sync(address));
-        }
-      });
+    if (syncOrFromTransport === true) {
+      arrayPush(tasks, () => boundSynclet?.sync(address));
+    } else if (syncOrFromTransport) {
+      arrayPush(tasks, () => boundSynclet?._[0](address, syncOrFromTransport));
     }
+
     await queue(...tasks);
   };
 
@@ -177,13 +178,9 @@ export const createConnector: typeof createConnectorDecl = async (
 
     disconnect: async () => {
       if (connected) {
-        if (
-          setEvery(boundSynclets, (boundSynclet) => !boundSynclet.isStarted())
-        ) {
-          log('disconnect');
-          await disconnect?.();
-          connected = false;
-        }
+        log('disconnect');
+        await disconnect?.();
+        connected = false;
       }
     },
 

src/core/synclet.ts

@@ -6,22 +6,27 @@ import {
   LogLevel,
   ProtocolNode,
   ProtocolSubNodes,
-  Synclet,
   SyncletImplementations,
   SyncletOptions,
   Timestamp,
   TimestampAndAtom,
 } from '@synclets/@types';
 import {getUniqueId} from '@synclets/utils';
-import {arrayDifference, arrayMap} from '../common/array.ts';
+import {
+  arrayDifference,
+  arrayForEach,
+  arrayMap,
+  arrayPush,
+  isArray,
+} from '../common/array.ts';
 import {objKeys, objNotEmpty, objToArray} from '../common/object.ts';
 import {
   ifNotUndefined,
   isUndefined,
   promiseAll,
   size,
 } from '../common/other.ts';
-import {getQueueFunctions} from '../common/queue.ts';
+import {getQueueFunctions, Task} from '../common/queue.ts';
 import {
   ASTERISK,
   EMPTY_STRING,
@@ -35,7 +40,12 @@ import {
   isTimestamp,
   isTimestampAndAtom,
 } from '../common/types.ts';
-import {Message, ProtectedConnector, ProtectedTransport} from './types.js';
+import {
+  Message,
+  ProtectedConnector,
+  ProtectedSynclet,
+  ProtectedTransport,
+} from './types.js';
 
 const VERSION = 1;
 
@@ -53,17 +63,10 @@ export const createSynclet: typeof createSyncletDecl = (async (
       connectorSetOrDelAtom,
     ],
   }: ProtectedConnector,
-  {
-    _: [
-      transportBind,
-      transportConnect,
-      transportDisconnect,
-      transportSendMessage,
-    ],
-  }: ProtectedTransport,
+  transport: ProtectedTransport | ProtectedTransport[],
   {canReceiveMessage, getSendContext}: SyncletImplementations = {},
   options: SyncletOptions = {},
-): Promise<Synclet> => {
+): Promise<ProtectedSynclet> => {
   let started = false;
   const id = options.id ?? getUniqueId();
   const logger = options.logger ?? {};
@@ -75,6 +78,8 @@ export const createSynclet: typeof createSyncletDecl = (async (
     }
   };
 
+  const transports = isArray(transport) ? transport : [transport];
+
   const readHashOrTimestamp = async (address: Address, context: Context) =>
     size(address) < connectorDepth
       ? ((await connectorReadHash(address, context)) ?? 0)
@@ -117,19 +122,35 @@ export const createSynclet: typeof createSyncletDecl = (async (
     return [subNodeObj];
   };
 
-  const sync = (address: Address) =>
-    queueIfStarted(async () => {
+  const sync = (address: Address) => syncExcept(address);
+
+  const syncExcept = async (
+    address: Address,
+    exceptTransport?: ProtectedTransport,
+  ) => {
+    if (started) {
       log(`sync ` + address);
-      await sendNodeMessage(address, await readHashOrTimestamp(address, {}));
-    });
+      const hashOrTimestamp = await readHashOrTimestamp(address, {});
+      const tasks: Task[] = [];
+      arrayForEach(transports, (transport) => {
+        if (transport !== exceptTransport) {
+          arrayPush(tasks, () =>
+            sendNodeMessage(transport, address, hashOrTimestamp),
+          );
+        }
+      });
+      await queue(...tasks);
+    }
+  };
 
   const sendNodeMessage = async (
+    transport: ProtectedTransport,
     address: Address,
     node: ProtocolNode,
     receivedContext: Context = {},
     to?: string,
   ) =>
-    await transportSendMessage(
+    await transport._[3](
       [
         VERSION,
         0,
@@ -141,7 +162,11 @@ export const createSynclet: typeof createSyncletDecl = (async (
       to,
     );
 
-  const receiveMessage = (message: Message, from: string) =>
+  const receiveMessage = (
+    transport: ProtectedTransport,
+    message: Message,
+    from: string,
+  ) =>
     queueIfStarted(async () => {
       const [version, type, depth, address, node, context] = message;
 
@@ -164,12 +189,13 @@ export const createSynclet: typeof createSyncletDecl = (async (
         return log(`can't receive message: ${from}`, WARN);
       }
 
-      await transformNode(address, node, context, (newNode) =>
-        sendNodeMessage(address, newNode, context, from),
+      await transformNode(transport, address, node, context, (newNode) =>
+        sendNodeMessage(transport, address, newNode, context, from),
       );
     });
 
   const transformNode = async (
+    transport: ProtectedTransport,
     address: Address,
     node: ProtocolNode,
     context: Context,
@@ -187,12 +213,13 @@ export const createSynclet: typeof createSyncletDecl = (async (
               : isProtocolSubNodes(node)
                 ? transformSubNodes
                 : () => log(INVALID_NODE + ' ' + address, WARN)) as any
-      )?.(address, node, context),
+      )?.(transport, address, node, context),
       ifTransformedToDefined,
       ifTransformedToUndefined,
     );
 
   const transformTimestamp = async (
+    _transport: ProtectedTransport,
     address: Address,
     otherTimestamp: Timestamp,
     context: Context,
@@ -211,6 +238,7 @@ export const createSynclet: typeof createSyncletDecl = (async (
   };
 
   const transformTimestampAndAtom = async (
+    transport: ProtectedTransport,
     address: Address,
     otherTimestampAndAtom: TimestampAndAtom,
     context: Context,
@@ -224,7 +252,7 @@ export const createSynclet: typeof createSyncletDecl = (async (
           address,
           otherAtom,
           context,
-          synclet,
+          transport,
           otherTimestamp,
           myTimestamp,
         );
@@ -237,6 +265,7 @@ export const createSynclet: typeof createSyncletDecl = (async (
   };
 
   const transformHash = async (
+    _transport: ProtectedTransport,
     address: Address,
     otherHash: Hash,
     context: Context,
@@ -263,6 +292,7 @@ export const createSynclet: typeof createSyncletDecl = (async (
   };
 
   const transformSubNodes = async (
+    transport: ProtectedTransport,
     address: Address,
     [otherSubNodeObj, partial]: ProtocolSubNodes,
     context: Context,
@@ -274,6 +304,7 @@ export const createSynclet: typeof createSyncletDecl = (async (
       await promiseAll(
         objToArray(otherSubNodeObj, (id, otherSubNode) =>
           transformNode(
+            transport,
             [...address, id],
             otherSubNode,
             context,
@@ -297,14 +328,20 @@ export const createSynclet: typeof createSyncletDecl = (async (
   const log = (string: string, level: LogLevel = 'info') =>
     logger?.[level]?.(`[${id}] ${string}`);
 
-  const synclet: Synclet = {
+  const synclet: ProtectedSynclet = {
     log,
 
     start: async () => {
       if (!started) {
         log('start');
         await connectorConnect();
-        await transportConnect(receiveMessage);
+        await promiseAll(
+          arrayMap(transports, (transport) =>
+            transport._[1]((message: Message, from: string) =>
+              receiveMessage(transport, message, from),
+            ),
+          ),
+        );
         started = true;
         await sync([]);
       }
@@ -315,16 +352,19 @@ export const createSynclet: typeof createSyncletDecl = (async (
         log('stop');
         started = false;
         await connectorDisconnect();
-        await transportDisconnect();
+        await promiseAll(arrayMap(transports, (transport) => transport._[2]()));
       }
     },
 
     isStarted: () => started,
 
     sync,
+
+    _: [syncExcept],
   };
 
   connectorBind(synclet, id);
-  transportBind(synclet, id);
+  arrayForEach(transports, (transport) => transport._[0](synclet, id));
+
   return synclet;
 }) as any;

src/core/transport.ts

@@ -1,7 +1,6 @@
 import type {
   createTransport as createTransportDecl,
   LogLevel,
-  Synclet,
   TransportImplementations,
   TransportOptions,
 } from '@synclets/@types';
@@ -10,7 +9,12 @@ import {arrayJoin, arrayMap} from '../common/array.ts';
 import {mapEnsure, mapNew} from '../common/map.ts';
 import {errorNew, promiseAll, size} from '../common/other.ts';
 import {ASTERISK, SPACE} from '../common/string.ts';
-import {Message, ProtectedTransport, ReceiveMessage} from './types.js';
+import {
+  Message,
+  ProtectedSynclet,
+  ProtectedTransport,
+  ReceiveMessage,
+} from './types.js';
 
 type Pending = [fragments: string[], due: number];
 
@@ -99,7 +103,7 @@ export const createTransport: typeof createTransportDecl = async (
   options: TransportOptions = {},
 ): Promise<ProtectedTransport> => {
   let connected = false;
-  let boundSynclet: Synclet | undefined;
+  let boundSynclet: ProtectedSynclet | undefined;
   let id = options.id ?? getUniqueId();
 
   const logger = options.logger ?? {};
@@ -110,7 +114,7 @@ export const createTransport: typeof createTransportDecl = async (
   const [startBuffer, stopBuffer, receivePacket, sendPackets] =
     getPacketFunctions(log, sendPacket, options.fragmentSize ?? 4096);
 
-  const bind = (synclet: Synclet, syncletId: string) => {
+  const bind = (synclet: ProtectedSynclet, syncletId: string) => {
     if (boundSynclet) {
       errorNew('Transport is already attached to Synclet');
     }