dist/peer.d.ts

@@ -27,6 +27,7 @@ export interface DataSocketOptions {
     onclose?: OnClose;
     onchannel?: OnChannel;
     emulateWebsocket?: boolean;
+    createDefaultMessage?: boolean;
 }
 export interface PeerOptions {
     peer: any;
@@ -35,25 +36,21 @@ export interface PeerOptions {
 export interface PeerOptionsWithProxy extends PeerOptions {
     proxy: Proxy;
 }
-export default abstract class Peer {
-    protected _proxy: Proxy;
-    protected _peer: any;
-    protected _muxer: any;
-    protected _onopen: OnOpenBound;
-    protected _onreceive: OnReceiveBound;
-    protected _onsend: OnSendBound;
-    protected _onclose: OnCloseBound;
-    protected _onchannel: OnChannelBound;
-    protected _emulateWebsocket: boolean;
-    constructor({ proxy, peer, muxer, onopen, onreceive, onsend, onclose, onchannel, emulateWebsocket, }: PeerOptionsWithProxy & DataSocketOptions);
-    protected _socket?: Socket;
+export default class Peer {
+    private _proxy;
+    private _peer;
+    private _muxer;
+    private _onopen;
+    private _onreceive;
+    private _onsend;
+    private _onclose;
+    private _onchannel;
+    private _emulateWebsocket;
+    private _createDefaultMessage;
+    constructor({ proxy, peer, muxer, onopen, onreceive, onsend, onclose, onchannel, emulateWebsocket, createDefaultMessage, }: PeerOptionsWithProxy & DataSocketOptions);
+    private _socket?;
     get socket(): Socket;
-    protected _channel?: any;
+    private _channel?;
     get channel(): any;
-    protected abstract initSocket(): any;
-    protected abstract handleChannelOnOpen(m: any): Promise<void>;
-    protected abstract handleChannelOnClose(socket: Socket): Promise<void>;
-    protected initChannel(): Promise<void>;
     init(): Promise<void>;
-    protected initMessages(): Promise<void>;
 }

dist/peer.js

@@ -1,5 +1,10 @@
 "use strict";
+var __importDefault = (this && this.__importDefault) || function (mod) {
+    return (mod && mod.__esModule) ? mod : { "default": mod };
+};
 Object.defineProperty(exports, "__esModule", { value: true });
+const socket_js_1 = __importDefault(require("./socket.js"));
+const buffer_1 = require("buffer");
 class Peer {
     _proxy;
     _peer;
@@ -10,7 +15,8 @@ class Peer {
     _onclose;
     _onchannel;
     _emulateWebsocket;
-    constructor({ proxy, peer, muxer, onopen, onreceive, onsend, onclose, onchannel, emulateWebsocket = false, }) {
+    _createDefaultMessage;
+    constructor({ proxy, peer, muxer, onopen, onreceive, onsend, onclose, onchannel, emulateWebsocket = false, createDefaultMessage = true, }) {
         this._proxy = proxy;
         this._peer = peer;
         this._muxer = muxer;
@@ -20,6 +26,7 @@ class Peer {
         this._onclose = onclose?.bind(undefined, this);
         this._onchannel = onchannel?.bind(undefined, this);
         this._emulateWebsocket = emulateWebsocket;
+        this._createDefaultMessage = createDefaultMessage;
     }
     _socket;
     get socket() {
@@ -29,21 +36,72 @@ class Peer {
     get channel() {
         return this._channel;
     }
-    async initChannel() {
+    async init() {
         const self = this;
+        let pipe;
+        const raw = await maybeGetAsyncProperty(self._peer.rawStream);
+        this._socket = new socket_js_1.default({
+            remoteAddress: raw.remoteHost,
+            remotePort: raw.remotePort,
+            remotePublicKey: await maybeGetAsyncProperty(self._peer.remotePublicKey),
+            async write(data, cb) {
+                if (pipe) {
+                    pipe.send(data);
+                }
+                await self._onsend?.(data);
+                cb();
+            },
+            emulateWebsocket: self._emulateWebsocket,
+        });
         this._channel = await this._muxer.createChannel({
             protocol: this._proxy.protocol,
-            onopen: this.handleChannelOnOpen.bind(this),
-            onclose: this.handleChannelOnClose.bind(this),
+            async onopen(m) {
+                if (!m) {
+                    m = buffer_1.Buffer.from([]);
+                }
+                if (m instanceof Uint8Array) {
+                    m = buffer_1.Buffer.from(m);
+                }
+                self._socket.on("end", () => this._channel.close());
+                let ret = await self._onopen?.(self._socket, m);
+                if (!ret || (ret && ret.connect === false)) {
+                    // @ts-ignore
+                    self._socket.emit("connect");
+                }
+                self._socket.emit("data", m);
+            },
+            async onclose() {
+                self._socket?.destroy();
+                await self._onclose?.(self._socket);
+            },
         });
-        await this.initMessages();
+        if (this._createDefaultMessage) {
+            pipe = await this._channel.addMessage({
+                async onmessage(m) {
+                    if (m instanceof Uint8Array) {
+                        m = buffer_1.Buffer.from(m);
+                    }
+                    self._socket.emit("data", m);
+                    await self._onreceive?.(m);
+                },
+            });
+        }
         await this._onchannel?.(this._channel);
         await this._channel.open();
     }
-    async init() {
-        await this.initSocket();
-        await this.initChannel();
-    }
-    async initMessages() { }
 }
 exports.default = Peer;
+async function maybeGetAsyncProperty(object) {
+    if (typeof object === "function") {
+        object = object();
+    }
+    if (isPromise(object)) {
+        object = await object;
+    }
+    return object;
+}
+function isPromise(obj) {
+    return (!!obj &&
+        (typeof obj === "object" || typeof obj === "function") &&
+        typeof obj.then === "function");
+}

dist/proxies/basic.d.ts

@@ -1,5 +1,3 @@
 import Proxy from "../proxy.js";
-import { DataSocketOptions, PeerOptions } from "../peer.js";
 export default class BasicProxy extends Proxy {
-    protected handlePeer({ peer, muxer, ...options }: DataSocketOptions & PeerOptions): void;
 }

dist/proxies/basic.js

@@ -4,11 +4,6 @@ var __importDefault = (this && this.__importDefault) || function (mod) {
 };
 Object.defineProperty(exports, "__esModule", { value: true });
 const proxy_js_1 = __importDefault(require("../proxy.js"));
-const peer_js_1 = __importDefault(require("./basic/peer.js"));
 class BasicProxy extends proxy_js_1.default {
-    handlePeer({ peer, muxer, ...options }) {
-        const conn = new peer_js_1.default({ proxy: this, peer, muxer, ...options });
-        conn.init();
-    }
 }
 exports.default = BasicProxy;

dist/proxies/basic/peer.d.ts

@@ -1,9 +0,0 @@
-import BasePeer from "../../peer.js";
-import Socket from "../../socket.js";
-export default class Peer extends BasePeer {
-    private _pipe?;
-    protected initSocket(): Promise<void>;
-    protected handleChannelOnOpen(m: any): Promise<void>;
-    protected handleChannelOnClose(socket: Socket): Promise<void>;
-    protected initMessages(): Promise<void>;
-}

dist/proxies/basic/peer.js

@@ -1,59 +0,0 @@
-"use strict";
-var __importDefault = (this && this.__importDefault) || function (mod) {
-    return (mod && mod.__esModule) ? mod : { "default": mod };
-};
-Object.defineProperty(exports, "__esModule", { value: true });
-const peer_js_1 = __importDefault(require("../../peer.js"));
-const util_js_1 = require("../../util.js");
-const socket_js_1 = __importDefault(require("../../socket.js"));
-const buffer_1 = require("buffer");
-class Peer extends peer_js_1.default {
-    _pipe;
-    async initSocket() {
-        const self = this;
-        const raw = await (0, util_js_1.maybeGetAsyncProperty)(self._peer.rawStream);
-        this._socket = new socket_js_1.default({
-            remoteAddress: raw.remoteHost,
-            remotePort: raw.remotePort,
-            remotePublicKey: await (0, util_js_1.maybeGetAsyncProperty)(self._peer.remotePublicKey),
-            async write(data, cb) {
-                self._pipe?.send(data);
-                await self._onsend?.(data);
-                cb();
-            },
-            emulateWebsocket: self._emulateWebsocket,
-        });
-    }
-    async handleChannelOnOpen(m) {
-        if (!m) {
-            m = buffer_1.Buffer.from([]);
-        }
-        if (m instanceof Uint8Array) {
-            m = buffer_1.Buffer.from(m);
-        }
-        this._socket?.on("end", () => this._channel.close());
-        let ret = await this._onopen?.(this._socket, m);
-        if (!ret || (ret && ret.connect === false)) {
-            // @ts-ignore
-            self._socket?.emit("connect");
-        }
-        this._socket?.emit("data", m);
-    }
-    async handleChannelOnClose(socket) {
-        this._socket?.destroy();
-        await this._onclose?.(this._socket);
-    }
-    async initMessages() {
-        const self = this;
-        this._pipe = await this._channel.addMessage({
-            async onmessage(m) {
-                if (m instanceof Uint8Array) {
-                    m = buffer_1.Buffer.from(m);
-                }
-                self._socket.emit("data", m);
-                await self._onreceive?.(m);
-            },
-        });
-    }
-}
-exports.default = Peer;

dist/proxies/multiSocket.d.ts

@@ -1,7 +1,7 @@
 /// <reference types="node" />
 import Proxy, { ProxyOptions } from "../proxy.js";
 import type { TcpSocketConnectOpts } from "net";
-import Peer, { DataSocketOptions, PeerOptions } from "../peer.js";
+import Peer from "../peer.js";
 import { PeerEntity } from "./multiSocket/types.js";
 export interface MultiSocketProxyOptions extends ProxyOptions {
     socketClass?: any;
@@ -9,7 +9,6 @@ export interface MultiSocketProxyOptions extends ProxyOptions {
     allowedPorts?: number[];
 }
 export default class MultiSocketProxy extends Proxy {
-    handlePeer({ peer, muxer, ...options }: DataSocketOptions & PeerOptions): void;
     private socketClass;
     private _peers;
     private _nextPeer;

dist/proxies/multiSocket.js

@@ -54,14 +54,16 @@ const errorSocketEncoding = {
 };
 const nextSocketId = (0, util_js_1.idFactory)(1);
 class MultiSocketProxy extends proxy_js_1.default {
-    handlePeer({ peer, muxer, ...options }) { }
     socketClass;
     _peers = new Map();
     _nextPeer = (0, util_js_1.roundRobinFactory)(this._peers);
     _server = false;
     _allowedPorts = [];
     constructor(options) {
-        super(options);
+        super({
+            createDefaultMessage: false,
+            ...options,
+        });
         this._socketOptions.onchannel = this.handleNewPeerChannel.bind(this);
         this._socketOptions.onclose = this.handleClosePeer.bind(this);
         this._socketOptions.onopen = this.handlePeer.bind(this);

dist/proxies/multiSocket/dummySocket.d.ts

@@ -1,15 +1,14 @@
 /// <reference types="node" />
-import { Callback } from "streamx";
+import { Callback, Duplex } from "streamx";
 import { TcpSocketConnectOpts } from "net";
 import MultiSocketProxy from "../multiSocket.js";
 import { PeerEntity } from "./types.js";
-import Socket, { SocketOptions } from "../../socket.js";
-export default class DummySocket extends Socket {
+export default class DummySocket extends Duplex {
     private _options;
     private _id;
     private _proxy;
     private _connectTimeout?;
-    constructor(id: number, manager: MultiSocketProxy, peer: PeerEntity, connectOptions: TcpSocketConnectOpts, socketOptions: SocketOptions);
+    constructor(id: number, manager: MultiSocketProxy, peer: PeerEntity, options: TcpSocketConnectOpts);
     private _remoteId;
     set remoteId(value: number);
     private _peer;

dist/proxies/multiSocket/dummySocket.js

@@ -1,22 +1,19 @@
 "use strict";
-var __importDefault = (this && this.__importDefault) || function (mod) {
-    return (mod && mod.__esModule) ? mod : { "default": mod };
-};
 Object.defineProperty(exports, "__esModule", { value: true });
+const streamx_1 = require("streamx");
 const timers_1 = require("timers");
 const util_js_1 = require("../../util.js");
-const socket_js_1 = __importDefault(require("../../socket.js"));
-class DummySocket extends socket_js_1.default {
+class DummySocket extends streamx_1.Duplex {
     _options;
     _id;
     _proxy;
     _connectTimeout;
-    constructor(id, manager, peer, connectOptions, socketOptions) {
-        super(socketOptions);
+    constructor(id, manager, peer, options) {
+        super();
         this._id = id;
         this._proxy = manager;
         this._peer = peer;
-        this._options = connectOptions;
+        this._options = options;
         // @ts-ignore
         this.on("timeout", () => {
             if (this._connectTimeout) {

dist/proxies/multiSocket/peer.d.ts

@@ -1,9 +0,0 @@
-import BasePeer from "../../peer.js";
-import Socket from "../../socket.js";
-import MultiSocketProxy from "../multiSocket.js";
-export default class Peer extends BasePeer {
-    protected _proxy: MultiSocketProxy;
-    protected initSocket(): Promise<void>;
-    protected handleChannelOnClose(socket: Socket): Promise<void>;
-    protected handleChannelOnOpen(m: any): Promise<void>;
-}

dist/proxies/multiSocket/peer.js

@@ -1,16 +0,0 @@
-"use strict";
-var __importDefault = (this && this.__importDefault) || function (mod) {
-    return (mod && mod.__esModule) ? mod : { "default": mod };
-};
-Object.defineProperty(exports, "__esModule", { value: true });
-const peer_js_1 = __importDefault(require("../../peer.js"));
-class Peer extends peer_js_1.default {
-    async initSocket() { }
-    async handleChannelOnClose(socket) {
-        return this._proxy.handleClosePeer(this);
-    }
-    async handleChannelOnOpen(m) {
-        await this._proxy.handleNewPeerChannel(this._peer, this._channel);
-    }
-}
-exports.default = Peer;

dist/proxies/multiSocket/tcpSocket.d.ts

@@ -1,10 +1,9 @@
 /// <reference types="node" />
-import { Callback } from "streamx";
+import { Callback, Duplex } from "streamx";
 import { TcpSocketConnectOpts } from "net";
 import MultiSocketProxy from "../multiSocket.js";
 import { PeerEntity } from "./types.js";
-import BaseSocket from "../../socket.js";
-export default class TcpSocket extends BaseSocket {
+export default class TcpSocket extends Duplex {
     private _options;
     private _id;
     private _remoteId;

dist/proxies/multiSocket/tcpSocket.js

@@ -22,13 +22,10 @@ var __importStar = (this && this.__importStar) || function (mod) {
     __setModuleDefault(result, mod);
     return result;
 };
-var __importDefault = (this && this.__importDefault) || function (mod) {
-    return (mod && mod.__esModule) ? mod : { "default": mod };
-};
 Object.defineProperty(exports, "__esModule", { value: true });
+const streamx_1 = require("streamx");
 const net = __importStar(require("net"));
-const socket_js_1 = __importDefault(require("../../socket.js"));
-class TcpSocket extends socket_js_1.default {
+class TcpSocket extends streamx_1.Duplex {
     _options;
     _id;
     _remoteId;

dist/proxy.d.ts

@@ -1,4 +1,4 @@
-import { DataSocketOptions, PeerOptions } from "./peer.js";
+import { PeerOptions, DataSocketOptions } from "./peer.js";
 export interface ProxyOptions extends DataSocketOptions {
     swarm: any;
     protocol: string;
@@ -8,14 +8,14 @@ export interface ProxyOptions extends DataSocketOptions {
 export default abstract class Proxy {
     protected _listen: any;
     protected _autostart: boolean;
-    constructor({ swarm, protocol, onopen, onreceive, onsend, onclose, onchannel, listen, autostart, emulateWebsocket, }: ProxyOptions);
+    constructor({ swarm, protocol, onopen, onreceive, onsend, onclose, onchannel, listen, autostart, emulateWebsocket, createDefaultMessage, }: ProxyOptions);
     protected _socketOptions: DataSocketOptions;
     get socketOptions(): DataSocketOptions;
     private _swarm;
     get swarm(): any;
     private _protocol;
     get protocol(): string;
-    protected abstract handlePeer({ peer, muxer, ...options }: DataSocketOptions & PeerOptions): any;
+    handlePeer({ peer, muxer, ...options }: DataSocketOptions & PeerOptions): void;
     protected _init(): void;
     private init;
     private _handleConnection;

dist/proxy.js

@@ -4,10 +4,11 @@ var __importDefault = (this && this.__importDefault) || function (mod) {
 };
 Object.defineProperty(exports, "__esModule", { value: true });
 const protomux_1 = __importDefault(require("protomux"));
+const peer_js_1 = __importDefault(require("./peer.js"));
 class Proxy {
     _listen;
     _autostart;
-    constructor({ swarm, protocol, onopen, onreceive, onsend, onclose, onchannel, listen = false, autostart = false, emulateWebsocket = false, }) {
+    constructor({ swarm, protocol, onopen, onreceive, onsend, onclose, onchannel, listen = false, autostart = false, emulateWebsocket = false, createDefaultMessage = true, }) {
         this._swarm = swarm;
         this._protocol = protocol;
         this._listen = listen;
@@ -19,6 +20,7 @@ class Proxy {
             onclose,
             onchannel,
             emulateWebsocket,
+            createDefaultMessage,
         };
         this.init();
     }
@@ -34,6 +36,10 @@ class Proxy {
     get protocol() {
         return this._protocol;
     }
+    handlePeer({ peer, muxer, ...options }) {
+        const conn = new peer_js_1.default({ proxy: this, peer, muxer, ...options });
+        conn.init();
+    }
     _init() {
         // Implement in subclasses
     }

dist/socket.d.ts

@@ -1,6 +1,6 @@
 import { Duplex, DuplexEvents, Callback } from "streamx";
 type AddressFamily = "IPv6" | "IPv4";
-export interface SocketOptions {
+interface SocketOptions {
     allowHalfOpen?: boolean;
     remoteAddress?: string;
     remotePort?: number;

src/peer.ts

@@ -1,7 +1,6 @@
 import Proxy from "./proxy.js";
 import Socket from "./socket.js";
 import { Buffer } from "buffer";
-import { maybeGetAsyncProperty } from "./util.js";
 export type OnOpen = (
   peer: Peer,
   socket: Socket,
@@ -39,6 +38,7 @@ export interface DataSocketOptions {
   onclose?: OnClose;
   onchannel?: OnChannel;
   emulateWebsocket?: boolean;
+  createDefaultMessage?: boolean;
 }
 
 export interface PeerOptions {
@@ -50,16 +50,17 @@ export interface PeerOptionsWithProxy extends PeerOptions {
   proxy: Proxy;
 }
 
-export default abstract class Peer {
-  protected _proxy: Proxy;
-  protected _peer: any;
-  protected _muxer: any;
-  protected _onopen: OnOpenBound;
-  protected _onreceive: OnReceiveBound;
-  protected _onsend: OnSendBound;
-  protected _onclose: OnCloseBound;
-  protected _onchannel: OnChannelBound;
-  protected _emulateWebsocket: boolean;
+export default class Peer {
+  private _proxy: Proxy;
+  private _peer: any;
+  private _muxer: any;
+  private _onopen: OnOpenBound;
+  private _onreceive: OnReceiveBound;
+  private _onsend: OnSendBound;
+  private _onclose: OnCloseBound;
+  private _onchannel: OnChannelBound;
+  private _emulateWebsocket: boolean;
+  private _createDefaultMessage: boolean;
 
   constructor({
     proxy,
@@ -71,6 +72,7 @@ export default abstract class Peer {
     onclose,
     onchannel,
     emulateWebsocket = false,
+    createDefaultMessage = true,
   }: PeerOptionsWithProxy & DataSocketOptions) {
     this._proxy = proxy;
     this._peer = peer;
@@ -81,44 +83,98 @@ export default abstract class Peer {
     this._onclose = onclose?.bind(undefined, this);
     this._onchannel = onchannel?.bind(undefined, this);
     this._emulateWebsocket = emulateWebsocket;
+    this._createDefaultMessage = createDefaultMessage;
   }
 
-  protected _socket?: Socket;
+  private _socket?: Socket;
 
   get socket(): Socket {
     return this._socket;
   }
 
-  protected _channel?: any;
+  private _channel?: any;
 
   get channel(): any {
     return this._channel;
   }
 
-  protected abstract initSocket();
-
-  protected abstract handleChannelOnOpen(m: any): Promise<void>;
-  protected abstract handleChannelOnClose(socket: Socket): Promise<void>;
-
-  protected async initChannel() {
+  async init() {
     const self = this;
+    let pipe;
+    const raw = await maybeGetAsyncProperty(self._peer.rawStream);
+    this._socket = new Socket({
+      remoteAddress: raw.remoteHost,
+      remotePort: raw.remotePort,
+      remotePublicKey: await maybeGetAsyncProperty(self._peer.remotePublicKey),
+      async write(data: any, cb: Function) {
+        if (pipe) {
+          pipe.send(data);
+        }
+        await self._onsend?.(data);
+        cb();
+      },
+      emulateWebsocket: self._emulateWebsocket,
+    });
 
     this._channel = await this._muxer.createChannel({
       protocol: this._proxy.protocol,
-      onopen: this.handleChannelOnOpen.bind(this),
-      onclose: this.handleChannelOnClose.bind(this),
+      async onopen(m: any) {
+        if (!m) {
+          m = Buffer.from([]);
+        }
+
+        if (m instanceof Uint8Array) {
+          m = Buffer.from(m);
+        }
+
+        self._socket.on("end", () => this._channel.close());
+        let ret = await self._onopen?.(self._socket, m);
+        if (!ret || (ret && ret.connect === false)) {
+          // @ts-ignore
+          self._socket.emit("connect");
+        }
+
+        self._socket.emit("data", m);
+      },
+      async onclose() {
+        self._socket?.destroy();
+        await self._onclose?.(self._socket);
+      },
     });
 
-    await this.initMessages();
+    if (this._createDefaultMessage) {
+      pipe = await this._channel.addMessage({
+        async onmessage(m: any) {
+          if (m instanceof Uint8Array) {
+            m = Buffer.from(m);
+          }
+          self._socket.emit("data", m);
+          await self._onreceive?.(m);
+        },
+      });
+    }
 
     await this._onchannel?.(this._channel);
     await this._channel.open();
   }
+}
 
-  async init() {
-    await this.initSocket();
-    await this.initChannel();
+async function maybeGetAsyncProperty(object: any) {
+  if (typeof object === "function") {
+    object = object();
   }
 
-  protected async initMessages() {}
+  if (isPromise(object)) {
+    object = await object;
+  }
+
+  return object;
+}
+
+function isPromise(obj: Promise<any>) {
+  return (
+    !!obj &&
+    (typeof obj === "object" || typeof obj === "function") &&
+    typeof obj.then === "function"
+  );
 }

src/proxies/basic.ts

@@ -1,14 +1,3 @@
 import Proxy from "../proxy.js";
-import { DataSocketOptions, PeerOptions } from "../peer.js";
-import BasicPeer from "./basic/peer.js";
 
-export default class BasicProxy extends Proxy {
-  protected handlePeer({
-    peer,
-    muxer,
-    ...options
-  }: DataSocketOptions & PeerOptions) {
-    const conn = new BasicPeer({ proxy: this, peer, muxer, ...options });
-    conn.init();
-  }
-}
+export default class BasicProxy extends Proxy {}

src/proxies/basic/peer.ts

@@ -1,61 +0,0 @@
-import BasePeer from "../../peer.js";
-import { maybeGetAsyncProperty } from "../../util.js";
-import Socket from "../../socket.js";
-import { Buffer } from "buffer";
-
-export default class Peer extends BasePeer {
-  private _pipe?: any;
-  protected async initSocket() {
-    const self = this;
-
-    const raw = await maybeGetAsyncProperty(self._peer.rawStream);
-    this._socket = new Socket({
-      remoteAddress: raw.remoteHost,
-      remotePort: raw.remotePort,
-      remotePublicKey: await maybeGetAsyncProperty(self._peer.remotePublicKey),
-      async write(data: any, cb: Function) {
-        self._pipe?.send(data);
-        await self._onsend?.(data);
-        cb();
-      },
-      emulateWebsocket: self._emulateWebsocket,
-    });
-  }
-
-  protected async handleChannelOnOpen(m: any) {
-    if (!m) {
-      m = Buffer.from([]);
-    }
-
-    if (m instanceof Uint8Array) {
-      m = Buffer.from(m);
-    }
-
-    this._socket?.on("end", () => this._channel.close());
-    let ret = await this._onopen?.(this._socket, m);
-    if (!ret || (ret && ret.connect === false)) {
-      // @ts-ignore
-      self._socket?.emit("connect");
-    }
-
-    this._socket?.emit("data", m);
-  }
-
-  protected async handleChannelOnClose(socket: Socket): Promise<void> {
-    this._socket?.destroy();
-    await this._onclose?.(this._socket);
-  }
-
-  protected async initMessages(): Promise<void> {
-    const self = this;
-    this._pipe = await this._channel.addMessage({
-      async onmessage(m: any) {
-        if (m instanceof Uint8Array) {
-          m = Buffer.from(m);
-        }
-        self._socket.emit("data", m);
-        await self._onreceive?.(m);
-      },
-    });
-  }
-}

src/proxies/multiSocket.ts

@@ -4,7 +4,7 @@ import { json, raw, uint } from "compact-encoding";
 import { deserializeError } from "serialize-error";
 import b4a from "b4a";
 import type { TcpSocketConnectOpts } from "net";
-import Peer, { DataSocketOptions, PeerOptions } from "../peer.js";
+import Peer from "../peer.js";
 import { roundRobinFactory, idFactory } from "../util.js";
 import {
   CloseSocketRequest,
@@ -69,7 +69,6 @@ const errorSocketEncoding = {
 const nextSocketId = idFactory(1);
 
 export default class MultiSocketProxy extends Proxy {
-  handlePeer({ peer, muxer, ...options }: DataSocketOptions & PeerOptions) {}
   private socketClass: any;
   private _peers: Map<string, PeerEntity> = new Map<string, PeerEntity>();
   private _nextPeer = roundRobinFactory(this._peers);
@@ -77,7 +76,10 @@ export default class MultiSocketProxy extends Proxy {
   private _allowedPorts = [];
 
   constructor(options: MultiSocketProxyOptions) {
-    super(options);
+    super({
+      createDefaultMessage: false,
+      ...options,
+    });
     this._socketOptions.onchannel = this.handleNewPeerChannel.bind(this);
     this._socketOptions.onclose = this.handleClosePeer.bind(this);
     this._socketOptions.onopen = this.handlePeer.bind(this);

src/proxies/multiSocket/dummySocket.ts

@@ -4,9 +4,8 @@ import { clearTimeout } from "timers";
 import MultiSocketProxy from "../multiSocket.js";
 import { PeerEntity, SocketRequest, WriteSocketRequest } from "./types.js";
 import { maybeGetAsyncProperty } from "../../util.js";
-import Socket, { SocketOptions } from "../../socket.js";
 
-export default class DummySocket extends Socket {
+export default class DummySocket extends Duplex {
   private _options: TcpSocketConnectOpts;
   private _id: number;
   private _proxy: MultiSocketProxy;
@@ -17,14 +16,13 @@ export default class DummySocket extends Socket {
     id: number,
     manager: MultiSocketProxy,
     peer: PeerEntity,
-    connectOptions: TcpSocketConnectOpts,
-    socketOptions: SocketOptions
+    options: TcpSocketConnectOpts
   ) {
-    super(socketOptions);
+    super();
     this._id = id;
     this._proxy = manager;
     this._peer = peer;
-    this._options = connectOptions;
+    this._options = options;
 
     // @ts-ignore
     this.on("timeout", () => {

src/proxies/multiSocket/peer.ts

@@ -1,16 +0,0 @@
-import BasePeer from "../../peer.js";
-import Socket from "../../socket.js";
-import MultiSocketProxy from "../multiSocket.js";
-
-export default class Peer extends BasePeer {
-  protected declare _proxy: MultiSocketProxy;
-  protected async initSocket() {}
-
-  protected async handleChannelOnClose(socket: Socket): Promise<void> {
-    return this._proxy.handleClosePeer(this);
-  }
-
-  protected async handleChannelOnOpen(m: any): Promise<void> {
-    await this._proxy.handleNewPeerChannel(this._peer, this._channel);
-  }
-}

src/proxies/multiSocket/tcpSocket.ts

@@ -3,9 +3,8 @@ import { Socket, TcpSocketConnectOpts } from "net";
 import MultiSocketProxy from "../multiSocket.js";
 import { PeerEntity, SocketRequest, WriteSocketRequest } from "./types.js";
 import * as net from "net";
-import BaseSocket from "../../socket.js";
 
-export default class TcpSocket extends BaseSocket {
+export default class TcpSocket extends Duplex {
   private _options;
   private _id: number;
   private _remoteId: number;

src/proxy.ts

@@ -1,5 +1,5 @@
 import Protomux from "protomux";
-import { DataSocketOptions, PeerOptions } from "./peer.js";
+import Peer, { PeerOptions, DataSocketOptions } from "./peer.js";
 
 export interface ProxyOptions extends DataSocketOptions {
   swarm: any;
@@ -23,6 +23,7 @@ export default abstract class Proxy {
     listen = false,
     autostart = false,
     emulateWebsocket = false,
+    createDefaultMessage = true,
   }: ProxyOptions) {
     this._swarm = swarm;
     this._protocol = protocol;
@@ -35,6 +36,7 @@ export default abstract class Proxy {
       onclose,
       onchannel,
       emulateWebsocket,
+      createDefaultMessage,
     };
     this.init();
   }
@@ -56,12 +58,14 @@ export default abstract class Proxy {
   get protocol(): string {
     return this._protocol;
   }
-
-  protected abstract handlePeer({
+  public handlePeer({
     peer,
     muxer,
     ...options
-  }: DataSocketOptions & PeerOptions);
+  }: DataSocketOptions & PeerOptions) {
+    const conn = new Peer({ proxy: this, peer, muxer, ...options });
+    conn.init();
+  }
 
   protected _init() {
     // Implement in subclasses

src/socket.ts

@@ -6,7 +6,7 @@ const IPV6 = "IPv6";
 
 type AddressFamily = "IPv6" | "IPv4";
 
-export interface SocketOptions {
+interface SocketOptions {
   allowHalfOpen?: boolean;
   remoteAddress?: string;
   remotePort?: number;