*Refactor again to use an api for protomux channels and messages

*Only set mux.syncState if not previously set
*change mux.syncState to emit syncProtomux
*ensure the state data are numbers
*check for undefined in the state data

package.json

@@ -14,14 +14,18 @@
     "type": "module",
     "dependencies": {
         "@lumeweb/hyperswarm-web": "git+https://git.lumeweb.com/LumeWeb/hyperswarm-web.git",
+        "@lumeweb/rpc": "git+https://git.lumeweb.com/LumeWeb/rpc.git",
         "@noble/ed25519": "^1.7.3",
         "@peculiar/webcrypto": "git+https://git.lumeweb.com/LumeWeb/webcrypto.git",
+        "async-mutex": "^0.4.0",
         "b4a": "^1.6.3",
         "eventemitter2": "^6.4.9",
         "hyperswarm": "^4.4.0",
         "libkmodule": "^0.2.53",
         "libskynet": "^0.0.62",
         "noise-handshake": "^3.0.2",
+        "p-defer": "^4.0.0",
+        "protomux": "^3.4.1",
         "randombytes": "github:LumeWeb/randombytes-browser"
     },
     "devDependencies": {
@@ -52,7 +56,7 @@
         "stream-browserify": "^3.0.0",
         "ts-loader": "^9.4.2",
         "typescript": "^4.9.5",
-        "webpack": "^5.77.0",
+        "webpack": "^5.78.0",
         "webpack-cli": "^4.10.0"
     },
     "browser": {

src/index.ts

@@ -3,18 +3,22 @@ import Hyperswarm from "@lumeweb/hyperswarm-web";
 import type { ActiveQuery } from "libkmodule";
 import { addHandler, getSeed, handleMessage } from "libkmodule";
 import { handlePresentSeed as handlePresentSeedModule } from "libkmodule/dist/seed.js";
-import type { Buffer } from "buffer";
+import { Buffer } from "buffer";
 import * as ed from "@noble/ed25519";
 import b4a from "b4a";
 import { pubKeyToIpv6 } from "./addr.js";
 import { EventEmitter2 as EventEmitter } from "eventemitter2";
 import { logErr } from "libkmodule/dist";
+// @ts-ignore
+import Protomux from "protomux";
+import defer, { DeferredPromise } from "p-defer";
 
 const MAX_PEER_LISTENERS = 20;
 
 interface SwarmConnection {
   swarm: number;
   conn: any;
+  channels: Map<number, Protomux>;
 }
 
 interface SwarmEvents {
@@ -34,6 +38,7 @@ let moduleReady: Promise<void> = new Promise((resolve) => {
 });
 
 onmessage = handleMessage;
+
 function idFactory(start = 1) {
   let id = start;
 
@@ -46,6 +51,7 @@ function idFactory(start = 1) {
 
 const getSwarmId = idFactory();
 const getSocketId = idFactory();
+const getChannelId = idFactory();
 
 addHandler("presentSeed", handlePresentSeed);
 addHandler("join", handleJoin);
@@ -68,6 +74,13 @@ addHandler("socketListenEvent", handleSocketListenEvent, {
 });
 addHandler("socketWrite", handleWriteSocketEvent);
 addHandler("socketClose", handleCloseSocketEvent);
+addHandler("createProtomuxChannel", createProtomuxChannel, {
+  receiveUpdates: true,
+});
+addHandler("createProtomuxMessage", createProtomuxMessage, {
+  receiveUpdates: true,
+});
+
 async function handlePresentSeed(aq: ActiveQuery) {
   const pubkey = await ed.getPublicKey(aq.callerInput.rootKey);
   handlePresentSeedModule({
@@ -99,6 +112,7 @@ async function createSwarm(): Promise<number> {
       connections.set(socketId, {
         swarm: id,
         conn: peer,
+        channels: new Map<number, Protomux>(),
       });
 
       peer.once("close", () => {
@@ -142,7 +156,11 @@ function handleSocketListenEvent(aq: ActiveQuery) {
     aq.respond();
   };
 
-  const cb = (data: Buffer) => {
+  const cb = async (data: Buffer) => {
+    await socket.mutex?.waitForUnlock();
+    if (responded) {
+      return;
+    }
     aq.sendUpdate(data);
   };
 
@@ -176,7 +194,7 @@ function handleCloseSocketEvent(aq: ActiveQuery) {
   aq.respond();
 }
 
-function handleWriteSocketEvent(aq: ActiveQuery) {
+async function handleWriteSocketEvent(aq: ActiveQuery) {
   const socket = validateConnection(aq);
 
   if (!socket) {
@@ -189,6 +207,8 @@ function handleWriteSocketEvent(aq: ActiveQuery) {
     return false;
   }
 
+  await socket.mutex?.waitForUnlock();
+
   socket.write(message);
 
   aq.respond();
@@ -219,9 +239,16 @@ async function getSwarm(aq: ActiveQuery): Promise<Hyperswarm> {
   }
 
   if (!swarm) {
+    if (defaultSwarm.activeRelay && defaultSwarm.ready) {
+      await defaultSwarm.activeRelay.dht._protocol.opened;
+    }
     return defaultSwarm;
   }
 
+  if (swarm.activeRelay && swarm.ready) {
+    await swarm.activeRelay.dht._protocol.opened;
+  }
+
   return swarmInstances.get(swarm) as Hyperswarm;
 }
 
@@ -281,6 +308,7 @@ async function handleJoin(aq: ActiveQuery) {
   swarm.join(topic, { server: false });
   aq.respond();
 }
+
 async function handleGetPeerByPubkey(aq: ActiveQuery) {
   const { pubkey = null } = aq.callerInput;
 
@@ -319,17 +347,21 @@ async function handleInit(aq: ActiveQuery) {
 
   aq.respond();
 }
+
 async function handleReady(aq: ActiveQuery) {
   const swarm = await getSwarm(aq);
 
   if (swarm.activeRelay && swarm.ready) {
     aq.respond();
+    await swarm.activeRelay.dht._protocol.opened;
     return;
   }
-  swarm.once("ready", () => {
+  swarm.once("ready", async () => {
+    await swarm.activeRelay.dht._protocol.opened;
     aq.respond();
   });
 }
+
 async function handleListenConnections(aq: ActiveQuery) {
   const swarm = await getSwarm(aq);
   const swarmId = getSwarmToSwarmId(swarm);
@@ -351,6 +383,12 @@ async function handleListenConnections(aq: ActiveQuery) {
     aq.respond();
   });
 
+  for (const conn of connections) {
+    if (conn[1].swarm === swarmId) {
+      listener(conn[1].conn);
+    }
+  }
+
   const closeCb = () => {
     swarmEvent?.off("connection", listener);
     swarmEvent?.emit("close");
@@ -366,12 +404,6 @@ async function handleListenConnections(aq: ActiveQuery) {
     return;
   }
   swarm.onceSelf("ready", hookClose);
-
-  for (const conn of connections) {
-    if (conn[1].swarm === swarmId) {
-      listener(conn[1].conn);
-    }
-  }
 }
 
 async function handleGetSocketInfo(aq: ActiveQuery) {
@@ -392,6 +424,154 @@ async function handleGetSocketInfo(aq: ActiveQuery) {
   });
 }
 
+async function createProtomuxChannel(aq: ActiveQuery) {
+  const socket = validateConnection(aq);
+
+  if (!socket) {
+    return;
+  }
+
+  if (!("data" in aq.callerInput)) {
+    aq.reject("data required");
+    return;
+  }
+
+  const mux = Protomux.from(socket);
+  const data = aq.callerInput.data;
+
+  const handleCallback = (name: string, enabled: boolean) => {
+    if (!enabled && name !== "destroy") {
+      return undefined;
+    }
+    return (...args: any) => {
+      args = args.filter(
+        (item: any) => item.constructor.name.toLowerCase() !== "channel"
+      );
+
+      if (name === "destroy") {
+        connections.get(aq.callerInput.id)?.channels.delete(channelId);
+        aq.respond();
+      }
+
+      if (!enabled) {
+        return;
+      }
+
+      aq.sendUpdate({
+        action: name,
+        args,
+      });
+    };
+  };
+
+  aq.setReceiveUpdate?.((data: any) => {
+    switch (data.action) {
+      case "open":
+        channel.open();
+    }
+  });
+
+  const channel = mux.createChannel({
+    protocol: data?.protocol,
+    id: data?.id,
+    handshake: data?.handshake,
+    onopen: handleCallback("onopen", data?.onopen ?? undefined),
+    onclose: handleCallback("onclose", data?.onclose ?? undefined),
+    ondestroy: handleCallback("ondestroy", data?.ondestroy ?? undefined),
+  });
+
+  if (channel === null) {
+    aq.reject("duplicate channel");
+    return;
+  }
+
+  const channelId = getChannelId();
+
+  connections.get(aq.callerInput.id)?.channels.set(channelId, channel);
+
+  aq.sendUpdate(channelId);
+}
+
+async function createProtomuxMessage(aq: ActiveQuery) {
+  const socket = validateConnection(aq);
+
+  if (!socket) {
+    return;
+  }
+
+  if (!("data" in aq.callerInput)) {
+    aq.reject("action required");
+    return;
+  }
+
+  if (!("channelId" in aq.callerInput)) {
+    aq.reject("channel id required");
+    return;
+  }
+
+  const channel = connections
+    .get(aq.callerInput.id)
+    ?.channels.get(aq.callerInput.channelId);
+
+  if (!channel) {
+    aq.reject("invalid channel");
+  }
+
+  const data = aq.callerInput.data;
+
+  const defers: { [action: string]: DeferredPromise<any> } = {};
+
+  const handleEncoding = (enabled: boolean) => {
+    if (!enabled) {
+      return undefined;
+    }
+
+    const update = async (action: string, args: any) => {
+      await defers[action]?.promise;
+      defers[action] = defer();
+      aq.sendUpdate({
+        action,
+        args,
+      });
+
+      const ret = await defers[action]?.promise;
+
+      if (ret[1]) {
+        args[0].buffer = Buffer.from(ret[1].buffer);
+        args[0].start = ret[1].start;
+        args[0].end = ret[1].end;
+      }
+
+      return ret[0];
+    };
+
+    return {
+      async preencode(...args: any) {
+        return update("preencode", args);
+      },
+      async encode(...args: any) {
+        return update("encode", args);
+      },
+      async decode(...args: any) {
+        return update("encode", args);
+      },
+    };
+  };
+
+  aq.setReceiveUpdate?.((data) => {
+    defers[data.action]?.resolve(data.args[0]);
+  });
+
+  const message = channel.addMessage({
+    encoding: handleEncoding(data.encoding ?? false),
+    onmessage: data.encoding ?? undefined,
+  });
+
+  aq.sendUpdate({
+    action: "created",
+  });
+}
+
 function getSwarmToSocketConnectionId(socket: any) {
   for (const conn of connections) {
     if (conn[1].conn === socket) {
@@ -401,6 +581,7 @@ function getSwarmToSocketConnectionId(socket: any) {
 
   return false;
 }
+
 function getSwarmToSwarmId(swarm: any) {
   for (const swarmInstance of swarmInstances) {
     if (swarmInstance[1] === swarm) {