Implement peer package to send messages between peers (#279)

* Initial implementation of class and discoving other peers * Implement peer connection and sending messages between peers * Add react app and use peer package for broadcasting * Maintain stream for each remote peer * Refactor code in peer package * Add serve package for react-app * Add readme for running react app * Add peer package readme * Add logs for events with details * Add a chat CLI using peer package (#280) * Add a flag to instantiate Peer for nodejs * Add a basic chat CLI using peer * Add a signal server arg to chat CLI * Add instructions for chat CLI * Fix typescript and ESM issues after adding peer package (#282) * Fix build issues in util package * Update eslint TS plugins * Scope react app package name * Convert cli package back to CJS and dynamically import ESM * Upgrade ts-node version * Fix tests * Setup a relay node and pubsub based discovery (#284) * Add a script to setup a relay node * Use pubsub based peer discovery * Add peer multiaddr to connection log * Catch relay node dial errors * Increase discovery interval and dial all mutiaddr * Add UI to display self peer ID and multiaddrs * Add UI for displaying live remote connections * Send js objects in peer broadcast messages * Add react-peer package for using peer in react app * Reduce disconnect frequency between peers (#287) * Restrict number of max concurrent dials per peer * Increase hop relay timeout to 1 day * Self review changes * Review changes * Increase pubsub discovery interval and add steps to create a peer id (#290) * Increase pubsub discovery interval * Disable autodial to avoid peer dials without protocol on a reconnect * Add steps to create a peer id and use for relay node * Add back dependency to run signalling server * Avoid bootstrapping and dial to relay node directly Co-authored-by: prathamesh0 <42446521+prathamesh0@users.noreply.github.com> Co-authored-by: prathamesh0 <prathamesh.musale0@gmail.com>
2026-09-07 16:34:06 +00:00 · 2023-01-10 20:10:27 +05:30
parent aa4a954330
commit cd29b47ecc
82 changed files with 17136 additions and 2475 deletions
@@ -0,0 +1,6 @@
+//
+// Copyright 2023 Vulcanize, Inc.
+//
+
+export const PUBSUB_DISCOVERY_INTERVAL = 10000; // 10 seconds
+export const HOP_TIMEOUT = 24 * 60 * 60 * 1000; // 1 day
@@ -0,0 +1,55 @@
+//
+// Copyright 2022 Vulcanize, Inc.
+//
+
+import assert from 'assert';
+import fs from 'fs';
+import path from 'path';
+import { hideBin } from 'yargs/helpers';
+import yargs from 'yargs';
+
+import { createEd25519PeerId } from '@libp2p/peer-id-factory';
+
+interface Arguments {
+  file: string;
+}
+
+async function main (): Promise<void> {
+  const argv: Arguments = _getArgv();
+
+  const exportFilePath = path.resolve(argv.file);
+  const exportFileDir = path.dirname(exportFilePath);
+
+  const peerId = await createEd25519PeerId();
+  assert(peerId.privateKey);
+
+  const obj = {
+    id: peerId.toString(),
+    privKey: Buffer.from(peerId.privateKey).toString('base64'),
+    pubKey: Buffer.from(peerId.publicKey).toString('base64')
+  };
+
+  if (!fs.existsSync(exportFileDir)) {
+    fs.mkdirSync(exportFileDir, { recursive: true });
+  }
+
+  fs.writeFileSync(exportFilePath, JSON.stringify(obj));
+  console.log(`Peer id ${peerId.toString()} exported to file ${exportFilePath}`);
+}
+
+function _getArgv (): any {
+  return yargs(hideBin(process.argv)).parserConfiguration({
+    'parse-numbers': false
+  }).options({
+    file: {
+      type: 'string',
+      alias: 'f',
+      describe: 'Peer Id export file path (json)',
+      demandOption: true
+    }
+  }).argv;
+}
+
+main().catch(err => {
+  console.log(err);
+});
@@ -0,0 +1,276 @@
+//
+// Copyright 2022 Vulcanize, Inc.
+//
+
+import { createLibp2p, Libp2p } from 'libp2p';
+// For nodejs.
+import wrtc from 'wrtc';
+import assert from 'assert';
+import { pipe } from 'it-pipe';
+import * as lp from 'it-length-prefixed';
+import map from 'it-map';
+import { pushable, Pushable } from 'it-pushable';
+import { fromString as uint8ArrayFromString } from 'uint8arrays/from-string';
+import { toString as uint8ArrayToString } from 'uint8arrays/to-string';
+
+import { webRTCStar, WebRTCStarTuple } from '@libp2p/webrtc-star';
+import { noise } from '@chainsafe/libp2p-noise';
+import { mplex } from '@libp2p/mplex';
+import type { Stream as P2PStream, Connection } from '@libp2p/interface-connection';
+import type { PeerInfo } from '@libp2p/interface-peer-info';
+import { PeerId } from '@libp2p/interface-peer-id';
+import { multiaddr, Multiaddr } from '@multiformats/multiaddr';
+import { floodsub } from '@libp2p/floodsub';
+import { pubsubPeerDiscovery } from '@libp2p/pubsub-peer-discovery';
+
+import { PUBSUB_DISCOVERY_INTERVAL } from './constants.js';
+
+export const CHAT_PROTOCOL = '/chat/1.0.0';
+export const DEFAULT_SIGNAL_SERVER_URL = '/ip4/127.0.0.1/tcp/13579/wss/p2p-webrtc-star';
+
+export class Peer {
+  _node?: Libp2p
+  _wrtcStar: WebRTCStarTuple
+  _relayNodeMultiaddr?: Multiaddr
+
+  _remotePeerIds: PeerId[] = []
+  _peerStreamMap: Map<string, Pushable<any>> = new Map()
+  _messageHandlers: Array<(peerId: PeerId, message: any) => void> = []
+
+  constructor (nodejs?: boolean) {
+    // Instantiation in nodejs.
+    if (nodejs) {
+      this._wrtcStar = webRTCStar({ wrtc });
+    } else {
+      this._wrtcStar = webRTCStar();
+    }
+  }
+
+  get peerId (): PeerId | undefined {
+    return this._node?.peerId;
+  }
+
+  get node (): Libp2p | undefined {
+    return this._node;
+  }
+
+  async init (signalServerURL = DEFAULT_SIGNAL_SERVER_URL, relayNodeURL?: string): Promise<void> {
+    let peerDiscovery: any;
+    if (relayNodeURL) {
+      this._relayNodeMultiaddr = multiaddr(relayNodeURL);
+
+      peerDiscovery = [
+        pubsubPeerDiscovery({
+          interval: PUBSUB_DISCOVERY_INTERVAL
+        })
+      ];
+    } else {
+      peerDiscovery = [this._wrtcStar.discovery];
+    }
+
+    this._node = await createLibp2p({
+      addresses: {
+        // Add the signaling server address, along with our PeerId to our multiaddrs list
+        // libp2p will automatically attempt to dial to the signaling server so that it can
+        // receive inbound connections from other peers
+        listen: [
+          // Public signal servers
+          // '/dns4/wrtc-star1.par.dwebops.pub/tcp/443/wss/p2p-webrtc-star',
+          // '/dns4/wrtc-star2.sjc.dwebops.pub/tcp/443/wss/p2p-webrtc-star'
+          signalServerURL
+        ]
+      },
+      transports: [
+        this._wrtcStar.transport
+      ],
+      connectionEncryption: [noise()],
+      streamMuxers: [mplex()],
+      pubsub: floodsub(),
+      peerDiscovery,
+      relay: {
+        enabled: true,
+        autoRelay: {
+          enabled: true,
+          maxListeners: 2
+        }
+      },
+      connectionManager: {
+        maxDialsPerPeer: 3, // Number of max concurrent dials per peer
+        autoDial: false
+      }
+    });
+
+    console.log('libp2p node created', this._node);
+
+    // Dial to the HOP enabled relay node if available
+    if (this._relayNodeMultiaddr) {
+      const relayMultiaddr = this._relayNodeMultiaddr;
+
+      console.log(`Dialling relay node ${relayMultiaddr.getPeerId()} using multiaddr ${relayMultiaddr.toString()}`);
+      await this._node.dial(relayMultiaddr);
+    }
+
+    // Listen for change in stored multiaddrs
+    this._node.peerStore.addEventListener('change:multiaddrs', (evt) => {
+      assert(this._node);
+      const { peerId, multiaddrs } = evt.detail;
+
+      // Log updated self multiaddrs
+      if (peerId.equals(this._node.peerId)) {
+        console.log('Updated self multiaddrs', this._node.getMultiaddrs().map(addr => addr.toString()));
+      } else {
+        console.log('Updated peer node multiaddrs', multiaddrs.map((addr: Multiaddr) => addr.toString()));
+      }
+    });
+
+    // Listen for peers discovery
+    this._node.addEventListener('peer:discovery', (evt) => {
+      // console.log('event peer:discovery', evt);
+      this._handleDiscovery(evt.detail);
+    });
+
+    // Listen for peers connection
+    this._node.connectionManager.addEventListener('peer:connect', (evt) => {
+      console.log('event peer:connect', evt);
+      this._handleConnect(evt.detail);
+    });
+
+    // Listen for peers disconnecting
+    this._node.connectionManager.addEventListener('peer:disconnect', (evt) => {
+      console.log('event peer:disconnect', evt);
+      this._handleDisconnect(evt.detail);
+    });
+
+    // Handle messages for the protocol
+    await this._node.handle(CHAT_PROTOCOL, async ({ stream, connection }) => {
+      this._handleStream(connection.remotePeer, stream);
+    });
+  }
+
+  async close (): Promise<void> {
+    assert(this._node);
+
+    this._node.removeEventListener('peer:discovery');
+    this._node.connectionManager.removeEventListener('peer:connect');
+    this._node.connectionManager.removeEventListener('peer:disconnect');
+
+    await this._node.unhandle(CHAT_PROTOCOL);
+    const hangUpPromises = this._remotePeerIds.map(async peerId => this._node?.hangUp(peerId));
+    await Promise.all(hangUpPromises);
+  }
+
+  broadcastMessage (message: any): void {
+    for (const [, stream] of this._peerStreamMap) {
+      stream.push(message);
+    }
+  }
+
+  subscribeMessage (handler: (peerId: PeerId, message: any) => void) : () => void {
+    this._messageHandlers.push(handler);
+
+    const unsubscribe = () => {
+      this._messageHandlers = this._messageHandlers
+        .filter(registeredHandler => registeredHandler !== handler);
+    };
+
+    return unsubscribe;
+  }
+
+  _handleDiscovery (peer: PeerInfo): void {
+    // Check connected peers as they are discovered repeatedly.
+    if (!this._remotePeerIds.some(remotePeerId => remotePeerId.toString() === peer.id.toString())) {
+      console.log('Discovered peer multiaddrs', peer.multiaddrs.map(addr => addr.toString()));
+      this._connectPeer(peer);
+    }
+  }
+
+  _handleConnect (connection: Connection): void {
+    const remotePeerId = connection.remotePeer;
+    this._remotePeerIds.push(remotePeerId);
+
+    // Log connected peer
+    console.log(`Connected to ${remotePeerId.toString()} using multiaddr ${connection.remoteAddr.toString()}`);
+  }
+
+  _handleDisconnect (connection: Connection): void {
+    const disconnectedPeerId = connection.remotePeer;
+    this._remotePeerIds = this._remotePeerIds.filter(remotePeerId => remotePeerId.toString() !== disconnectedPeerId.toString());
+
+    // Log disconnected peer
+    console.log(`Disconnected from ${disconnectedPeerId.toString()} using multiaddr ${connection.remoteAddr.toString()}`);
+  }
+
+  async _connectPeer (peer: PeerInfo): Promise<void> {
+    assert(this._node);
+
+    // Check if discovered the relay node
+    if (this._relayNodeMultiaddr) {
+      const relayMultiaddr = this._relayNodeMultiaddr;
+      const relayNodePeerId = relayMultiaddr.getPeerId();
+
+      if (relayNodePeerId && relayNodePeerId === peer.id.toString()) {
+        console.log(`Dialling relay peer ${peer.id.toString()} using multiaddr ${relayMultiaddr.toString()}`);
+        await this._node.dial(relayMultiaddr).catch(err => {
+          console.log(`Could not dial relay ${relayMultiaddr.toString()}`, err);
+        });
+
+        return;
+      }
+    }
+
+    // Dial them when we discover them
+    // Attempt to dial all the multiaddrs of the discovered peer (to connect through relay)
+    for (const peerMultiaddr of peer.multiaddrs) {
+      try {
+        console.log(`Dialling peer ${peer.id.toString()} using multiaddr ${peerMultiaddr.toString()}`);
+        const stream = await this._node.dialProtocol(peerMultiaddr, CHAT_PROTOCOL);
+
+        this._handleStream(peer.id, stream);
+        break;
+      } catch (err) {
+        console.log(`Could not dial ${peerMultiaddr.toString()}`, err);
+      }
+    }
+  }
+
+  _handleStream (peerId: PeerId, stream: P2PStream): void {
+    // console.log('Stream after connection', stream);
+    const messageStream = pushable<any>({ objectMode: true });
+
+    // Send message to pipe from stdin
+    pipe(
+      // Read from stream (the source)
+      messageStream,
+      // Turn objects into buffers
+      (source) => map(source, (value) => {
+        return uint8ArrayFromString(JSON.stringify(value));
+      }),
+      // Encode with length prefix (so receiving side knows how much data is coming)
+      lp.encode(),
+      // Write to the stream (the sink)
+      stream.sink
+    );
+
+    // Handle message from stream
+    pipe(
+      // Read from the stream (the source)
+      stream.source,
+      // Decode length-prefixed data
+      lp.decode(),
+      // Turn buffers into objects
+      (source) => map(source, (buf) => {
+        return JSON.parse(uint8ArrayToString(buf.subarray()));
+      }),
+      // Sink function
+      async (source) => {
+        // For each chunk of data
+        for await (const msg of source) {
+          this._messageHandlers.forEach(messageHandler => messageHandler(peerId, msg));
+        }
+      }
+    );
+
+    // TODO: Check if stream already exists for peer id
+    this._peerStreamMap.set(peerId.toString(), messageStream);
+  }
+}
@@ -0,0 +1,98 @@
+//
+// Copyright 2022 Vulcanize, Inc.
+//
+
+import { createLibp2p } from 'libp2p';
+import wrtc from 'wrtc';
+import { hideBin } from 'yargs/helpers';
+import yargs from 'yargs';
+import fs from 'fs';
+import path from 'path';
+
+import { noise } from '@chainsafe/libp2p-noise';
+import { mplex } from '@libp2p/mplex';
+import { webRTCStar, WebRTCStarTuple } from '@libp2p/webrtc-star';
+import { floodsub } from '@libp2p/floodsub';
+import { pubsubPeerDiscovery } from '@libp2p/pubsub-peer-discovery';
+import { createFromJSON } from '@libp2p/peer-id-factory';
+
+import { DEFAULT_SIGNAL_SERVER_URL } from './index.js';
+import { HOP_TIMEOUT, PUBSUB_DISCOVERY_INTERVAL } from './constants.js';
+
+interface Arguments {
+  signalServer: string;
+  peerIdFile: string;
+}
+
+async function main (): Promise<void> {
+  const argv: Arguments = _getArgv();
+  if (!argv.signalServer) {
+    console.log('Using the default signalling server URL');
+  }
+
+  let peerId: any;
+  if (argv.peerIdFile) {
+    const peerIdFilePath = path.resolve(argv.peerIdFile);
+    console.log(`Reading peer id from file ${peerIdFilePath}`);
+
+    const peerIdObj = fs.readFileSync(peerIdFilePath, 'utf-8');
+    const peerIdJson = JSON.parse(peerIdObj);
+    peerId = await createFromJSON(peerIdJson);
+  } else {
+    console.log('Creating a new peer id');
+  }
+
+  const wrtcStar: WebRTCStarTuple = webRTCStar({ wrtc });
+  const node = await createLibp2p({
+    peerId,
+    addresses: {
+      listen: [
+        argv.signalServer || DEFAULT_SIGNAL_SERVER_URL
+      ]
+    },
+    transports: [
+      wrtcStar.transport
+    ],
+    connectionEncryption: [noise()],
+    streamMuxers: [mplex()],
+    pubsub: floodsub(),
+    peerDiscovery: [
+      pubsubPeerDiscovery({
+        interval: PUBSUB_DISCOVERY_INTERVAL
+      })
+    ],
+    relay: {
+      enabled: true,
+      hop: {
+        enabled: true,
+        timeout: HOP_TIMEOUT
+      },
+      advertise: {
+        enabled: true
+      }
+    }
+  });
+
+  console.log(`Relay node started with id ${node.peerId.toString()}`);
+  console.log('Listening on:');
+  node.getMultiaddrs().forEach((ma) => console.log(ma.toString()));
+}
+
+function _getArgv (): any {
+  return yargs(hideBin(process.argv)).parserConfiguration({
+    'parse-numbers': false
+  }).options({
+    signalServer: {
+      type: 'string',
+      describe: 'Signalling server URL'
+    },
+    peerIdFile: {
+      type: 'string',
+      describe: 'Relay Peer Id file path (json)'
+    }
+  }).argv;
+}
+
+main().catch(err => {
+  console.log(err);
+});
@@ -0,0 +1,5 @@
+//
+// Copyright 2022 Vulcanize, Inc.
+//
+
+declare module 'wrtc';
@@ -0,0 +1,6 @@
+{
+  "name": "common",
+  "version": "0.1.0",
+  "license": "AGPL-3.0",
+  "typings": "main.d.ts"
+}