package.json

@@ -14,18 +14,14 @@
     "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": {
@@ -56,7 +52,7 @@
         "stream-browserify": "^3.0.0",
         "ts-loader": "^9.4.2",
         "typescript": "^4.9.5",
-        "webpack": "^5.78.0",
+        "webpack": "^5.77.0",
         "webpack-cli": "^4.10.0"
     },
     "browser": {

src/index.ts

@@ -3,22 +3,18 @@ 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 { Buffer } from "buffer";
+import type { 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 {
@@ -38,7 +34,6 @@ let moduleReady: Promise<void> = new Promise((resolve) => {
 });
 
 onmessage = handleMessage;
-
 function idFactory(start = 1) {
   let id = start;
 
@@ -51,7 +46,6 @@ function idFactory(start = 1) {
 
 const getSwarmId = idFactory();
 const getSocketId = idFactory();
-const getChannelId = idFactory();
 
 addHandler("presentSeed", handlePresentSeed);
 addHandler("join", handleJoin);
@@ -74,13 +68,6 @@ 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({
@@ -112,7 +99,6 @@ async function createSwarm(): Promise<number> {
       connections.set(socketId, {
         swarm: id,
         conn: peer,
-        channels: new Map<number, Protomux>(),
       });
 
       peer.once("close", () => {
@@ -156,11 +142,7 @@ function handleSocketListenEvent(aq: ActiveQuery) {
     aq.respond();
   };
 
-  const cb = async (data: Buffer) => {
+  const cb = (data: Buffer) => {
-    await socket.mutex?.waitForUnlock();
-    if (responded) {
-      return;
-    }
     aq.sendUpdate(data);
   };
 
@@ -194,7 +176,7 @@ function handleCloseSocketEvent(aq: ActiveQuery) {
   aq.respond();
 }
 
-async function handleWriteSocketEvent(aq: ActiveQuery) {
+function handleWriteSocketEvent(aq: ActiveQuery) {
   const socket = validateConnection(aq);
 
   if (!socket) {
@@ -207,8 +189,6 @@ async function handleWriteSocketEvent(aq: ActiveQuery) {
     return false;
   }
 
-  await socket.mutex?.waitForUnlock();
-
   socket.write(message);
 
   aq.respond();
@@ -239,16 +219,9 @@ 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;
 }
 
@@ -308,7 +281,6 @@ async function handleJoin(aq: ActiveQuery) {
   swarm.join(topic, { server: false });
   aq.respond();
 }
-
 async function handleGetPeerByPubkey(aq: ActiveQuery) {
   const { pubkey = null } = aq.callerInput;
 
@@ -347,21 +319,17 @@ 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", async () => {
+  swarm.once("ready", () => {
-    await swarm.activeRelay.dht._protocol.opened;
     aq.respond();
   });
 }
-
 async function handleListenConnections(aq: ActiveQuery) {
   const swarm = await getSwarm(aq);
   const swarmId = getSwarmToSwarmId(swarm);
@@ -383,12 +351,6 @@ 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");
@@ -404,6 +366,12 @@ 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) {
@@ -424,154 +392,6 @@ 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) {
@@ -581,7 +401,6 @@ function getSwarmToSocketConnectionId(socket: any) {
 
   return false;
 }
-
 function getSwarmToSwarmId(swarm: any) {
   for (const swarmInstance of swarmInstances) {
     if (swarmInstance[1] === swarm) {