fix: dont disconnect peer on inbound stream close

BREAKING CHANGE:
new getMsgId return type is not backwards compatible with prior `string`
return type.

CHANGELOG.md

@@ -1,3 +1,82 @@
+<a name="0.6.0"></a>
+# [0.6.0](https://github.com/libp2p/js-interfaces/compare/v0.5.2...v0.6.0) (2020-10-05)
+
+
+### Features
+
+* update pubsub getMsgId return type to Uint8Array ([#65](https://github.com/libp2p/js-interfaces/issues/65)) ([e148443](https://github.com/libp2p/js-interfaces/commit/e148443))
+
+
+### BREAKING CHANGES
+
+* new getMsgId return type is not backwards compatible with prior `string`
+return type.
+
+
+
+<a name="0.5.2"></a>
+## [0.5.2](https://github.com/libp2p/js-interfaces/compare/v0.3.1...v0.5.2) (2020-09-30)
+
+
+### Bug Fixes
+
+* replace remaining Buffer usage with Uint8Array ([#62](https://github.com/libp2p/js-interfaces/issues/62)) ([4130e7f](https://github.com/libp2p/js-interfaces/commit/4130e7f))
+
+
+### Chores
+
+* update deps ([#57](https://github.com/libp2p/js-interfaces/issues/57)) ([75f6777](https://github.com/libp2p/js-interfaces/commit/75f6777))
+
+
+### Features
+
+* interface pubsub ([#60](https://github.com/libp2p/js-interfaces/issues/60)) ([ba15a48](https://github.com/libp2p/js-interfaces/commit/ba15a48))
+* record interface ([#52](https://github.com/libp2p/js-interfaces/issues/52)) ([1cc943e](https://github.com/libp2p/js-interfaces/commit/1cc943e))
+
+
+### BREAKING CHANGES
+
+* records now marshal as Uint8Array instead of Buffer
+
+* fix: refactor remaining Buffer usage to Uint8Array
+* - The peer id dep of this module has replaced node Buffers with Uint8Arrays
+
+* chore: update gh deps
+
+
+
+<a name="0.5.1"></a>
+## [0.5.1](https://github.com/libp2p/js-interfaces/compare/v0.5.0...v0.5.1) (2020-08-25)
+
+
+### Features
+
+* interface pubsub ([#60](https://github.com/libp2p/js-interfaces/issues/60)) ([ba15a48](https://github.com/libp2p/js-interfaces/commit/ba15a48))
+
+
+
+<a name="0.5.0"></a>
+# [0.5.0](https://github.com/libp2p/js-interfaces/compare/v0.4.1...v0.5.0) (2020-08-24)
+
+
+### Bug Fixes
+
+* replace remaining Buffer usage with Uint8Array ([#62](https://github.com/libp2p/js-interfaces/issues/62)) ([4130e7f](https://github.com/libp2p/js-interfaces/commit/4130e7f))
+
+
+### BREAKING CHANGES
+
+* records now marshal as Uint8Array instead of Buffer
+
+* fix: refactor remaining Buffer usage to Uint8Array
+
+
+
+<a name="0.4.1"></a>
+## [0.4.1](https://github.com/libp2p/js-interfaces/compare/v0.4.0...v0.4.1) (2020-08-11)
+
+
+
 <a name="0.4.0"></a>
 # [0.4.0](https://github.com/libp2p/js-interfaces/compare/v0.3.2...v0.4.0) (2020-08-10)
 

package.json

@@ -1,19 +1,23 @@
 {
   "name": "libp2p-interfaces",
-  "version": "0.4.0",
+  "version": "0.6.0",
   "description": "Interfaces for JS Libp2p",
   "leadMaintainer": "Jacob Heun <jacobheun@gmail.com>",
   "main": "src/index.js",
   "files": [
     "src",
+    "types",
     "dist"
   ],
   "scripts": {
     "lint": "aegir lint",
     "build": "aegir build",
+    "pregenerate:types": "rimraf './src/**/*.d.ts'",
+    "generate:types": "tsc",
     "test": "aegir test",
     "test:node": "aegir test --target node",
     "test:browser": "aegir test --target browser",
+    "prepublishOnly": "npm run generate:types",
     "release": "aegir release -t node -t browser",
     "release-minor": "aegir release --type minor -t node -t browser",
     "release-major": "aegir release --type major -t node -t browser"
@@ -35,29 +39,37 @@
   "dependencies": {
     "abort-controller": "^3.0.0",
     "abortable-iterator": "^3.0.0",
-    "buffer": "^5.6.0",
     "chai": "^4.2.0",
     "chai-checkmark": "^1.0.1",
     "class-is": "^1.1.0",
+    "debug": "^4.1.1",
     "delay": "^4.3.0",
     "detect-node": "^2.0.4",
     "dirty-chai": "^2.0.1",
     "err-code": "^2.0.0",
     "it-goodbye": "^2.0.1",
+    "it-length-prefixed": "^3.1.0",
     "it-pair": "^1.0.0",
     "it-pipe": "^1.1.0",
-    "libp2p-tcp": "^0.14.5",
+    "it-pushable": "^1.4.0",
+    "libp2p-crypto": "^0.18.0",
+    "libp2p-tcp": "^0.15.0",
     "multiaddr": "^8.0.0",
+    "multibase": "^3.0.0",
     "p-defer": "^3.0.0",
     "p-limit": "^2.3.0",
     "p-wait-for": "^3.1.0",
     "peer-id": "^0.14.0",
+    "protons": "^2.0.0",
     "sinon": "^9.0.2",
-    "streaming-iterables": "^5.0.2"
+    "streaming-iterables": "^5.0.2",
+    "uint8arrays": "^1.1.0"
   },
   "devDependencies": {
     "aegir": "^25.0.0",
-    "it-handshake": "^1.0.1"
+    "it-handshake": "^1.0.1",
+    "rimraf": "^3.0.2",
+    "typescript": "3.7.5"
   },
   "contributors": [
     "Alan Shaw <alan.shaw@protocol.ai>",

src/connection/connection.d.ts

@@ -0,0 +1,149 @@
+declare const _exports: typeof Connection;
+export = _exports;
+/**
+ * An implementation of the js-libp2p connection.
+ * Any libp2p transport should use an upgrader to return this connection.
+ */
+declare class Connection {
+    /**
+     * Creates an instance of Connection.
+     * @param {object} properties properties of the connection.
+     * @param {multiaddr} [properties.localAddr] local multiaddr of the connection if known.
+     * @param {multiaddr} [properties.remoteAddr] remote multiaddr of the connection.
+     * @param {PeerId} properties.localPeer local peer-id.
+     * @param {PeerId} properties.remotePeer remote peer-id.
+     * @param {function} properties.newStream new stream muxer function.
+     * @param {function} properties.close close raw connection function.
+     * @param {function(): Stream[]} properties.getStreams get streams from muxer function.
+     * @param {object} properties.stat metadata of the connection.
+     * @param {string} properties.stat.direction connection establishment direction ("inbound" or "outbound").
+     * @param {object} properties.stat.timeline connection relevant events timestamp.
+     * @param {string} properties.stat.timeline.open connection opening timestamp.
+     * @param {string} properties.stat.timeline.upgraded connection upgraded timestamp.
+     * @param {string} [properties.stat.multiplexer] connection multiplexing identifier.
+     * @param {string} [properties.stat.encryption] connection encryption method identifier.
+     */
+    constructor({ localAddr, remoteAddr, localPeer, remotePeer, newStream, close, getStreams, stat }: {
+        localAddr?: import("multiaddr");
+        remoteAddr?: import("multiaddr");
+        localPeer: import("peer-id");
+        remotePeer: import("peer-id");
+        newStream: Function;
+        close: Function;
+        getStreams: () => any[];
+        stat: {
+            direction: string;
+            timeline: {
+                open: string;
+                upgraded: string;
+            };
+            multiplexer?: string;
+            encryption?: string;
+        };
+    });
+    /**
+     * Connection identifier.
+     */
+    id: any;
+    /**
+     * Observed multiaddr of the local peer
+     */
+    localAddr: import("multiaddr");
+    /**
+     * Observed multiaddr of the remote peer
+     */
+    remoteAddr: import("multiaddr");
+    /**
+     * Local peer id.
+     */
+    localPeer: import("peer-id");
+    /**
+     * Remote peer id.
+     */
+    remotePeer: import("peer-id");
+    /**
+     * Connection metadata.
+     */
+    _stat: {
+        status: string;
+        direction: string;
+        timeline: {
+            open: string;
+            upgraded: string;
+        };
+        multiplexer?: string;
+        encryption?: string;
+    };
+    /**
+     * Reference to the new stream function of the multiplexer
+     */
+    _newStream: Function;
+    /**
+     * Reference to the close function of the raw connection
+     */
+    _close: Function;
+    /**
+     * Reference to the getStreams function of the muxer
+     */
+    _getStreams: () => any[];
+    /**
+     * Connection streams registry
+     */
+    registry: Map<any, any>;
+    /**
+     * User provided tags
+     * @type {string[]}
+     */
+    tags: string[];
+    /**
+     * Get connection metadata
+     * @this {Connection}
+     */
+    get stat(): {
+        status: string;
+        direction: string;
+        timeline: {
+            open: string;
+            upgraded: string;
+        };
+        multiplexer?: string;
+        encryption?: string;
+    };
+    /**
+     * Get all the streams of the muxer.
+     * @this {Connection}
+     */
+    get streams(): any[];
+    /**
+     * Create a new stream from this connection
+     * @param {string[]} protocols intended protocol for the stream
+     * @return {Promise<{stream: Stream, protocol: string}>} with muxed+multistream-selected stream and selected protocol
+     */
+    newStream(protocols: string[]): Promise<{
+        stream: any;
+        protocol: string;
+    }>;
+    /**
+     * Add a stream when it is opened to the registry.
+     * @param {*} muxedStream a muxed stream
+     * @param {object} properties the stream properties to be registered
+     * @param {string} properties.protocol the protocol used by the stream
+     * @param {object} properties.metadata metadata of the stream
+     * @return {void}
+     */
+    addStream(muxedStream: any, { protocol, metadata }: {
+        protocol: string;
+        metadata: any;
+    }): void;
+    /**
+     * Remove stream registry after it is closed.
+     * @param {string} id identifier of the stream
+     */
+    removeStream(id: string): void;
+    /**
+     * Close the connection.
+     * @return {Promise<void>}
+     */
+    close(): Promise<void>;
+    _closing: any;
+}

src/connection/connection.js

@@ -66,7 +66,7 @@ class Connection {
    * @param {PeerId} properties.remotePeer remote peer-id.
    * @param {function} properties.newStream new stream muxer function.
    * @param {function} properties.close close raw connection function.
-   * @param {function} properties.getStreams get streams from muxer function.
+   * @param {function(): Stream[]} properties.getStreams get streams from muxer function.
    * @param {object} properties.stat metadata of the connection.
    * @param {string} properties.stat.direction connection establishment direction ("inbound" or "outbound").
    * @param {object} properties.stat.timeline connection relevant events timestamp.
@@ -133,13 +133,14 @@ class Connection {
 
     /**
      * User provided tags
+     * @type {string[]}
      */
     this.tags = []
   }
 
   /**
    * Get connection metadata
-   * @return {Object}
+   * @this {Connection}
    */
   get stat () {
     return this._stat
@@ -147,7 +148,7 @@ class Connection {
 
   /**
    * Get all the streams of the muxer.
-   * @return {Array<*>}
+   * @this {Connection}
    */
   get streams () {
     return this._getStreams()
@@ -156,7 +157,7 @@ class Connection {
   /**
    * Create a new stream from this connection
    * @param {string[]} protocols intended protocol for the stream
-   * @return {Promise<object>} with muxed+multistream-selected stream and selected protocol
+   * @return {Promise<{stream: Stream, protocol: string}>} with muxed+multistream-selected stream and selected protocol
    */
   async newStream (protocols) {
     if (this.stat.status === Status.CLOSING) {
@@ -205,7 +206,7 @@ class Connection {
 
   /**
    * Close the connection.
-   * @return {Promise}
+   * @return {Promise<void>}
    */
   async close () {
     if (this.stat.status === Status.CLOSED) {
@@ -226,4 +227,8 @@ class Connection {
   }
 }
 
+/**
+ * @module
+ * @type {typeof Connection}
+ */
 module.exports = withIs(Connection, { className: 'Connection', symbolName: '@libp2p/interface-connection/connection' })

src/connection/index.d.ts

@@ -0,0 +1 @@
+export var Connection: typeof import('./connection');

src/connection/index.js

@@ -1,3 +1,7 @@
 'use strict'
 
+/**
+ * @module connection/index
+ * @type {typeof import('./connection')}
+ */
 exports.Connection = require('./connection')

src/connection/status.d.ts

@@ -0,0 +1,3 @@
+export declare const OPEN: string;
+export declare const CLOSING: string;
+export declare const CLOSED: string;

src/crypto/errors.d.ts

@@ -0,0 +1,15 @@
+export class UnexpectedPeerError extends Error {
+    static get code(): string;
+    constructor(message?: string);
+    code: string;
+}
+export class InvalidCryptoExchangeError extends Error {
+    static get code(): string;
+    constructor(message?: string);
+    code: string;
+}
+export class InvalidCryptoTransmissionError extends Error {
+    static get code(): string;
+    constructor(message?: string);
+    code: string;
+}

src/crypto/tests/index.js

@@ -1,17 +1,19 @@
 /* eslint-env mocha */
 'use strict'
 
-const { Buffer } = require('buffer')
-const duplexPair = require('it-pair/duplex')
-const pipe = require('it-pipe')
-const peers = require('../../utils/peers')
-const { UnexpectedPeerError } = require('../errors')
-const PeerId = require('peer-id')
-const { collect } = require('streaming-iterables')
 const chai = require('chai')
 const expect = chai.expect
 chai.use(require('dirty-chai'))
 
+const duplexPair = require('it-pair/duplex')
+const pipe = require('it-pipe')
+const PeerId = require('peer-id')
+const { collect } = require('streaming-iterables')
+const uint8arrayFromString = require('uint8arrays/from-string')
+
+const peers = require('../../utils/peers')
+const { UnexpectedPeerError } = require('../errors')
+
 module.exports = (common) => {
   describe('interface-crypto', () => {
     let crypto
@@ -55,7 +57,7 @@ module.exports = (common) => {
       pipe(inboundResult.conn, inboundResult.conn)
 
       // Send some data and collect the result
-      const input = Buffer.from('data to encrypt')
+      const input = uint8arrayFromString('data to encrypt')
       const result = await pipe(
         [input],
         outboundResult.conn,

src/peer-discovery/README.md

@@ -33,19 +33,19 @@ Include this badge in your readme if you make a new module that uses interface-p
 Install `interface-discovery` as one of the dependencies of your project and as a test file. Then, using `mocha` (for JavaScript) or a test runner with compatible API, do:
 
 ```js
-const test = require('interface-discovery')
+const tests = require('libp2p-interfaces/src/peer-discovery/tests')
 
-const common = {
-  setup () {
-    return YourDiscovery
-  },
-  teardown () {
-    // Clean up any resources created by setup()
-  }
-}
-
-// use all of the test suits
-test(common)
+describe('your discovery', () => {
+  // use all of the test suits
+  tests({
+    setup () {
+      return YourDiscovery
+    },
+    teardown () {
+      // Clean up any resources created by setup()
+    }
+  })
+})
 ```
 
 ## API

src/pubsub/README.md

@@ -0,0 +1,236 @@
+interface-pubsub
+==================
+
+The `interface-pubsub` contains the base implementation for a libp2p pubsub router implementation. This interface should be used to implement a pubsub router compatible with libp2p. It includes a test suite that pubsub routers should run, in order to ensure compatibility with libp2p.
+
+Table of Contents
+=================
+
+* [Implementations using this interface](#implementations-using-this-interface)
+* [Interface usage](#interface-usage)
+    * [Extend interface](#extend-interface)
+    * [Example](#example)
+* [API](#api)
+    * [Start](#start)
+      * [pubsub.start()](#pubsubstart)
+          * [Returns](#returns)
+    * [Stop](#stop)
+      * [pubsub.stop()](#pubsubstop)
+          * [Returns](#returns-1)
+    * [Publish](#publish)
+      * [pubsub.publish(topics, message)](#pubsubpublishtopics-message)
+          * [Parameters](#parameters)
+          * [Returns](#returns-2)
+    * [Subscribe](#subscribe)
+      * [pubsub.subscribe(topic)](#pubsubsubscribetopic)
+          * [Parameters](#parameters-1)
+    * [Unsubscribe](#unsubscribe)
+      * [pubsub.unsubscribe(topic)](#pubsubunsubscribetopic)
+          * [Parameters](#parameters-2)
+    * [Get Topics](#get-topics)
+      * [pubsub.getTopics()](#pubsubgettopics)
+          * [Returns](#returns-3)
+    * [Get Peers Subscribed to a topic](#get-peers-subscribed-to-a-topic)
+      * [pubsub.getSubscribers(topic)](#pubsubgetsubscriberstopic)
+          * [Parameters](#parameters-3)
+          * [Returns](#returns-4)
+    * [Validate](#validate)
+      * [pubsub.validate(message)](#pubsubvalidatemessage)
+          * [Parameters](#parameters-4)
+      * [Returns](#returns-5)
+* [Test suite usage](#test-suite-usage)
+
+## Implementations using this interface
+
+You can check the following implementations as examples for building your own pubsub router.
+
+- [libp2p/js-libp2p-floodsub](https://github.com/libp2p/js-libp2p-floodsub)
+- [ChainSafe/js-libp2p-gossipsub](https://github.com/ChainSafe/js-libp2p-gossipsub)
+
+## Interface usage
+
+`interface-pubsub` abstracts the implementation protocol registration within `libp2p` and takes care of all the protocol connections and streams, as well as the subscription management. This way, a pubsub implementation can focus on its message routing algorithm, instead of also needing to create the setup for it.
+
+### Extend interface
+
+A pubsub router implementation should start by extending the `interface-pubsub` class and **MUST** override the `_publish` function, according to the router algorithms. This function is responsible for forwarding publish messages to other peers, as well as forwarding received messages if the router provides the `canRelayMessage` option to the base implementation.
+
+Other functions, such as `start`, `stop`, `subscribe`, `unsubscribe`, `_encodeRpc`, `_decodeRpc`, `_processRpcMessage`, `_addPeer` and `_removePeer` may be overwritten if the pubsub implementation needs to customize their logic. Implementations overriding these functions **MUST** call `super`.
+
+The `start` and `stop` functions are responsible for the registration of the pubsub protocol with libp2p. The `stop` function also guarantees that the open streams in the protocol are properly closed.
+
+The `subscribe` and `unsubscribe` functions take care of the subscription management and its inherent message propagation.
+
+When using a custom protobuf definition for message marshalling, you should override `_encodeRpc` and `_decodeRpc` to use the new protobuf instead of the default one.
+
+`_processRpcMessage` is responsible for handling messages received from other peers. This should be extended if further operations/validations are needed by the router.
+
+The `_addPeer` and `_removePeer` functions are called when new peers running the pubsub router protocol establish a connection with the peer. They are used for tracking the open streams between the peers.
+
+All the remaining functions **MUST NOT** be overwritten.
+
+### Example
+
+The following example aims to show how to create your pubsub implementation extending this base protocol. The pubsub implementation will handle the subscriptions logic.
+
+```JavaScript
+const Pubsub = require('libp2p-pubsub')
+
+class PubsubImplementation extends Pubsub {
+  constructor({ libp2p, options })
+    super({
+      debugName: 'libp2p:pubsub',
+      multicodecs: '/pubsub-implementation/1.0.0',
+      libp2p,
+      signMessages: options.signMessages,
+      strictSigning: options.strictSigning
+    })
+  }
+
+  _publish (message) {
+    // Required to be implemented by the subclass
+    // Routing logic for the message
+  }
+}
+```
+
+## API
+
+The interface aims to specify a common interface that all pubsub router implementation should follow. A pubsub router implementation should extend the [EventEmitter](https://nodejs.org/api/events.html#events_class_eventemitter). When peers receive pubsub messages, these messages will be emitted by the event emitter where the `eventName` will be the `topic` associated with the message.
+
+### Start
+
+Starts the pubsub subsystem. The protocol will be registered to `libp2p`, which will result in pubsub being notified when peers who support the protocol connect/disconnect to `libp2p`.
+
+#### `pubsub.start()`
+
+### Stop
+
+Stops the pubsub subsystem. The protocol will be unregistered from `libp2p`, which will remove all listeners for the protocol and the established connections will be closed.
+
+#### `pubsub.stop()`
+
+### Publish
+
+Publish data message to pubsub topics.
+
+#### `pubsub.publish(topic, message)`
+
+##### Parameters
+
+| Name | Type | Description |
+|------|------|-------------|
+| topic | `string` | pubsub topic |
+| message | `Uint8Array` | message to publish |
+
+##### Returns
+
+| Type | Description |
+|------|-------------|
+| `Promise<void>` | resolves once the message is published to the network |
+
+### Subscribe
+
+Subscribe to the given topic.
+
+#### `pubsub.subscribe(topic)`
+
+##### Parameters
+
+| Name | Type | Description |
+|------|------|-------------|
+| topic | `string` | pubsub topic |
+
+### Unsubscribe
+
+Unsubscribe from the given topic.
+
+#### `pubsub.unsubscribe(topic)`
+
+##### Parameters
+
+| Name | Type | Description |
+|------|------|-------------|
+| topic | `string` | pubsub topic |
+
+### Get Topics
+
+Get the list of topics which the peer is subscribed to.
+
+#### `pubsub.getTopics()`
+
+##### Returns
+
+| Type | Description |
+|------|-------------|
+| `Array<String>` | Array of subscribed topics |
+
+### Get Peers Subscribed to a topic
+
+Get a list of the [PeerId](https://github.com/libp2p/js-peer-id) strings that are subscribed to one topic.
+
+#### `pubsub.getSubscribers(topic)`
+
+##### Parameters
+
+| Name | Type | Description |
+|------|------|-------------|
+| topic | `string` | pubsub topic |
+
+##### Returns
+
+| Type | Description |
+|------|-------------|
+| `Array<string>` | Array of base-58 PeerId's |
+
+### Validate
+
+Validates the signature of a message.
+
+#### `pubsub.validate(message)`
+
+##### Parameters
+
+| Name | Type | Description |
+|------|------|-------------|
+| message | `Message` | a pubsub message |
+
+#### Returns
+
+| Type | Description |
+|------|-------------|
+| `Promise<void>` | resolves if the message is valid |
+
+## Test suite usage
+
+```js
+'use strict'
+
+const tests = require('libp2p-interfaces/src/pubsub/tests')
+const YourPubsubRouter = require('../src')
+
+describe('compliance', () => {
+  let peers
+  let pubsubNodes = []
+
+  tests({
+    async setup (number = 1, options = {}) {
+      // Create number pubsub nodes with libp2p
+      peers = await createPeers({ number })
+
+      peers.forEach((peer) => {
+        const ps = new YourPubsubRouter(peer, options)
+
+        pubsubNodes.push(ps)
+      })
+
+      return pubsubNodes
+    },
+    async teardown () {
+      // Clean up any resources created by setup()
+      await Promise.all(pubsubNodes.map(ps => ps.stop()))
+      peers.length && await Promise.all(peers.map(peer => peer.stop()))
+    }
+  })
+})
+```

src/pubsub/errors.d.ts

@@ -0,0 +1,4 @@
+export namespace codes {
+    export const ERR_MISSING_SIGNATURE: string;
+    export const ERR_INVALID_SIGNATURE: string;
+}

src/pubsub/errors.js

@@ -0,0 +1,6 @@
+'use strict'
+
+exports.codes = {
+  ERR_MISSING_SIGNATURE: 'ERR_MISSING_SIGNATURE',
+  ERR_INVALID_SIGNATURE: 'ERR_INVALID_SIGNATURE'
+}

src/pubsub/index.d.ts

@@ -0,0 +1,307 @@
+export = PubsubBaseProtocol;
+/**
+ * @typedef {Object} InMessage
+ * @property {string} [from]
+ * @property {string} receivedFrom
+ * @property {string[]} topicIDs
+ * @property {Uint8Array} [seqno]
+ * @property {Uint8Array} data
+ * @property {Uint8Array} [signature]
+ * @property {Uint8Array} [key]
+ *
+ * @typedef PeerId
+ * @type import('peer-id')
+ */
+/**
+* PubsubBaseProtocol handles the peers and connections logic for pubsub routers
+* and specifies the API that pubsub routers should have.
+*/
+declare class PubsubBaseProtocol {
+    /**
+     * @param {Object} props
+     * @param {String} props.debugName log namespace
+     * @param {Array<string>|string} props.multicodecs protocol identificers to connect
+     * @param {Libp2p} props.libp2p
+     * @param {boolean} [props.signMessages = true] if messages should be signed
+     * @param {boolean} [props.strictSigning = true] if message signing should be required
+     * @param {boolean} [props.canRelayMessage = false] if can relay messages not subscribed
+     * @param {boolean} [props.emitSelf = false] if publish should emit to self, if subscribed
+     * @abstract
+     */
+    constructor({ debugName, multicodecs, libp2p, signMessages, strictSigning, canRelayMessage, emitSelf }: {
+        debugName: string;
+        multicodecs: string | string[];
+        libp2p: any;
+        signMessages?: boolean;
+        strictSigning?: boolean;
+        canRelayMessage?: boolean;
+        emitSelf?: boolean;
+    });
+    log: any;
+    /**
+     * @type {Array<string>}
+     */
+    multicodecs: Array<string>;
+    _libp2p: any;
+    registrar: any;
+    /**
+     * @type {PeerId}
+     */
+    peerId: PeerId;
+    started: boolean;
+    /**
+     * Map of topics to which peers are subscribed to
+     *
+     * @type {Map<string, Set<string>>}
+     */
+    topics: Map<string, Set<string>>;
+    /**
+     * List of our subscriptions
+     * @type {Set<string>}
+     */
+    subscriptions: Set<string>;
+    /**
+     * Map of peer streams
+     *
+     * @type {Map<string, import('./peer-streams')>}
+     */
+    peers: Map<string, import('./peer-streams')>;
+    signMessages: boolean;
+    /**
+     * If message signing should be required for incoming messages
+     * @type {boolean}
+     */
+    strictSigning: boolean;
+    /**
+     * If router can relay received messages, even if not subscribed
+     * @type {boolean}
+     */
+    canRelayMessage: boolean;
+    /**
+     * if publish should emit to self, if subscribed
+     * @type {boolean}
+     */
+    emitSelf: boolean;
+    /**
+     * Topic validator function
+     * @typedef {function(string, InMessage): Promise<void>} validator
+     */
+    /**
+     * Topic validator map
+     *
+     * Keyed by topic
+     * Topic validators are functions with the following input:
+     * @type {Map<string, validator>}
+     */
+    topicValidators: Map<string, validator>;
+    _registrarId: any;
+    /**
+     * On an inbound stream opened.
+     * @private
+     * @param {Object} props
+     * @param {string} props.protocol
+     * @param {DuplexIterableStream} props.stream
+     * @param {Connection} props.connection connection
+     */
+    _onIncomingStream({ protocol, stream, connection }: {
+        protocol: string;
+        stream: any;
+        connection: any;
+    }): void;
+    /**
+     * Registrar notifies an established connection with pubsub protocol.
+     * @private
+     * @param {PeerId} peerId remote peer-id
+     * @param {Connection} conn connection to the peer
+     */
+    _onPeerConnected(peerId: import("peer-id"), conn: any): Promise<void>;
+    /**
+     * Registrar notifies a closing connection with pubsub protocol.
+     * @private
+     * @param {PeerId} peerId peerId
+     * @param {Error} err error for connection end
+     */
+    _onPeerDisconnected(peerId: import("peer-id"), err: Error): void;
+    /**
+     * Register the pubsub protocol onto the libp2p node.
+     * @returns {void}
+     */
+    start(): void;
+    /**
+     * Unregister the pubsub protocol and the streams with other peers will be closed.
+     * @returns {void}
+     */
+    stop(): void;
+    /**
+     * Notifies the router that a peer has been connected
+     * @private
+     * @param {PeerId} peerId
+     * @param {string} protocol
+     * @returns {PeerStreams}
+     */
+    _addPeer(peerId: import("peer-id"), protocol: string): import("./peer-streams");
+    /**
+     * Notifies the router that a peer has been disconnected.
+     * @private
+     * @param {PeerId} peerId
+     * @returns {PeerStreams | undefined}
+     */
+    _removePeer(peerId: import("peer-id")): import("./peer-streams");
+    /**
+     * Responsible for processing each RPC message received by other peers.
+     * @param {string} idB58Str peer id string in base58
+     * @param {DuplexIterableStream} stream inbound stream
+     * @param {PeerStreams} peerStreams PubSub peer
+     * @returns {Promise<void>}
+     */
+    _processMessages(idB58Str: string, stream: any, peerStreams: import("./peer-streams")): Promise<void>;
+    /**
+     * Handles an rpc request from a peer
+     * @param {String} idB58Str
+     * @param {PeerStreams} peerStreams
+     * @param {RPC} rpc
+     * @returns {boolean}
+     */
+    _processRpc(idB58Str: string, peerStreams: import("./peer-streams"), rpc: any): boolean;
+    /**
+     * Handles a subscription change from a peer
+     * @param {string} id
+     * @param {RPC.SubOpt} subOpt
+     */
+    _processRpcSubOpt(id: string, subOpt: any): void;
+    /**
+     * Handles an message from a peer
+     * @param {InMessage} msg
+     * @returns {Promise<void>}
+     */
+    _processRpcMessage(msg: InMessage): Promise<void>;
+    /**
+     * Emit a message from a peer
+     * @param {InMessage} message
+     */
+    _emitMessage(message: InMessage): void;
+    /**
+     * The default msgID implementation
+     * Child class can override this.
+     * @param {RPC.Message} msg the message object
+     * @returns {Uint8Array} message id as bytes
+     */
+    getMsgId(msg: any): Uint8Array;
+    /**
+     * Whether to accept a message from a peer
+     * Override to create a graylist
+     * @override
+     * @param {string} id
+     * @returns {boolean}
+     */
+    _acceptFrom(id: string): boolean;
+    /**
+     * Decode Uint8Array into an RPC object.
+     * This can be override to use a custom router protobuf.
+     * @param {Uint8Array} bytes
+     * @returns {RPC}
+     */
+    _decodeRpc(bytes: Uint8Array): any;
+    /**
+     * Encode RPC object into a Uint8Array.
+     * This can be override to use a custom router protobuf.
+     * @param {RPC} rpc
+     * @returns {Uint8Array}
+     */
+    _encodeRpc(rpc: any): Uint8Array;
+    /**
+     * Send an rpc object to a peer
+     * @param {string} id peer id
+     * @param {RPC} rpc
+     * @returns {void}
+     */
+    _sendRpc(id: string, rpc: any): void;
+    /**
+     * Send subscroptions to a peer
+     * @param {string} id peer id
+     * @param {string[]} topics
+     * @param {boolean} subscribe set to false for unsubscriptions
+     * @returns {void}
+     */
+    _sendSubscriptions(id: string, topics: string[], subscribe: boolean): void;
+    /**
+     * Validates the given message. The signature will be checked for authenticity.
+     * Throws an error on invalid messages
+     * @param {InMessage} message
+     * @returns {Promise<void>}
+     */
+    validate(message: InMessage): Promise<void>;
+    /**
+     * Normalizes the message and signs it, if signing is enabled.
+     * Should be used by the routers to create the message to send.
+     * @private
+     * @param {Message} message
+     * @returns {Promise<Message>}
+     */
+    _buildMessage(message: any): Promise<any>;
+    /**
+     * Get a list of the peer-ids that are subscribed to one topic.
+     * @param {string} topic
+     * @returns {Array<string>}
+     */
+    getSubscribers(topic: string): string[];
+    /**
+     * Publishes messages to all subscribed peers
+     * @override
+     * @param {string} topic
+     * @param {Buffer} message
+     * @returns {Promise<void>}
+     */
+    publish(topic: string, message: Buffer): Promise<void>;
+    /**
+     * Overriding the implementation of publish should handle the appropriate algorithms for the publish/subscriber implementation.
+     * For example, a Floodsub implementation might simply publish each message to each topic for every peer
+     * @abstract
+     * @param {InMessage} message
+     * @returns {Promise<void>}
+     *
+     */
+    _publish(message: InMessage): Promise<void>;
+    /**
+     * Subscribes to a given topic.
+     * @abstract
+     * @param {string} topic
+     * @returns {void}
+     */
+    subscribe(topic: string): void;
+    /**
+     * Unsubscribe from the given topic.
+     * @override
+     * @param {string} topic
+     * @returns {void}
+     */
+    unsubscribe(topic: string): void;
+    /**
+     * Get the list of topics which the peer is subscribed to.
+     * @override
+     * @returns {Array<String>}
+     */
+    getTopics(): string[];
+}
+declare namespace PubsubBaseProtocol {
+    export { message, utils, InMessage, PeerId };
+}
+type PeerId = import("peer-id");
+/**
+ * Topic validator function
+ */
+type validator = (arg0: string, arg1: InMessage) => Promise<void>;
+type InMessage = {
+    from?: string;
+    receivedFrom: string;
+    topicIDs: string[];
+    seqno?: Uint8Array;
+    data: Uint8Array;
+    signature?: Uint8Array;
+    key?: Uint8Array;
+};
+/**
+ * @type {typeof import('./message')}
+ */
+declare const message: typeof import('./message');
+declare const utils: typeof import("./utils");

src/pubsub/index.js

@@ -0,0 +1,668 @@
+'use strict'
+
+const debug = require('debug')
+const EventEmitter = require('events')
+const errcode = require('err-code')
+
+const pipe = require('it-pipe')
+
+const MulticodecTopology = require('../topology/multicodec-topology')
+const { codes } = require('./errors')
+/**
+ * @type {typeof import('./message')}
+ */
+const message = require('./message')
+const PeerStreams = require('./peer-streams')
+const utils = require('./utils')
+
+const {
+  signMessage,
+  verifySignature
+} = require('./message/sign')
+
+/**
+ * @typedef {Object} InMessage
+ * @property {string} [from]
+ * @property {string} receivedFrom
+ * @property {string[]} topicIDs
+ * @property {Uint8Array} [seqno]
+ * @property {Uint8Array} data
+ * @property {Uint8Array} [signature]
+ * @property {Uint8Array} [key]
+ *
+ * @typedef PeerId
+ * @type import('peer-id')
+ */
+
+/**
+* PubsubBaseProtocol handles the peers and connections logic for pubsub routers
+* and specifies the API that pubsub routers should have.
+*/
+class PubsubBaseProtocol extends EventEmitter {
+  /**
+   * @param {Object} props
+   * @param {String} props.debugName log namespace
+   * @param {Array<string>|string} props.multicodecs protocol identificers to connect
+   * @param {Libp2p} props.libp2p
+   * @param {boolean} [props.signMessages = true] if messages should be signed
+   * @param {boolean} [props.strictSigning = true] if message signing should be required
+   * @param {boolean} [props.canRelayMessage = false] if can relay messages not subscribed
+   * @param {boolean} [props.emitSelf = false] if publish should emit to self, if subscribed
+   * @abstract
+   */
+  constructor ({
+    debugName,
+    multicodecs,
+    libp2p,
+    signMessages = true,
+    strictSigning = true,
+    canRelayMessage = false,
+    emitSelf = false
+  }) {
+    if (typeof debugName !== 'string') {
+      throw new Error('a debugname `string` is required')
+    }
+
+    if (!multicodecs) {
+      throw new Error('multicodecs are required')
+    }
+
+    if (!libp2p) {
+      throw new Error('libp2p is required')
+    }
+
+    super()
+
+    this.log = debug(debugName)
+    this.log.err = debug(`${debugName}:error`)
+
+    /**
+     * @type {Array<string>}
+     */
+    this.multicodecs = utils.ensureArray(multicodecs)
+    this._libp2p = libp2p
+    this.registrar = libp2p.registrar
+    /**
+     * @type {PeerId}
+     */
+    this.peerId = libp2p.peerId
+
+    this.started = false
+
+    /**
+     * Map of topics to which peers are subscribed to
+     *
+     * @type {Map<string, Set<string>>}
+     */
+    this.topics = new Map()
+
+    /**
+     * List of our subscriptions
+     * @type {Set<string>}
+     */
+    this.subscriptions = new Set()
+
+    /**
+     * Map of peer streams
+     *
+     * @type {Map<string, import('./peer-streams')>}
+     */
+    this.peers = new Map()
+
+    // Message signing
+    this.signMessages = signMessages
+
+    /**
+     * If message signing should be required for incoming messages
+     * @type {boolean}
+     */
+    this.strictSigning = strictSigning
+
+    /**
+     * If router can relay received messages, even if not subscribed
+     * @type {boolean}
+     */
+    this.canRelayMessage = canRelayMessage
+
+    /**
+     * if publish should emit to self, if subscribed
+     * @type {boolean}
+     */
+    this.emitSelf = emitSelf
+
+    /**
+     * Topic validator function
+     * @typedef {function(string, InMessage): Promise<void>} validator
+     */
+    /**
+     * Topic validator map
+     *
+     * Keyed by topic
+     * Topic validators are functions with the following input:
+     * @type {Map<string, validator>}
+     */
+    this.topicValidators = new Map()
+
+    this._registrarId = undefined
+    this._onIncomingStream = this._onIncomingStream.bind(this)
+    this._onPeerConnected = this._onPeerConnected.bind(this)
+    this._onPeerDisconnected = this._onPeerDisconnected.bind(this)
+  }
+
+  // LIFECYCLE METHODS
+
+  /**
+   * Register the pubsub protocol onto the libp2p node.
+   * @returns {void}
+   */
+  start () {
+    if (this.started) {
+      return
+    }
+    this.log('starting')
+
+    // Incoming streams
+    // Called after a peer dials us
+    this.registrar.handle(this.multicodecs, this._onIncomingStream)
+
+    // register protocol with topology
+    // Topology callbacks called on connection manager changes
+    const topology = new MulticodecTopology({
+      multicodecs: this.multicodecs,
+      handlers: {
+        onConnect: this._onPeerConnected,
+        onDisconnect: this._onPeerDisconnected
+      }
+    })
+    this._registrarId = this.registrar.register(topology)
+
+    this.log('started')
+    this.started = true
+  }
+
+  /**
+   * Unregister the pubsub protocol and the streams with other peers will be closed.
+   * @returns {void}
+   */
+  stop () {
+    if (!this.started) {
+      return
+    }
+
+    // unregister protocol and handlers
+    this.registrar.unregister(this._registrarId)
+
+    this.log('stopping')
+    this.peers.forEach((peerStreams) => peerStreams.close())
+
+    this.peers = new Map()
+    this.subscriptions = new Set()
+    this.started = false
+    this.log('stopped')
+  }
+
+  /**
+   * On an inbound stream opened.
+   * @private
+   * @param {Object} props
+   * @param {string} props.protocol
+   * @param {DuplexIterableStream} props.stream
+   * @param {Connection} props.connection connection
+   */
+  _onIncomingStream ({ protocol, stream, connection }) {
+    const peerId = connection.remotePeer
+    const idB58Str = peerId.toB58String()
+    const peer = this._addPeer(peerId, protocol)
+    peer.attachInboundStream(stream)
+
+    this._processMessages(idB58Str, peer.inboundStream, peer)
+  }
+
+  /**
+   * Registrar notifies an established connection with pubsub protocol.
+   * @private
+   * @param {PeerId} peerId remote peer-id
+   * @param {Connection} conn connection to the peer
+   */
+  async _onPeerConnected (peerId, conn) {
+    const idB58Str = peerId.toB58String()
+    this.log('connected', idB58Str)
+
+    try {
+      const { stream, protocol } = await conn.newStream(this.multicodecs)
+      const peer = this._addPeer(peerId, protocol)
+      await peer.attachOutboundStream(stream)
+    } catch (err) {
+      this.log.err(err)
+    }
+
+    // Immediately send my own subscriptions to the newly established conn
+    this._sendSubscriptions(idB58Str, Array.from(this.subscriptions), true)
+  }
+
+  /**
+   * Registrar notifies a closing connection with pubsub protocol.
+   * @private
+   * @param {PeerId} peerId peerId
+   * @param {Error} err error for connection end
+   */
+  _onPeerDisconnected (peerId, err) {
+    const idB58Str = peerId.toB58String()
+
+    this.log('connection ended', idB58Str, err ? err.message : '')
+    this._removePeer(peerId)
+  }
+
+  /**
+   * Notifies the router that a peer has been connected
+   * @private
+   * @param {PeerId} peerId
+   * @param {string} protocol
+   * @returns {PeerStreams}
+   */
+  _addPeer (peerId, protocol) {
+    const id = peerId.toB58String()
+    const existing = this.peers.get(id)
+
+    // If peer streams already exists, do nothing
+    if (existing) {
+      return existing
+    }
+
+    // else create a new peer streams
+    this.log('new peer', id)
+
+    const peerStreams = new PeerStreams({
+      id: peerId,
+      protocol
+    })
+
+    this.peers.set(id, peerStreams)
+    peerStreams.once('close', () => this._removePeer(peerId))
+
+    return peerStreams
+  }
+
+  /**
+   * Notifies the router that a peer has been disconnected.
+   * @private
+   * @param {PeerId} peerId
+   * @returns {PeerStreams | undefined}
+   */
+  _removePeer (peerId) {
+    if (!peerId) return
+    const id = peerId.toB58String()
+    const peerStreams = this.peers.get(id)
+    if (!peerStreams) return
+
+    // close peer streams
+    peerStreams.removeAllListeners()
+    peerStreams.close()
+
+    // delete peer streams
+    this.log('delete peer', id)
+    this.peers.delete(id)
+
+    // remove peer from topics map
+    for (const peers of this.topics.values()) {
+      peers.delete(id)
+    }
+
+    return peerStreams
+  }
+
+  // MESSAGE METHODS
+
+  /**
+   * Responsible for processing each RPC message received by other peers.
+   * @param {string} idB58Str peer id string in base58
+   * @param {DuplexIterableStream} stream inbound stream
+   * @param {PeerStreams} peerStreams PubSub peer
+   * @returns {Promise<void>}
+   */
+  async _processMessages (idB58Str, stream, peerStreams) {
+    try {
+      await pipe(
+        stream,
+        async (source) => {
+          for await (const data of source) {
+            const rpcBytes = data instanceof Uint8Array ? data : data.slice()
+            const rpcMsg = this._decodeRpc(rpcBytes)
+
+            this._processRpc(idB58Str, peerStreams, rpcMsg)
+          }
+        }
+      )
+    } catch (err) {
+      this.log.err(err)
+    }
+  }
+
+  /**
+   * Handles an rpc request from a peer
+   * @param {String} idB58Str
+   * @param {PeerStreams} peerStreams
+   * @param {RPC} rpc
+   * @returns {boolean}
+   */
+  _processRpc (idB58Str, peerStreams, rpc) {
+    this.log('rpc from', idB58Str)
+    const subs = rpc.subscriptions
+    const msgs = rpc.msgs
+
+    if (subs.length) {
+      // update peer subscriptions
+      subs.forEach((subOpt) => this._processRpcSubOpt(idB58Str, subOpt))
+      this.emit('pubsub:subscription-change', peerStreams.id, subs)
+    }
+
+    if (!this._acceptFrom(idB58Str)) {
+      this.log('received message from unacceptable peer %s', idB58Str)
+      return false
+    }
+
+    if (msgs.length) {
+      msgs.forEach(message => {
+        if (!(this.canRelayMessage || message.topicIDs.some((topic) => this.subscriptions.has(topic)))) {
+          this.log('received message we didn\'t subscribe to. Dropping.')
+          return
+        }
+        const msg = utils.normalizeInRpcMessage(message, idB58Str)
+        this._processRpcMessage(msg)
+      })
+    }
+    return true
+  }
+
+  /**
+   * Handles a subscription change from a peer
+   * @param {string} id
+   * @param {RPC.SubOpt} subOpt
+   */
+  _processRpcSubOpt (id, subOpt) {
+    const t = subOpt.topicID
+
+    let topicSet = this.topics.get(t)
+    if (!topicSet) {
+      topicSet = new Set()
+      this.topics.set(t, topicSet)
+    }
+
+    if (subOpt.subscribe) {
+      // subscribe peer to new topic
+      topicSet.add(id)
+    } else {
+      // unsubscribe from existing topic
+      topicSet.delete(id)
+    }
+  }
+
+  /**
+   * Handles an message from a peer
+   * @param {InMessage} msg
+   * @returns {Promise<void>}
+   */
+  async _processRpcMessage (msg) {
+    if (this.peerId.toB58String() === msg.from && !this.emitSelf) {
+      return
+    }
+
+    // Ensure the message is valid before processing it
+    try {
+      await this.validate(msg)
+    } catch (err) {
+      this.log('Message is invalid, dropping it. %O', err)
+      return
+    }
+
+    // Emit to self
+    this._emitMessage(msg)
+
+    this._publish(utils.normalizeOutRpcMessage(msg))
+  }
+
+  /**
+   * Emit a message from a peer
+   * @param {InMessage} message
+   */
+  _emitMessage (message) {
+    message.topicIDs.forEach((topic) => {
+      if (this.subscriptions.has(topic)) {
+        this.emit(topic, message)
+      }
+    })
+  }
+
+  /**
+   * The default msgID implementation
+   * Child class can override this.
+   * @param {RPC.Message} msg the message object
+   * @returns {Uint8Array} message id as bytes
+   */
+  getMsgId (msg) {
+    return utils.msgId(msg.from, msg.seqno)
+  }
+
+  /**
+   * Whether to accept a message from a peer
+   * Override to create a graylist
+   * @override
+   * @param {string} id
+   * @returns {boolean}
+   */
+  _acceptFrom (id) {
+    return true
+  }
+
+  /**
+   * Decode Uint8Array into an RPC object.
+   * This can be override to use a custom router protobuf.
+   * @param {Uint8Array} bytes
+   * @returns {RPC}
+   */
+  _decodeRpc (bytes) {
+    return message.rpc.RPC.decode(bytes)
+  }
+
+  /**
+   * Encode RPC object into a Uint8Array.
+   * This can be override to use a custom router protobuf.
+   * @param {RPC} rpc
+   * @returns {Uint8Array}
+   */
+  _encodeRpc (rpc) {
+    return message.rpc.RPC.encode(rpc)
+  }
+
+  /**
+   * Send an rpc object to a peer
+   * @param {string} id peer id
+   * @param {RPC} rpc
+   * @returns {void}
+   */
+  _sendRpc (id, rpc) {
+    const peerStreams = this.peers.get(id)
+    if (!peerStreams || !peerStreams.isWritable) {
+      const msg = `Cannot send RPC to ${id} as there is no open stream to it available`
+
+      this.log.err(msg)
+      return
+    }
+    peerStreams.write(this._encodeRpc(rpc))
+  }
+
+  /**
+   * Send subscroptions to a peer
+   * @param {string} id peer id
+   * @param {string[]} topics
+   * @param {boolean} subscribe set to false for unsubscriptions
+   * @returns {void}
+   */
+  _sendSubscriptions (id, topics, subscribe) {
+    return this._sendRpc(id, {
+      subscriptions: topics.map(t => ({ topicID: t, subscribe: subscribe }))
+    })
+  }
+
+  /**
+   * Validates the given message. The signature will be checked for authenticity.
+   * Throws an error on invalid messages
+   * @param {InMessage} message
+   * @returns {Promise<void>}
+   */
+  async validate (message) { // eslint-disable-line require-await
+    // If strict signing is on and we have no signature, abort
+    if (this.strictSigning && !message.signature) {
+      throw errcode(new Error('Signing required and no signature was present'), codes.ERR_MISSING_SIGNATURE)
+    }
+
+    // Check the message signature if present
+    if (message.signature && !(await verifySignature(message))) {
+      throw errcode(new Error('Invalid message signature'), codes.ERR_INVALID_SIGNATURE)
+    }
+
+    for (const topic of message.topicIDs) {
+      const validatorFn = this.topicValidators.get(topic)
+      if (!validatorFn) {
+        continue
+      }
+      await validatorFn(topic, message)
+    }
+  }
+
+  /**
+   * Normalizes the message and signs it, if signing is enabled.
+   * Should be used by the routers to create the message to send.
+   * @private
+   * @param {Message} message
+   * @returns {Promise<Message>}
+   */
+  _buildMessage (message) {
+    const msg = utils.normalizeOutRpcMessage(message)
+    if (this.signMessages) {
+      return signMessage(this.peerId, msg)
+    } else {
+      return message
+    }
+  }
+
+  // API METHODS
+
+  /**
+   * Get a list of the peer-ids that are subscribed to one topic.
+   * @param {string} topic
+   * @returns {Array<string>}
+   */
+  getSubscribers (topic) {
+    if (!this.started) {
+      throw errcode(new Error('not started yet'), 'ERR_NOT_STARTED_YET')
+    }
+
+    if (!topic || typeof topic !== 'string') {
+      throw errcode(new Error('a string topic must be provided'), 'ERR_NOT_VALID_TOPIC')
+    }
+
+    const peersInTopic = this.topics.get(topic)
+    if (!peersInTopic) {
+      return []
+    }
+    return Array.from(peersInTopic)
+  }
+
+  /**
+   * Publishes messages to all subscribed peers
+   * @override
+   * @param {string} topic
+   * @param {Buffer} message
+   * @returns {Promise<void>}
+   */
+  async publish (topic, message) {
+    if (!this.started) {
+      throw new Error('Pubsub has not started')
+    }
+
+    this.log('publish', topic, message)
+
+    const from = this.peerId.toB58String()
+    let msgObject = {
+      receivedFrom: from,
+      from: from,
+      data: message,
+      seqno: utils.randomSeqno(),
+      topicIDs: [topic]
+    }
+
+    // ensure that any operations performed on the message will include the signature
+    const outMsg = await this._buildMessage(msgObject)
+    msgObject = utils.normalizeInRpcMessage(outMsg)
+
+    // Emit to self if I'm interested and emitSelf enabled
+    this.emitSelf && this._emitMessage(msgObject)
+
+    // send to all the other peers
+    await this._publish(msgObject)
+  }
+
+  /**
+   * Overriding the implementation of publish should handle the appropriate algorithms for the publish/subscriber implementation.
+   * For example, a Floodsub implementation might simply publish each message to each topic for every peer
+   * @abstract
+   * @param {InMessage} message
+   * @returns {Promise<void>}
+   *
+   */
+  _publish (message) {
+    throw errcode(new Error('publish must be implemented by the subclass'), 'ERR_NOT_IMPLEMENTED')
+  }
+
+  /**
+   * Subscribes to a given topic.
+   * @abstract
+   * @param {string} topic
+   * @returns {void}
+   */
+  subscribe (topic) {
+    if (!this.started) {
+      throw new Error('Pubsub has not started')
+    }
+
+    if (!this.subscriptions.has(topic)) {
+      this.subscriptions.add(topic)
+      this.peers.forEach((_, id) => this._sendSubscriptions(id, [topic], true))
+    }
+  }
+
+  /**
+   * Unsubscribe from the given topic.
+   * @override
+   * @param {string} topic
+   * @returns {void}
+   */
+  unsubscribe (topic) {
+    if (!this.started) {
+      throw new Error('Pubsub is not started')
+    }
+
+    if (this.subscriptions.has(topic) && this.listenerCount(topic) === 0) {
+      this.subscriptions.delete(topic)
+      this.peers.forEach((_, id) => this._sendSubscriptions(id, [topic], false))
+    }
+  }
+
+  /**
+   * Get the list of topics which the peer is subscribed to.
+   * @override
+   * @returns {Array<String>}
+   */
+  getTopics () {
+    if (!this.started) {
+      throw new Error('Pubsub is not started')
+    }
+
+    return Array.from(this.subscriptions)
+  }
+}
+
+module.exports = PubsubBaseProtocol
+module.exports.message = message
+module.exports.utils = utils

src/pubsub/message/index.d.ts

@@ -0,0 +1,5 @@
+export var rpc: any;
+export var td: any;
+export var RPC: any;
+export var Message: any;
+export var SubOpts: any;

src/pubsub/message/index.js

@@ -0,0 +1,17 @@
+'use strict'
+
+const protons = require('protons')
+
+const rpcProto = protons(require('./rpc.proto.js'))
+const RPC = rpcProto.RPC
+const topicDescriptorProto = protons(require('./topic-descriptor.proto.js'))
+
+/**
+ * @module pubsub/message/index
+ */
+exports = module.exports
+exports.rpc = rpcProto
+exports.td = topicDescriptorProto
+exports.RPC = RPC
+exports.Message = RPC.Message
+exports.SubOpts = RPC.SubOpts

src/pubsub/message/rpc.proto.d.ts

@@ -0,0 +1,2 @@
+declare const _exports: string;
+export = _exports;

src/pubsub/message/rpc.proto.js

@@ -0,0 +1,20 @@
+'use strict'
+module.exports = `
+message RPC {
+  repeated SubOpts subscriptions = 1;
+  repeated Message msgs = 2;
+
+  message SubOpts {
+    optional bool subscribe = 1; // subscribe or unsubcribe
+    optional string topicID = 2;
+  }
+
+  message Message {
+    optional bytes from = 1;
+    optional bytes data = 2;
+    optional bytes seqno = 3;
+    repeated string topicIDs = 4;
+    optional bytes signature = 5;
+    optional bytes key = 6;
+  }
+}`

src/pubsub/message/sign.d.ts

@@ -0,0 +1,23 @@
+/**
+ * Returns the PublicKey associated with the given message.
+ * If no, valid PublicKey can be retrieved an error will be returned.
+ *
+ * @param {InMessage} message
+ * @returns {Promise<PublicKey>}
+ */
+export function messagePublicKey(message: any): Promise<any>;
+/**
+ * Signs the provided message with the given `peerId`
+ *
+ * @param {PeerId} peerId
+ * @param {Message} message
+ * @returns {Promise<Message>}
+ */
+export function signMessage(peerId: import("peer-id"), message: any): Promise<any>;
+export const SignPrefix: any;
+/**
+ * Verifies the signature of the given message
+ * @param {InMessage} message
+ * @returns {Promise<Boolean>}
+ */
+export function verifySignature(message: any): Promise<boolean>;

src/pubsub/message/sign.js

@@ -0,0 +1,85 @@
+'use strict'
+
+const PeerId = require('peer-id')
+const { Message } = require('./index')
+const uint8ArrayConcat = require('uint8arrays/concat')
+const uint8ArrayFromString = require('uint8arrays/from-string')
+const SignPrefix = uint8ArrayFromString('libp2p-pubsub:')
+
+/**
+ * Signs the provided message with the given `peerId`
+ *
+ * @param {PeerId} peerId
+ * @param {Message} message
+ * @returns {Promise<Message>}
+ */
+async function signMessage (peerId, message) {
+  // Get the message in bytes, and prepend with the pubsub prefix
+  const bytes = uint8ArrayConcat([
+    SignPrefix,
+    Message.encode(message)
+  ])
+
+  const signature = await peerId.privKey.sign(bytes)
+
+  return {
+    ...message,
+    signature: signature,
+    key: peerId.pubKey.bytes
+  }
+}
+
+/**
+ * Verifies the signature of the given message
+ * @param {InMessage} message
+ * @returns {Promise<Boolean>}
+ */
+async function verifySignature (message) {
+  // Get message sans the signature
+  const baseMessage = { ...message }
+  delete baseMessage.signature
+  delete baseMessage.key
+  baseMessage.from = PeerId.createFromCID(baseMessage.from).toBytes()
+  const bytes = uint8ArrayConcat([
+    SignPrefix,
+    Message.encode(baseMessage)
+  ])
+
+  // Get the public key
+  const pubKey = await messagePublicKey(message)
+
+  // verify the base message
+  return pubKey.verify(bytes, message.signature)
+}
+
+/**
+ * Returns the PublicKey associated with the given message.
+ * If no, valid PublicKey can be retrieved an error will be returned.
+ *
+ * @param {InMessage} message
+ * @returns {Promise<PublicKey>}
+ */
+async function messagePublicKey (message) {
+  // should be available in the from property of the message (peer id)
+  const from = PeerId.createFromCID(message.from)
+
+  if (message.key) {
+    const keyPeerId = await PeerId.createFromPubKey(message.key)
+
+    // the key belongs to the sender, return the key
+    if (keyPeerId.isEqual(from)) return keyPeerId.pubKey
+    // We couldn't validate pubkey is from the originator, error
+    throw new Error('Public Key does not match the originator')
+  } else if (from.pubKey) {
+    return from.pubKey
+  } else {
+    throw new Error('Could not get the public key from the originator id')
+  }
+}
+
+module.exports = {
+  messagePublicKey,
+  signMessage,
+  SignPrefix,
+  verifySignature
+}

src/pubsub/message/topic-descriptor.proto.d.ts

@@ -0,0 +1,2 @@
+declare const _exports: string;
+export = _exports;

src/pubsub/message/topic-descriptor.proto.js

@@ -0,0 +1,30 @@
+'use strict'
+module.exports = `
+// topicCID = cid(merkledag_protobuf(topicDescriptor)); (not the topic.name)
+message TopicDescriptor {
+  optional string name = 1;
+  optional AuthOpts auth = 2;
+  optional EncOpts enc = 2;
+
+  message AuthOpts {
+    optional AuthMode mode = 1;
+    repeated bytes keys = 2; // root keys to trust
+
+    enum AuthMode {
+      NONE = 0; // no authentication, anyone can publish
+      KEY = 1; // only messages signed by keys in the topic descriptor are accepted
+      WOT = 2; // web of trust, certificates can allow publisher set to grow
+    }
+  }
+
+  message EncOpts {
+    optional EncMode mode = 1;
+    repeated bytes keyHashes = 2; // the hashes of the shared keys used (salted)
+
+    enum EncMode {
+      NONE = 0; // no encryption, anyone can read
+      SHAREDKEY = 1; // messages are encrypted with shared key
+      WOT = 2; // web of trust, certificates can allow publisher set to grow
+    }
+  }
+}`

src/pubsub/peer-streams.d.ts

@@ -0,0 +1,113 @@
+export = PeerStreams;
+/**
+ * @callback Sink
+ * @param {Uint8Array} source
+ * @returns {Promise<Uint8Array>}
+ *
+ * @typedef {object} DuplexIterableStream
+ * @property {Sink} sink
+ * @property {() AsyncIterator<Uint8Array>} source
+ *
+ * @typedef PeerId
+ * @type import('peer-id')
+ */
+/**
+ * Thin wrapper around a peer's inbound / outbound pubsub streams
+ */
+declare class PeerStreams {
+    /**
+     * @param {object} properties properties of the PeerStreams.
+     * @param {PeerId} properties.id
+     * @param {string} properties.protocol
+     */
+    constructor({ id, protocol }: {
+        id: import("peer-id");
+        protocol: string;
+    });
+    /**
+     * @type {import('peer-id')}
+     */
+    id: import('peer-id');
+    /**
+     * Established protocol
+     * @type {string}
+     */
+    protocol: string;
+    /**
+     * The raw outbound stream, as retrieved from conn.newStream
+     * @private
+     * @type {DuplexIterableStream}
+     */
+    _rawOutboundStream: DuplexIterableStream;
+    /**
+     * The raw inbound stream, as retrieved from the callback from libp2p.handle
+     * @private
+     * @type {DuplexIterableStream}
+     */
+    _rawInboundStream: DuplexIterableStream;
+    /**
+     * An AbortController for controlled shutdown of the inbound stream
+     * @private
+     * @type {typeof AbortController}
+     */
+    _inboundAbortController: typeof AbortController;
+    /**
+     * Write stream -- its preferable to use the write method
+     * @type {import('it-pushable').Pushable<Uint8Array>>}
+     */
+    outboundStream: import('it-pushable').Pushable<Uint8Array>;
+    /**
+     * Read stream
+     * @type {DuplexIterableStream}
+     */
+    inboundStream: DuplexIterableStream;
+    /**
+     * Do we have a connection to read from?
+     *
+     * @type {boolean}
+     */
+    get isReadable(): boolean;
+    /**
+     * Do we have a connection to write on?
+     *
+     * @type {boolean}
+     */
+    get isWritable(): boolean;
+    /**
+     * Send a message to this peer.
+     * Throws if there is no `stream` to write to available.
+     *
+     * @param {Uint8Array} data
+     * @returns {void}
+     */
+    write(data: Uint8Array): void;
+    /**
+     * Attach a raw inbound stream and setup a read stream
+     *
+     * @param {DuplexIterableStream} stream
+     * @returns {void}
+     */
+    attachInboundStream(stream: DuplexIterableStream): void;
+    /**
+     * Attach a raw outbound stream and setup a write stream
+     *
+     * @param {Stream} stream
+     * @returns {Promise<void>}
+     */
+    attachOutboundStream(stream: any): Promise<void>;
+    /**
+     * Closes the open connection to peer
+     * @returns {void}
+     */
+    close(): void;
+}
+declare namespace PeerStreams {
+    export { Sink, DuplexIterableStream, PeerId };
+}
+type DuplexIterableStream = {
+    sink: Sink;
+    source: () => AsyncIterator<Uint8Array, any, undefined>;
+};
+declare const AbortController: typeof import("abort-controller");
+type Sink = (source: Uint8Array) => Promise<Uint8Array>;
+type PeerId = import("peer-id");

src/pubsub/peer-streams.js

@@ -0,0 +1,202 @@
+'use strict'
+
+const EventEmitter = require('events')
+
+const lp = require('it-length-prefixed')
+const pushable = require('it-pushable')
+const pipe = require('it-pipe')
+const abortable = require('abortable-iterator')
+const AbortController = require('abort-controller')
+const debug = require('debug')
+
+const log = debug('libp2p-pubsub:peer-streams')
+log.error = debug('libp2p-pubsub:peer-streams:error')
+
+/**
+ * @callback Sink
+ * @param {Uint8Array} source
+ * @returns {Promise<Uint8Array>}
+ *
+ * @typedef {object} DuplexIterableStream
+ * @property {Sink} sink
+ * @property {() AsyncIterator<Uint8Array>} source
+ *
+ * @typedef PeerId
+ * @type import('peer-id')
+ */
+
+/**
+ * Thin wrapper around a peer's inbound / outbound pubsub streams
+ */
+class PeerStreams extends EventEmitter {
+  /**
+   * @param {object} properties properties of the PeerStreams.
+   * @param {PeerId} properties.id
+   * @param {string} properties.protocol
+   */
+  constructor ({ id, protocol }) {
+    super()
+
+    /**
+     * @type {import('peer-id')}
+     */
+    this.id = id
+    /**
+     * Established protocol
+     * @type {string}
+     */
+    this.protocol = protocol
+    /**
+     * The raw outbound stream, as retrieved from conn.newStream
+     * @private
+     * @type {DuplexIterableStream}
+     */
+    this._rawOutboundStream = null
+    /**
+     * The raw inbound stream, as retrieved from the callback from libp2p.handle
+     * @private
+     * @type {DuplexIterableStream}
+     */
+    this._rawInboundStream = null
+    /**
+     * An AbortController for controlled shutdown of the inbound stream
+     * @private
+     * @type {typeof AbortController}
+     */
+    this._inboundAbortController = null
+    /**
+     * Write stream -- its preferable to use the write method
+     * @type {import('it-pushable').Pushable<Uint8Array>>}
+     */
+    this.outboundStream = null
+    /**
+     * Read stream
+     * @type {DuplexIterableStream}
+     */
+    this.inboundStream = null
+  }
+
+  /**
+   * Do we have a connection to read from?
+   *
+   * @type {boolean}
+   */
+  get isReadable () {
+    return Boolean(this.inboundStream)
+  }
+
+  /**
+   * Do we have a connection to write on?
+   *
+   * @type {boolean}
+   */
+  get isWritable () {
+    return Boolean(this.outboundStream)
+  }
+
+  /**
+   * Send a message to this peer.
+   * Throws if there is no `stream` to write to available.
+   *
+   * @param {Uint8Array} data
+   * @returns {void}
+   */
+  write (data) {
+    if (!this.isWritable) {
+      const id = this.id.toB58String()
+      throw new Error('No writable connection to ' + id)
+    }
+
+    this.outboundStream.push(data)
+  }
+
+  /**
+   * Attach a raw inbound stream and setup a read stream
+   *
+   * @param {DuplexIterableStream} stream
+   * @returns {void}
+   */
+  attachInboundStream (stream) {
+    // Create and attach a new inbound stream
+    // The inbound stream is:
+    // - abortable, set to only return on abort, rather than throw
+    // - transformed with length-prefix transform
+    this._inboundAbortController = new AbortController()
+    this._rawInboundStream = stream
+    this.inboundStream = abortable(
+      pipe(
+        this._rawInboundStream,
+        lp.decode()
+      ),
+      this._inboundAbortController.signal,
+      { returnOnAbort: true }
+    )
+
+    this.emit('stream:inbound')
+  }
+
+  /**
+   * Attach a raw outbound stream and setup a write stream
+   *
+   * @param {Stream} stream
+   * @returns {Promise<void>}
+   */
+  async attachOutboundStream (stream) {
+    // If an outbound stream already exists,
+    // gently close it
+    const _prevStream = this.outboundStream
+    if (_prevStream) {
+      // End the stream without emitting a close event
+      await this.outboundStream.end(false)
+    }
+
+    this._rawOutboundStream = stream
+    this.outboundStream = pushable({
+      onEnd: (shouldEmit) => {
+        // close writable side of the stream
+        this._rawOutboundStream.reset && this._rawOutboundStream.reset()
+        this._rawOutboundStream = null
+        this.outboundStream = null
+        if (shouldEmit !== false) {
+          this.emit('close')
+        }
+      }
+    })
+
+    pipe(
+      this.outboundStream,
+      lp.encode(),
+      this._rawOutboundStream
+    ).catch(err => {
+      log.error(err)
+    })
+
+    // Only emit if the connection is new
+    if (!_prevStream) {
+      this.emit('stream:outbound')
+    }
+  }
+
+  /**
+   * Closes the open connection to peer
+   * @returns {void}
+   */
+  close () {
+    // End the outbound stream
+    if (this.outboundStream) {
+      this.outboundStream.end()
+    }
+    // End the inbound stream
+    if (this.inboundStream) {
+      this._inboundAbortController.abort()
+    }
+
+    this._rawOutboundStream = null
+    this.outboundStream = null
+    this._rawInboundStream = null
+    this.inboundStream = null
+    this.emit('close')
+  }
+}
+
+module.exports = PeerStreams

src/pubsub/tests/api.js

@@ -0,0 +1,93 @@
+/* eslint-env mocha */
+'use strict'
+
+const chai = require('chai')
+const { expect } = chai
+const sinon = require('sinon')
+
+const pDefer = require('p-defer')
+const pWaitFor = require('p-wait-for')
+const uint8ArrayFromString = require('uint8arrays/from-string')
+
+const topic = 'foo'
+const data = uint8ArrayFromString('bar')
+
+module.exports = (common) => {
+  describe('pubsub api', () => {
+    let pubsub
+
+    // Create pubsub router
+    beforeEach(async () => {
+      [pubsub] = await common.setup(1)
+    })
+
+    afterEach(async () => {
+      sinon.restore()
+      pubsub && pubsub.stop()
+      await common.teardown()
+    })
+
+    it('can start correctly', () => {
+      sinon.spy(pubsub.registrar, '_handle')
+      sinon.spy(pubsub.registrar, 'register')
+
+      pubsub.start()
+
+      expect(pubsub.started).to.eql(true)
+      expect(pubsub.registrar._handle.callCount).to.eql(1)
+      expect(pubsub.registrar.register.callCount).to.eql(1)
+    })
+
+    it('can stop correctly', () => {
+      sinon.spy(pubsub.registrar, 'unregister')
+
+      pubsub.start()
+      pubsub.stop()
+
+      expect(pubsub.started).to.eql(false)
+      expect(pubsub.registrar.unregister.callCount).to.eql(1)
+    })
+
+    it('can subscribe and unsubscribe correctly', async () => {
+      const handler = () => {
+        throw new Error('a message should not be received')
+      }
+
+      pubsub.start()
+      pubsub.subscribe(topic)
+      pubsub.on('topic', handler)
+
+      await pWaitFor(() => {
+        const topics = pubsub.getTopics()
+        return topics.length === 1 && topics[0] === topic
+      })
+
+      pubsub.unsubscribe(topic)
+
+      await pWaitFor(() => !pubsub.getTopics().length)
+
+      // Publish to guarantee the handler is not called
+      await pubsub.publish(topic, data)
+
+      pubsub.stop()
+    })
+
+    it('can subscribe and publish correctly', async () => {
+      const defer = pDefer()
+
+      const handler = (msg) => {
+        expect(msg).to.exist()
+        defer.resolve()
+      }
+
+      pubsub.start()
+
+      pubsub.subscribe(topic)
+      pubsub.on(topic, handler)
+      await pubsub.publish(topic, data)
+      await defer.promise
+
+      pubsub.stop()
+    })
+  })
+}

src/pubsub/tests/emit-self.js

@@ -0,0 +1,69 @@
+/* eslint-env mocha */
+'use strict'
+
+const chai = require('chai')
+const { expect } = chai
+const sinon = require('sinon')
+
+const uint8ArrayFromString = require('uint8arrays/from-string')
+
+const topic = 'foo'
+const data = uint8ArrayFromString('bar')
+const shouldNotHappen = (_) => expect.fail()
+
+module.exports = (common) => {
+  describe('emit self', () => {
+    let pubsub
+
+    describe('enabled', () => {
+      before(async () => {
+        [pubsub] = await common.setup(1, { emitSelf: true })
+      })
+
+      before(() => {
+        pubsub.start()
+        pubsub.subscribe(topic)
+      })
+
+      after(async () => {
+        sinon.restore()
+        pubsub && pubsub.stop()
+        await common.teardown()
+      })
+
+      it('should emit to self on publish', () => {
+        const promise = new Promise((resolve) => pubsub.once(topic, resolve))
+
+        pubsub.publish(topic, data)
+
+        return promise
+      })
+    })
+
+    describe('disabled', () => {
+      before(async () => {
+        [pubsub] = await common.setup(1, { emitSelf: false })
+      })
+
+      before(() => {
+        pubsub.start()
+        pubsub.subscribe(topic)
+      })
+
+      after(async () => {
+        sinon.restore()
+        pubsub && pubsub.stop()
+        await common.teardown()
+      })
+
+      it('should not emit to self on publish', () => {
+        pubsub.once(topic, (m) => shouldNotHappen)
+
+        pubsub.publish(topic, data)
+
+        // Wait 1 second to guarantee that self is not noticed
+        return new Promise((resolve) => setTimeout(() => resolve(), 1000))
+      })
+    })
+  })
+}

src/pubsub/tests/index.js

@@ -0,0 +1,18 @@
+/* eslint-env mocha */
+'use strict'
+
+const apiTest = require('./api')
+const emitSelfTest = require('./emit-self')
+const messagesTest = require('./messages')
+const twoNodesTest = require('./two-nodes')
+const multipleNodesTest = require('./multiple-nodes')
+
+module.exports = (common) => {
+  describe('interface-pubsub', () => {
+    apiTest(common)
+    emitSelfTest(common)
+    messagesTest(common)
+    twoNodesTest(common)
+    multipleNodesTest(common)
+  })
+}

src/pubsub/tests/messages.js

@@ -0,0 +1,116 @@
+/* eslint-env mocha */
+'use strict'
+
+const chai = require('chai')
+const { expect } = chai
+const sinon = require('sinon')
+
+const PeerId = require('peer-id')
+const uint8ArrayFromString = require('uint8arrays/from-string')
+
+const { utils } = require('..')
+const PeerStreams = require('../peer-streams')
+
+const topic = 'foo'
+const data = uint8ArrayFromString('bar')
+
+module.exports = (common) => {
+  describe('messages', () => {
+    let pubsub
+
+    // Create pubsub router
+    beforeEach(async () => {
+      [pubsub] = await common.setup(1)
+      pubsub.start()
+    })
+
+    afterEach(async () => {
+      sinon.restore()
+      pubsub && pubsub.stop()
+      await common.teardown()
+    })
+
+    it('should emit normalized signed messages on publish', async () => {
+      sinon.spy(pubsub, '_emitMessage')
+      sinon.spy(utils, 'randomSeqno')
+
+      await pubsub.publish(topic, data)
+      expect(pubsub._emitMessage.callCount).to.eql(1)
+
+      const [messageToEmit] = pubsub._emitMessage.getCall(0).args
+
+      const expected = utils.normalizeInRpcMessage(
+        await pubsub._buildMessage({
+          receivedFrom: pubsub.peerId.toB58String(),
+          from: pubsub.peerId.toB58String(),
+          data,
+          seqno: utils.randomSeqno.getCall(0).returnValue,
+          topicIDs: [topic]
+        }))
+
+      expect(messageToEmit).to.eql(expected)
+    })
+
+    it('should drop unsigned messages', async () => {
+      sinon.spy(pubsub, '_emitMessage')
+      sinon.spy(pubsub, '_publish')
+      sinon.spy(pubsub, 'validate')
+
+      const peerStream = new PeerStreams({ id: await PeerId.create() })
+      const rpc = {
+        subscriptions: [],
+        msgs: [{
+          receivedFrom: peerStream.id.toB58String(),
+          from: peerStream.id.toBytes(),
+          data,
+          seqno: utils.randomSeqno(),
+          topicIDs: [topic]
+        }]
+      }
+
+      pubsub.subscribe(topic)
+      pubsub._processRpc(peerStream.id.toB58String(), peerStream, rpc)
+
+      return new Promise((resolve) => {
+        setTimeout(() => {
+          expect(pubsub.validate.callCount).to.eql(1)
+          expect(pubsub._emitMessage.called).to.eql(false)
+          expect(pubsub._publish.called).to.eql(false)
+
+          resolve()
+        }, 50)
+      })
+    })
+
+    it('should not drop unsigned messages if strict signing is disabled', async () => {
+      sinon.spy(pubsub, '_emitMessage')
+      sinon.spy(pubsub, '_publish')
+      sinon.spy(pubsub, 'validate')
+      sinon.stub(pubsub, 'strictSigning').value(false)
+
+      const peerStream = new PeerStreams({ id: await PeerId.create() })
+      const rpc = {
+        subscriptions: [],
+        msgs: [{
+          from: peerStream.id.toBytes(),
+          data,
+          seqno: utils.randomSeqno(),
+          topicIDs: [topic]
+        }]
+      }
+
+      pubsub.subscribe(topic)
+      pubsub._processRpc(peerStream.id.toB58String(), peerStream, rpc)
+
+      return new Promise((resolve) => {
+        setTimeout(() => {
+          expect(pubsub.validate.callCount).to.eql(1)
+          expect(pubsub._emitMessage.called).to.eql(true)
+          expect(pubsub._publish.called).to.eql(true)
+
+          resolve()
+        }, 50)
+      })
+    })
+  })
+}

src/pubsub/tests/multiple-nodes.js

@@ -0,0 +1,346 @@
+/* eslint-env mocha */
+/* eslint max-nested-callbacks: ["error", 6] */
+'use strict'
+
+const chai = require('chai')
+const { expect } = chai
+const sinon = require('sinon')
+
+const delay = require('delay')
+const pDefer = require('p-defer')
+const pWaitFor = require('p-wait-for')
+const uint8ArrayFromString = require('uint8arrays/from-string')
+const uint8ArrayToString = require('uint8arrays/to-string')
+
+const { expectSet } = require('./utils')
+
+module.exports = (common) => {
+  describe('pubsub with multiple nodes', function () {
+    this.timeout(10e3)
+    describe('every peer subscribes to the topic', () => {
+      describe('line', () => {
+        // line
+        // ◉────◉────◉
+        // a    b    c
+        let psA, psB, psC
+
+        // Create and start pubsub nodes
+        beforeEach(async () => {
+          [psA, psB, psC] = await common.setup(3)
+
+          // Start pubsub mpdes
+          ;[psA, psB, psC].map((p) => p.start())
+        })
+
+        // Connect nodes
+        beforeEach(async () => {
+          await psA._libp2p.dial(psB.peerId)
+          await psB._libp2p.dial(psC.peerId)
+
+          // Wait for peers to be ready in pubsub
+          await pWaitFor(() =>
+            psA.peers.size === 1 &&
+            psC.peers.size === 1 &&
+            psA.peers.size === 1
+          )
+        })
+
+        afterEach(async () => {
+          sinon.restore()
+
+          ;[psA, psB, psC].map((p) => p.stop())
+          await common.teardown()
+        })
+
+        it('subscribe to the topic on node a', () => {
+          const topic = 'Z'
+          const defer = pDefer()
+
+          psA.subscribe(topic)
+          expectSet(psA.subscriptions, [topic])
+
+          psB.once('pubsub:subscription-change', () => {
+            expect(psB.peers.size).to.equal(2)
+
+            const aPeerId = psA.peerId.toB58String()
+            expectSet(psB.topics.get(topic), [aPeerId])
+
+            expect(psC.peers.size).to.equal(1)
+            expect(psC.topics.get(topic)).to.not.exist()
+
+            defer.resolve()
+          })
+
+          return defer.promise
+        })
+
+        it('subscribe to the topic on node b', async () => {
+          const topic = 'Z'
+          psB.subscribe(topic)
+          expectSet(psB.subscriptions, [topic])
+
+          await Promise.all([
+            new Promise((resolve) => psA.once('pubsub:subscription-change', resolve)),
+            new Promise((resolve) => psC.once('pubsub:subscription-change', resolve))
+          ])
+
+          expect(psA.peers.size).to.equal(1)
+          expectSet(psA.topics.get(topic), [psB.peerId.toB58String()])
+
+          expect(psC.peers.size).to.equal(1)
+          expectSet(psC.topics.get(topic), [psB.peerId.toB58String()])
+        })
+
+        it('subscribe to the topic on node c', () => {
+          const topic = 'Z'
+          const defer = pDefer()
+
+          psC.subscribe(topic)
+          expectSet(psC.subscriptions, [topic])
+
+          psB.once('pubsub:subscription-change', () => {
+            expect(psA.peers.size).to.equal(1)
+            expect(psB.peers.size).to.equal(2)
+            expectSet(psB.topics.get(topic), [psC.peerId.toB58String()])
+
+            defer.resolve()
+          })
+
+          return defer.promise
+        })
+
+        it('publish on node a', async () => {
+          const topic = 'Z'
+          const defer = pDefer()
+
+          psA.subscribe(topic)
+          psB.subscribe(topic)
+          psC.subscribe(topic)
+
+          // await subscription change
+          await Promise.all([
+            new Promise(resolve => psA.once('pubsub:subscription-change', () => resolve())),
+            new Promise(resolve => psB.once('pubsub:subscription-change', () => resolve())),
+            new Promise(resolve => psC.once('pubsub:subscription-change', () => resolve()))
+          ])
+
+          // await a cycle
+          await delay(1000)
+
+          let counter = 0
+
+          psA.on(topic, incMsg)
+          psB.on(topic, incMsg)
+          psC.on(topic, incMsg)
+
+          psA.publish(topic, uint8ArrayFromString('hey'))
+
+          function incMsg (msg) {
+            expect(uint8ArrayToString(msg.data)).to.equal('hey')
+            check()
+          }
+
+          function check () {
+            if (++counter === 3) {
+              psA.removeListener(topic, incMsg)
+              psB.removeListener(topic, incMsg)
+              psC.removeListener(topic, incMsg)
+              defer.resolve()
+            }
+          }
+
+          return defer.promise
+        })
+
+        // since the topology is the same, just the publish
+        // gets sent by other peer, we reused the same peers
+        describe('1 level tree', () => {
+          // 1 level tree
+          //     ┌◉┐
+          //     │b│
+          //   ◉─┘ └─◉
+          //   a     c
+
+          it('publish on node b', async () => {
+            const topic = 'Z'
+            const defer = pDefer()
+            let counter = 0
+
+            psA.subscribe(topic)
+            psB.subscribe(topic)
+            psC.subscribe(topic)
+
+            // await subscription change
+            await Promise.all([
+              new Promise(resolve => psA.once('pubsub:subscription-change', () => resolve())),
+              new Promise(resolve => psB.once('pubsub:subscription-change', () => resolve())),
+              new Promise(resolve => psC.once('pubsub:subscription-change', () => resolve()))
+            ])
+
+            psA.on(topic, incMsg)
+            psB.on(topic, incMsg)
+            psC.on(topic, incMsg)
+
+            // await a cycle
+            await delay(1000)
+
+            psB.publish(topic, uint8ArrayFromString('hey'))
+
+            function incMsg (msg) {
+              expect(uint8ArrayToString(msg.data)).to.equal('hey')
+              check()
+            }
+
+            function check () {
+              if (++counter === 3) {
+                psA.removeListener(topic, incMsg)
+                psB.removeListener(topic, incMsg)
+                psC.removeListener(topic, incMsg)
+                defer.resolve()
+              }
+            }
+
+            return defer.promise
+          })
+        })
+      })
+
+      describe('2 level tree', () => {
+        // 2 levels tree
+        //      ┌◉┐
+        //      │c│
+        //   ┌◉─┘ └─◉┐
+        //   │b     d│
+        // ◉─┘       └─◉
+        // a
+        let psA, psB, psC, psD, psE
+
+        // Create and start pubsub nodes
+        beforeEach(async () => {
+          [psA, psB, psC, psD, psE] = await common.setup(5)
+
+          // Start pubsub nodes
+          ;[psA, psB, psC, psD, psE].map((p) => p.start())
+        })
+
+        // connect nodes
+        beforeEach(async () => {
+          await psA._libp2p.dial(psB.peerId)
+          await psB._libp2p.dial(psC.peerId)
+          await psC._libp2p.dial(psD.peerId)
+          await psD._libp2p.dial(psE.peerId)
+
+          // Wait for peers to be ready in pubsub
+          await pWaitFor(() =>
+            psA.peers.size === 1 &&
+            psB.peers.size === 2 &&
+            psC.peers.size === 2 &&
+            psD.peers.size === 2 &&
+            psE.peers.size === 1
+          )
+        })
+
+        afterEach(async () => {
+          [psA, psB, psC, psD, psE].map((p) => p.stop())
+          await common.teardown()
+        })
+
+        it('subscribes', () => {
+          psA.subscribe('Z')
+          expectSet(psA.subscriptions, ['Z'])
+          psB.subscribe('Z')
+          expectSet(psB.subscriptions, ['Z'])
+          psC.subscribe('Z')
+          expectSet(psC.subscriptions, ['Z'])
+          psD.subscribe('Z')
+          expectSet(psD.subscriptions, ['Z'])
+          psE.subscribe('Z')
+          expectSet(psE.subscriptions, ['Z'])
+        })
+
+        it('publishes from c', async function () {
+          this.timeout(30 * 1000)
+          const defer = pDefer()
+          let counter = 0
+
+          psA.subscribe('Z')
+          psA.on('Z', incMsg)
+          psB.subscribe('Z')
+          psB.on('Z', incMsg)
+          psC.subscribe('Z')
+          psC.on('Z', incMsg)
+          psD.subscribe('Z')
+          psD.on('Z', incMsg)
+          psE.subscribe('Z')
+          psE.on('Z', incMsg)
+
+          await Promise.all([
+            new Promise((resolve) => psA.once('pubsub:subscription-change', resolve)),
+            new Promise((resolve) => psB.once('pubsub:subscription-change', resolve)),
+            new Promise((resolve) => psC.once('pubsub:subscription-change', resolve)),
+            new Promise((resolve) => psD.once('pubsub:subscription-change', resolve)),
+            new Promise((resolve) => psE.once('pubsub:subscription-change', resolve))
+          ])
+
+          // await a cycle
+          await delay(1000)
+
+          psC.publish('Z', uint8ArrayFromString('hey from c'))
+
+          function incMsg (msg) {
+            expect(uint8ArrayToString(msg.data)).to.equal('hey from c')
+            check()
+          }
+
+          function check () {
+            if (++counter === 5) {
+              psA.unsubscribe('Z', incMsg)
+              psB.unsubscribe('Z', incMsg)
+              psC.unsubscribe('Z', incMsg)
+              psD.unsubscribe('Z', incMsg)
+              psE.unsubscribe('Z', incMsg)
+              defer.resolve()
+            }
+          }
+
+          return defer.promise
+        })
+      })
+    })
+
+    describe('only some nodes subscribe the networks', () => {
+      describe('line', () => {
+        // line
+        // ◉────◎────◉
+        // a    b    c
+
+        before(() => { })
+        after(() => { })
+      })
+
+      describe('1 level tree', () => {
+        // 1 level tree
+        //     ┌◉┐
+        //     │b│
+        //   ◎─┘ └─◉
+        //   a     c
+
+        before(() => { })
+        after(() => { })
+      })
+
+      describe('2 level tree', () => {
+        // 2 levels tree
+        //      ┌◉┐
+        //      │c│
+        //   ┌◎─┘ └─◉┐
+        //   │b     d│
+        // ◉─┘       └─◎
+        // a           e
+
+        before(() => { })
+        after(() => { })
+      })
+    })
+  })
+}

src/pubsub/tests/two-nodes.js

@@ -0,0 +1,228 @@
+/* eslint-env mocha */
+/* eslint max-nested-callbacks: ["error", 6] */
+'use strict'
+
+const chai = require('chai')
+const { expect } = chai
+const sinon = require('sinon')
+
+const pDefer = require('p-defer')
+const pWaitFor = require('p-wait-for')
+const uint8ArrayFromString = require('uint8arrays/from-string')
+const uint8ArrayToString = require('uint8arrays/to-string')
+
+const {
+  first,
+  expectSet
+} = require('./utils')
+
+const topic = 'foo'
+
+function shouldNotHappen (_) {
+  expect.fail()
+}
+
+module.exports = (common) => {
+  describe('pubsub with two nodes', () => {
+    describe('fresh nodes', () => {
+      let psA, psB
+
+      // Create pubsub nodes and connect them
+      before(async () => {
+        [psA, psB] = await common.setup(2)
+
+        expect(psA.peers.size).to.be.eql(0)
+        expect(psB.peers.size).to.be.eql(0)
+
+        // Start pubsub and connect nodes
+        psA.start()
+        psB.start()
+
+        await psA._libp2p.dial(psB.peerId)
+
+        // Wait for peers to be ready in pubsub
+        await pWaitFor(() => psA.peers.size === 1 && psB.peers.size === 1)
+      })
+
+      after(async () => {
+        sinon.restore()
+
+        psA && psA.stop()
+        psB && psB.stop()
+
+        await common.teardown()
+      })
+
+      it('Subscribe to a topic in nodeA', () => {
+        const defer = pDefer()
+
+        psB.once('pubsub:subscription-change', (changedPeerId, changedSubs) => {
+          expectSet(psA.subscriptions, [topic])
+          expect(psB.peers.size).to.equal(1)
+          expectSet(psB.topics.get(topic), [psA.peerId.toB58String()])
+          expect(changedPeerId.toB58String()).to.equal(first(psB.peers).id.toB58String())
+          expect(changedSubs).to.be.eql([{ topicID: topic, subscribe: true }])
+          defer.resolve()
+        })
+        psA.subscribe(topic)
+
+        return defer.promise
+      })
+
+      it('Publish to a topic in nodeA', () => {
+        const defer = pDefer()
+
+        psA.once(topic, (msg) => {
+          expect(uint8ArrayToString(msg.data)).to.equal('hey')
+          psB.removeListener(topic, shouldNotHappen)
+          defer.resolve()
+        })
+
+        psB.once(topic, shouldNotHappen)
+
+        psA.publish(topic, uint8ArrayFromString('hey'))
+
+        return defer.promise
+      })
+
+      it('Publish to a topic in nodeB', () => {
+        const defer = pDefer()
+
+        psA.once(topic, (msg) => {
+          psA.once(topic, shouldNotHappen)
+          expect(uint8ArrayToString(msg.data)).to.equal('banana')
+
+          setTimeout(() => {
+            psA.removeListener(topic, shouldNotHappen)
+            psB.removeListener(topic, shouldNotHappen)
+
+            defer.resolve()
+          }, 100)
+        })
+
+        psB.once(topic, shouldNotHappen)
+
+        psB.publish(topic, uint8ArrayFromString('banana'))
+
+        return defer.promise
+      })
+
+      it('Publish 10 msg to a topic in nodeB', () => {
+        const defer = pDefer()
+        let counter = 0
+
+        psB.once(topic, shouldNotHappen)
+        psA.on(topic, receivedMsg)
+
+        function receivedMsg (msg) {
+          expect(uint8ArrayToString(msg.data)).to.equal('banana')
+          expect(msg.from).to.be.eql(psB.peerId.toB58String())
+          expect(msg.seqno).to.be.a('Uint8Array')
+          expect(msg.topicIDs).to.be.eql([topic])
+
+          if (++counter === 10) {
+            psA.removeListener(topic, receivedMsg)
+            psB.removeListener(topic, shouldNotHappen)
+
+            defer.resolve()
+          }
+        }
+
+        Array.from({ length: 10 }, (_, i) => psB.publish(topic, uint8ArrayFromString('banana')))
+
+        return defer.promise
+      })
+
+      it('Unsubscribe from topic in nodeA', () => {
+        const defer = pDefer()
+
+        psA.unsubscribe(topic)
+        expect(psA.subscriptions.size).to.equal(0)
+
+        psB.once('pubsub:subscription-change', (changedPeerId, changedSubs) => {
+          expect(psB.peers.size).to.equal(1)
+          expectSet(psB.topics.get(topic), [])
+          expect(changedPeerId.toB58String()).to.equal(first(psB.peers).id.toB58String())
+          expect(changedSubs).to.be.eql([{ topicID: topic, subscribe: false }])
+
+          defer.resolve()
+        })
+
+        return defer.promise
+      })
+
+      it('Publish to a topic:Z in nodeA nodeB', () => {
+        const defer = pDefer()
+
+        psA.once('Z', shouldNotHappen)
+        psB.once('Z', shouldNotHappen)
+
+        setTimeout(() => {
+          psA.removeListener('Z', shouldNotHappen)
+          psB.removeListener('Z', shouldNotHappen)
+          defer.resolve()
+        }, 100)
+
+        psB.publish('Z', uint8ArrayFromString('banana'))
+        psA.publish('Z', uint8ArrayFromString('banana'))
+
+        return defer.promise
+      })
+    })
+
+    describe('nodes send state on connection', () => {
+      let psA, psB
+
+      // Create pubsub nodes and connect them
+      before(async () => {
+        [psA, psB] = await common.setup(2)
+
+        expect(psA.peers.size).to.be.eql(0)
+        expect(psB.peers.size).to.be.eql(0)
+
+        // Start pubsub and connect nodes
+        psA.start()
+        psB.start()
+      })
+
+      // Make subscriptions prior to nodes connected
+      before(() => {
+        psA.subscribe('Za')
+        psB.subscribe('Zb')
+
+        expect(psA.peers.size).to.equal(0)
+        expectSet(psA.subscriptions, ['Za'])
+        expect(psB.peers.size).to.equal(0)
+        expectSet(psB.subscriptions, ['Zb'])
+      })
+
+      after(async () => {
+        sinon.restore()
+
+        psA && psA.stop()
+        psB && psB.stop()
+
+        await common.teardown()
+      })
+
+      it('existing subscriptions are sent upon peer connection', async function () {
+        this.timeout(10e3)
+
+        await Promise.all([
+          psA._libp2p.dial(psB.peerId),
+          new Promise((resolve) => psA.once('pubsub:subscription-change', resolve)),
+          new Promise((resolve) => psB.once('pubsub:subscription-change', resolve))
+        ])
+
+        expect(psA.peers.size).to.equal(1)
+        expect(psB.peers.size).to.equal(1)
+
+        expectSet(psA.subscriptions, ['Za'])
+        expectSet(psB.topics.get('Za'), [psA.peerId.toB58String()])
+
+        expectSet(psB.subscriptions, ['Zb'])
+        expectSet(psA.topics.get('Zb'), [psB.peerId.toB58String()])
+      })
+    })
+  })
+}

src/pubsub/tests/utils.js

@@ -0,0 +1,9 @@
+'use strict'
+
+const { expect } = require('chai')
+
+exports.first = (map) => map.values().next().value
+
+exports.expectSet = (set, subs) => {
+  expect(Array.from(set.values())).to.eql(subs)
+}

src/pubsub/utils.d.ts

@@ -0,0 +1,6 @@
+export function randomSeqno(): Uint8Array;
+export function msgId(from: string, seqno: Uint8Array): Uint8Array;
+export function anyMatch(a: any[] | Set<any>, b: any[] | Set<any>): boolean;
+export function ensureArray(maybeArray: any): any[];
+export function normalizeInRpcMessage(message: any, peerId: string): any;
+export function normalizeOutRpcMessage(message: any): any;

src/pubsub/utils.js

@@ -0,0 +1,106 @@
+'use strict'
+
+const randomBytes = require('libp2p-crypto/src/random-bytes')
+const uint8ArrayToString = require('uint8arrays/to-string')
+const uint8ArrayFromString = require('uint8arrays/from-string')
+const PeerId = require('peer-id')
+exports = module.exports
+
+/**
+ * Generatea random sequence number.
+ *
+ * @returns {Uint8Array}
+ * @private
+ */
+exports.randomSeqno = () => {
+  return randomBytes(8)
+}
+
+/**
+ * Generate a message id, based on the `from` and `seqno`.
+ *
+ * @param {string} from
+ * @param {Uint8Array} seqno
+ * @returns {Uint8Array}
+ * @private
+ */
+exports.msgId = (from, seqno) => {
+  const fromBytes = PeerId.createFromB58String(from).id
+  const msgId = new Uint8Array(fromBytes.length + seqno.length)
+  msgId.set(fromBytes, 0)
+  msgId.set(seqno, fromBytes.length)
+  return msgId
+}
+
+/**
+ * Check if any member of the first set is also a member
+ * of the second set.
+ *
+ * @param {Set|Array} a
+ * @param {Set|Array} b
+ * @returns {boolean}
+ * @private
+ */
+exports.anyMatch = (a, b) => {
+  let bHas
+  if (Array.isArray(b)) {
+    bHas = (val) => b.indexOf(val) > -1
+  } else {
+    bHas = (val) => b.has(val)
+  }
+
+  for (const val of a) {
+    if (bHas(val)) {
+      return true
+    }
+  }
+
+  return false
+}
+
+/**
+ * Make everything an array.
+ *
+ * @param {any} maybeArray
+ * @returns {Array}
+ * @private
+ */
+exports.ensureArray = (maybeArray) => {
+  if (!Array.isArray(maybeArray)) {
+    return [maybeArray]
+  }
+
+  return maybeArray
+}
+
+/**
+ * Ensures `message.from` is base58 encoded
+ * @param {object} message
+ * @param {String} peerId
+ * @return {object}
+ */
+exports.normalizeInRpcMessage = (message, peerId) => {
+  const m = Object.assign({}, message)
+  if (message.from instanceof Uint8Array) {
+    m.from = uint8ArrayToString(message.from, 'base58btc')
+  }
+  if (peerId) {
+    m.receivedFrom = peerId
+  }
+  return m
+}
+
+/**
+ * @param {object} message
+ * @return {object}
+ */
+exports.normalizeOutRpcMessage = (message) => {
+  const m = Object.assign({}, message)
+  if (typeof message.from === 'string' || message.from instanceof String) {
+    m.from = uint8ArrayFromString(message.from, 'base58btc')
+  }
+  if (typeof message.data === 'string' || message.data instanceof String) {
+    m.data = uint8ArrayFromString(message.data)
+  }
+  return m
+}

src/record/README.md

@@ -7,7 +7,7 @@ The record represents the data that will be stored inside the **envelope** when
 
 Taking into account that a record might be used in different contexts, an **envelope** signature made for a specific purpose **must not** be considered valid for a different purpose. Accordingly, each record has a short and descriptive string representing the record use case, known as **domain**. The data to be signed will be prepended with the domain string, in order to create a domain signature.
 
-A record can also contain a Buffer codec (ideally registered as a [multicodec](https://github.com/multiformats/multicodec)). This codec will prefix the record data in the **envelope** , so that it can be deserialized deterministically.
+A record can also contain a Uint8Array codec (ideally registered as a [multicodec](https://github.com/multiformats/multicodec)). This codec will prefix the record data in the **envelope** , so that it can be deserialized deterministically.
 
 ## Usage
 
@@ -30,10 +30,11 @@ describe('your record', () => {
 ```js
 const multicodec = require('multicodec')
 const Record = require('libp2p-interfaces/src/record')
+const fromString = require('uint8arrays/from-string')
 // const Protobuf = require('./record.proto')
 
 const ENVELOPE_DOMAIN_PEER_RECORD = 'libp2p-peer-record'
-const ENVELOPE_PAYLOAD_TYPE_PEER_RECORD = Buffer.from('0301', 'hex')
+const ENVELOPE_PAYLOAD_TYPE_PEER_RECORD = fromString('0301', 'hex')
 
 class PeerRecord extends Record {
   constructor (peerId, multiaddrs, seqNumber) {

src/record/index.d.ts

@@ -0,0 +1,23 @@
+export = Record;
+/**
+ * Record is the base implementation of a record that can be used as the payload of a libp2p envelope.
+ */
+declare class Record {
+    /**
+     * @constructor
+     * @param {String} domain signature domain
+     * @param {Uint8Array} codec identifier of the type of record
+     */
+    constructor(domain: string, codec: Uint8Array);
+    domain: string;
+    codec: Uint8Array;
+    /**
+     * Marshal a record to be used in an envelope.
+     */
+    marshal(): void;
+    /**
+     * Verifies if the other provided Record is identical to this one.
+     * @param {Record} other
+     */
+    equals(other: Record): void;
+}

src/record/index.js

@@ -9,7 +9,7 @@ class Record {
   /**
    * @constructor
    * @param {String} domain signature domain
-   * @param {Buffer} codec identifier of the type of record
+   * @param {Uint8Array} codec identifier of the type of record
    */
   constructor (domain, codec) {
     this.domain = domain

src/record/tests/index.js

@@ -24,7 +24,7 @@ module.exports = (test) => {
 
     it('is able to marshal', () => {
       const rawData = record.marshal()
-      expect(Buffer.isBuffer(rawData)).to.eql(true)
+      expect(rawData).to.be.an.instanceof(Uint8Array)
     })
 
     it('is able to compare two records', () => {

src/stream-muxer/tests/close-test.js

@@ -2,7 +2,6 @@
 /* eslint max-nested-callbacks: ["error", 8] */
 'use strict'
 
-const { Buffer } = require('buffer')
 const pair = require('it-pair/duplex')
 const pipe = require('it-pipe')
 const { consume } = require('streaming-iterables')
@@ -10,6 +9,7 @@ const Tcp = require('libp2p-tcp')
 const multiaddr = require('multiaddr')
 const abortable = require('abortable-iterator')
 const AbortController = require('abort-controller')
+const uint8arrayFromString = require('uint8arrays/from-string')
 
 const mh = multiaddr('/ip4/127.0.0.1/tcp/0')
 
@@ -18,7 +18,7 @@ function pause (ms) {
 }
 
 function randomBuffer () {
-  return Buffer.from(Math.random().toString())
+  return uint8arrayFromString(Math.random().toString())
 }
 
 const infiniteRandom = {

src/topology/index.d.ts

@@ -0,0 +1,42 @@
+declare const _exports: Topology;
+export = _exports;
+declare class Topology {
+    /**
+     * @param {Object} props
+     * @param {number} props.min minimum needed connections (default: 0)
+     * @param {number} props.max maximum needed connections (default: Infinity)
+     * @param {Object} [props.handlers]
+     * @param {function} [props.handlers.onConnect] protocol "onConnect" handler
+     * @param {function} [props.handlers.onDisconnect] protocol "onDisconnect" handler
+     * @constructor
+     */
+    constructor({ min, max, handlers }: {
+        min: number;
+        max: number;
+        handlers?: {
+            onConnect?: Function;
+            onDisconnect?: Function;
+        };
+    });
+    min: number;
+    max: number;
+    _onConnect: Function;
+    _onDisconnect: Function;
+    /**
+     * Set of peers that support the protocol.
+     * @type {Set<string>}
+     */
+    peers: Set<string>;
+    set registrar(arg: any);
+    _registrar: any;
+    /**
+     * @typedef PeerId
+     * @type {import('peer-id')}
+     */
+    /**
+     * Notify about peer disconnected event.
+     * @param {PeerId} peerId
+     * @returns {void}
+     */
+    disconnect(peerId: import("peer-id")): void;
+}

src/topology/index.js

@@ -1,7 +1,6 @@
 'use strict'
 
 const withIs = require('class-is')
-
 const noop = () => {}
 
 class Topology {
@@ -37,6 +36,11 @@ class Topology {
     this._registrar = registrar
   }
 
+  /**
+   * @typedef PeerId
+   * @type {import('peer-id')}
+   */
+
   /**
    * Notify about peer disconnected event.
    * @param {PeerId} peerId
@@ -47,4 +51,8 @@ class Topology {
   }
 }
 
+/**
+ * @module
+ * @type {Topology}
+ */
 module.exports = withIs(Topology, { className: 'Topology', symbolName: '@libp2p/js-interfaces/topology' })

src/topology/multicodec-topology.d.ts

@@ -0,0 +1,52 @@
+declare const _exports: MulticodecTopology;
+export = _exports;
+declare class MulticodecTopology {
+    /**
+     * @param {Object} props
+     * @param {number} props.min minimum needed connections (default: 0)
+     * @param {number} props.max maximum needed connections (default: Infinity)
+     * @param {Array<string>} props.multicodecs protocol multicodecs
+     * @param {Object} props.handlers
+     * @param {function} props.handlers.onConnect protocol "onConnect" handler
+     * @param {function} props.handlers.onDisconnect protocol "onDisconnect" handler
+     * @constructor
+     */
+    constructor({ min, max, multicodecs, handlers }: {
+        min: number;
+        max: number;
+        multicodecs: string[];
+        handlers: {
+            onConnect: Function;
+            onDisconnect: Function;
+        };
+    });
+    multicodecs: string[];
+    _registrar: any;
+    /**
+     * Check if a new peer support the multicodecs for this topology.
+     * @param {Object} props
+     * @param {PeerId} props.peerId
+     * @param {Array<string>} props.protocols
+     */
+    _onProtocolChange({ peerId, protocols }: {
+        peerId: any;
+        protocols: string[];
+    }): void;
+    /**
+     * Verify if a new connected peer has a topology multicodec and call _onConnect.
+     * @param {Connection} connection
+     * @returns {void}
+     */
+    _onPeerConnect(connection: any): void;
+    set registrar(arg: any);
+    /**
+     * Update topology.
+     * @param {Array<{id: PeerId, multiaddrs: Array<Multiaddr>, protocols: Array<string>}>} peerDataIterable
+     * @returns {void}
+     */
+    _updatePeers(peerDataIterable: {
+        id: any;
+        multiaddrs: any[];
+        protocols: string[];
+    }[]): void;
+}

src/topology/multicodec-topology.js

@@ -120,4 +120,8 @@ class MulticodecTopology extends Topology {
   }
 }
 
+/**
+ * @module
+ * @type {MulticodecTopology}
+ */
 module.exports = withIs(MulticodecTopology, { className: 'MulticodecTopology', symbolName: '@libp2p/js-interfaces/topology/multicodec-topology' })

src/transport/errors.d.ts

@@ -0,0 +1,6 @@
+export class AbortError extends Error {
+    static get code(): string;
+    static get type(): string;
+    code: string;
+    type: string;
+}

src/transport/tests/dial-test.js

@@ -31,8 +31,7 @@ module.exports = (common) => {
     }
   }
 
-  describe('dial', function () {
-    this.timeout(20 * 1000)
+  describe('dial', () => {
     let addrs
     let transport
     let connector
@@ -130,7 +129,7 @@ module.exports = (common) => {
       expect.fail('Did not throw error with code ' + AbortError.code)
     })
 
-    it.skip('abort while reading throws AbortError', async () => {
+    it('abort while reading throws AbortError', async () => {
       // Add a delay to the response from the server
       async function * delayedResponse (source) {
         for await (const val of source) {

src/transport/tests/listen-test.js

@@ -2,7 +2,6 @@
 /* eslint-env mocha */
 'use strict'
 
-const { Buffer } = require('buffer')
 const chai = require('chai')
 const dirtyChai = require('dirty-chai')
 const expect = chai.expect
@@ -11,6 +10,7 @@ const sinon = require('sinon')
 
 const pWaitFor = require('p-wait-for')
 const pipe = require('it-pipe')
+const uint8arrayFromString = require('uint8arrays/from-string')
 const { isValidTick } = require('./utils')
 
 module.exports = (common) => {
@@ -31,8 +31,7 @@ module.exports = (common) => {
     }
   }
 
-  describe('listen', function () {
-    this.timeout(20 * 1000)
+  describe('listen', () => {
     let addrs
     let transport
 
@@ -52,7 +51,7 @@ module.exports = (common) => {
       await listener.close()
     })
 
-    it.skip('close listener with connections, through timeout', async () => {
+    it('close listener with connections, through timeout', async () => {
       const upgradeSpy = sinon.spy(upgrader, 'upgradeInbound')
       const listenerConns = []
 
@@ -77,7 +76,7 @@ module.exports = (common) => {
       // Wait for the data send and close to finish
       await Promise.all([
         pipe(
-          [Buffer.from('Some data that is never handled')],
+          [uint8arrayFromString('Some data that is never handled')],
           socket1
         ),
         // Closer the listener (will take a couple of seconds to time out)
@@ -95,7 +94,7 @@ module.exports = (common) => {
       expect(upgradeSpy.callCount).to.equal(2)
     })
 
-    it.skip('should not handle connection if upgradeInbound throws', async () => {
+    it('should not handle connection if upgradeInbound throws', async () => {
       sinon.stub(upgrader, 'upgradeInbound').throws()
 
       const listener = transport.createListener(() => {

test/pubsub/emit-self.spec.js

@@ -0,0 +1,78 @@
+/* eslint-env mocha */
+'use strict'
+
+const { expect } = require('aegir/utils/chai')
+
+const {
+  createPeerId,
+  mockRegistrar,
+  PubsubImplementation
+} = require('./utils')
+
+const uint8ArrayFromString = require('uint8arrays/from-string')
+
+const protocol = '/pubsub/1.0.0'
+const topic = 'foo'
+const data = uint8ArrayFromString('bar')
+const shouldNotHappen = (_) => expect.fail()
+
+describe('emitSelf', () => {
+  let pubsub
+
+  describe('enabled', () => {
+    before(async () => {
+      const peerId = await createPeerId()
+
+      pubsub = new PubsubImplementation(protocol, {
+        peerId,
+        registrar: mockRegistrar
+      }, { emitSelf: true })
+    })
+
+    before(() => {
+      pubsub.start()
+      pubsub.subscribe(topic)
+    })
+
+    after(() => {
+      pubsub.stop()
+    })
+
+    it('should emit to self on publish', () => {
+      const promise = new Promise((resolve) => pubsub.once(topic, resolve))
+
+      pubsub.publish(topic, data)
+
+      return promise
+    })
+  })
+
+  describe('disabled', () => {
+    before(async () => {
+      const peerId = await createPeerId()
+
+      pubsub = new PubsubImplementation(protocol, {
+        peerId,
+        registrar: mockRegistrar
+      }, { emitSelf: false })
+    })
+
+    before(() => {
+      pubsub.start()
+      pubsub.subscribe(topic)
+    })
+
+    after(() => {
+      pubsub.stop()
+    })
+
+    it('should not emit to self on publish', () => {
+      pubsub.once(topic, (m) => shouldNotHappen)
+
+      pubsub.publish(topic, data)
+
+      // Wait 1 second to guarantee that self is not noticed
+      return new Promise((resolve) => setTimeout(() => resolve(), 1000))
+    })
+  })
+})

test/pubsub/instance.spec.js

@@ -0,0 +1,54 @@
+/* eslint-env mocha */
+'use strict'
+
+const { expect } = require('aegir/utils/chai')
+
+const PubsubBaseImpl = require('../../src/pubsub')
+const {
+  createPeerId,
+  mockRegistrar
+} = require('./utils')
+
+describe('pubsub instance', () => {
+  let peerId
+
+  before(async () => {
+    peerId = await createPeerId()
+  })
+
+  it('should throw if no debugName is provided', () => {
+    expect(() => {
+      new PubsubBaseImpl() // eslint-disable-line no-new
+    }).to.throw()
+  })
+
+  it('should throw if no multicodec is provided', () => {
+    expect(() => {
+      new PubsubBaseImpl({ // eslint-disable-line no-new
+        debugName: 'pubsub'
+      })
+    }).to.throw()
+  })
+
+  it('should throw if no libp2p is provided', () => {
+    expect(() => {
+      new PubsubBaseImpl({ // eslint-disable-line no-new
+        debugName: 'pubsub',
+        multicodecs: '/pubsub/1.0.0'
+      })
+    }).to.throw()
+  })
+
+  it('should accept valid parameters', () => {
+    expect(() => {
+      new PubsubBaseImpl({ // eslint-disable-line no-new
+        debugName: 'pubsub',
+        multicodecs: '/pubsub/1.0.0',
+        libp2p: {
+          peerId: peerId,
+          registrar: mockRegistrar
+        }
+      })
+    }).not.to.throw()
+  })
+})

test/pubsub/lifesycle.spec.js

@@ -0,0 +1,227 @@
+/* eslint-env mocha */
+'use strict'
+
+const { expect } = require('aegir/utils/chai')
+const sinon = require('sinon')
+
+const PubsubBaseImpl = require('../../src/pubsub')
+const {
+  createPeerId,
+  createMockRegistrar,
+  PubsubImplementation,
+  ConnectionPair
+} = require('./utils')
+
+describe('pubsub base lifecycle', () => {
+  describe('should start and stop properly', () => {
+    let pubsub
+    let sinonMockRegistrar
+
+    beforeEach(async () => {
+      const peerId = await createPeerId()
+      sinonMockRegistrar = {
+        handle: sinon.stub(),
+        register: sinon.stub(),
+        unregister: sinon.stub()
+      }
+
+      pubsub = new PubsubBaseImpl({
+        debugName: 'pubsub',
+        multicodecs: '/pubsub/1.0.0',
+        libp2p: {
+          peerId: peerId,
+          registrar: sinonMockRegistrar
+        }
+      })
+
+      expect(pubsub.peers.size).to.be.eql(0)
+    })
+
+    afterEach(() => {
+      sinon.restore()
+    })
+
+    it('should be able to start and stop', async () => {
+      await pubsub.start()
+      expect(sinonMockRegistrar.handle.calledOnce).to.be.true()
+      expect(sinonMockRegistrar.register.calledOnce).to.be.true()
+
+      await pubsub.stop()
+      expect(sinonMockRegistrar.unregister.calledOnce).to.be.true()
+    })
+
+    it('starting should not throw if already started', async () => {
+      await pubsub.start()
+      await pubsub.start()
+      expect(sinonMockRegistrar.handle.calledOnce).to.be.true()
+      expect(sinonMockRegistrar.register.calledOnce).to.be.true()
+
+      await pubsub.stop()
+      expect(sinonMockRegistrar.unregister.calledOnce).to.be.true()
+    })
+
+    it('stopping should not throw if not started', async () => {
+      await pubsub.stop()
+      expect(sinonMockRegistrar.register.calledOnce).to.be.false()
+      expect(sinonMockRegistrar.unregister.calledOnce).to.be.false()
+    })
+  })
+
+  describe('should be able to register two nodes', () => {
+    const protocol = '/pubsub/1.0.0'
+    let pubsubA, pubsubB
+    let peerIdA, peerIdB
+    const registrarRecordA = {}
+    const registrarRecordB = {}
+
+    // mount pubsub
+    beforeEach(async () => {
+      peerIdA = await createPeerId()
+      peerIdB = await createPeerId()
+
+      pubsubA = new PubsubImplementation(protocol, {
+        peerId: peerIdA,
+        registrar: createMockRegistrar(registrarRecordA)
+      })
+      pubsubB = new PubsubImplementation(protocol, {
+        peerId: peerIdB,
+        registrar: createMockRegistrar(registrarRecordB)
+      })
+    })
+
+    // start pubsub
+    beforeEach(() => {
+      pubsubA.start()
+      pubsubB.start()
+
+      expect(Object.keys(registrarRecordA)).to.have.lengthOf(1)
+      expect(Object.keys(registrarRecordB)).to.have.lengthOf(1)
+    })
+
+    afterEach(() => {
+      sinon.restore()
+
+      return Promise.all([
+        pubsubA.stop(),
+        pubsubB.stop()
+      ])
+    })
+
+    it('should handle onConnect as expected', async () => {
+      const onConnectA = registrarRecordA[protocol].onConnect
+      const handlerB = registrarRecordB[protocol].handler
+
+      // Notice peers of connection
+      const [c0, c1] = ConnectionPair()
+
+      await onConnectA(peerIdB, c0)
+      await handlerB({
+        protocol,
+        stream: c1.stream,
+        connection: {
+          remotePeer: peerIdA
+        }
+      })
+
+      expect(pubsubA.peers.size).to.be.eql(1)
+      expect(pubsubB.peers.size).to.be.eql(1)
+    })
+
+    it('should use the latest connection if onConnect is called more than once', async () => {
+      const onConnectA = registrarRecordA[protocol].onConnect
+      const handlerB = registrarRecordB[protocol].handler
+
+      // Notice peers of connection
+      const [c0, c1] = ConnectionPair()
+      const [c2] = ConnectionPair()
+
+      sinon.spy(c0, 'newStream')
+
+      await onConnectA(peerIdB, c0)
+      await handlerB({
+        protocol,
+        stream: c1.stream,
+        connection: {
+          remotePeer: peerIdA
+        }
+      })
+      expect(c0.newStream).to.have.property('callCount', 1)
+
+      sinon.spy(pubsubA, '_removePeer')
+
+      sinon.spy(c2, 'newStream')
+
+      await onConnectA(peerIdB, c2)
+      expect(c2.newStream).to.have.property('callCount', 1)
+      expect(pubsubA._removePeer).to.have.property('callCount', 0)
+
+      // Verify the first stream was closed
+      const { stream: firstStream } = await c0.newStream.returnValues[0]
+      try {
+        await firstStream.sink(['test'])
+      } catch (err) {
+        expect(err).to.exist()
+        return
+      }
+      expect.fail('original stream should have ended')
+    })
+
+    it('should handle newStream errors in onConnect', async () => {
+      const onConnectA = registrarRecordA[protocol].onConnect
+      const handlerB = registrarRecordB[protocol].handler
+
+      // Notice peers of connection
+      const [c0, c1] = ConnectionPair()
+      const error = new Error('new stream error')
+      sinon.stub(c0, 'newStream').throws(error)
+
+      await onConnectA(peerIdB, c0)
+      await handlerB({
+        protocol,
+        stream: c1.stream,
+        connection: {
+          remotePeer: peerIdA
+        }
+      })
+
+      expect(c0.newStream).to.have.property('callCount', 1)
+    })
+
+    it('should handle onDisconnect as expected', async () => {
+      const onConnectA = registrarRecordA[protocol].onConnect
+      const onDisconnectA = registrarRecordA[protocol].onDisconnect
+      const handlerB = registrarRecordB[protocol].handler
+      const onDisconnectB = registrarRecordB[protocol].onDisconnect
+
+      // Notice peers of connection
+      const [c0, c1] = ConnectionPair()
+
+      await onConnectA(peerIdB, c0)
+      await handlerB({
+        protocol,
+        stream: c1.stream,
+        connection: {
+          remotePeer: peerIdA
+        }
+      })
+
+      // Notice peers of disconnect
+      onDisconnectA(peerIdB)
+      onDisconnectB(peerIdA)
+
+      expect(pubsubA.peers.size).to.be.eql(0)
+      expect(pubsubB.peers.size).to.be.eql(0)
+    })
+
+    it('should handle onDisconnect for unknown peers', () => {
+      const onDisconnectA = registrarRecordA[protocol].onDisconnect
+
+      expect(pubsubA.peers.size).to.be.eql(0)
+
+      // Notice peers of disconnect
+      onDisconnectA(peerIdB)
+
+      expect(pubsubA.peers.size).to.be.eql(0)
+    })
+  })
+})

test/pubsub/message.spec.js

@@ -0,0 +1,73 @@
+/* eslint-env mocha */
+'use strict'
+
+const { expect } = require('aegir/utils/chai')
+const sinon = require('sinon')
+
+const PubsubBaseImpl = require('../../src/pubsub')
+const { randomSeqno } = require('../../src/pubsub/utils')
+const {
+  createPeerId,
+  mockRegistrar
+} = require('./utils')
+
+describe('pubsub base messages', () => {
+  let peerId
+  let pubsub
+
+  before(async () => {
+    peerId = await createPeerId()
+    pubsub = new PubsubBaseImpl({
+      debugName: 'pubsub',
+      multicodecs: '/pubsub/1.0.0',
+      libp2p: {
+        peerId: peerId,
+        registrar: mockRegistrar
+      }
+    })
+  })
+
+  afterEach(() => {
+    sinon.restore()
+  })
+
+  it('_buildMessage normalizes and signs messages', async () => {
+    const message = {
+      receivedFrom: peerId.id,
+      from: peerId.id,
+      data: 'hello',
+      seqno: randomSeqno(),
+      topicIDs: ['test-topic']
+    }
+
+    const signedMessage = await pubsub._buildMessage(message)
+    expect(pubsub.validate(signedMessage)).to.not.be.rejected()
+  })
+
+  it('validate with strict signing off will validate a present signature', async () => {
+    const message = {
+      receivedFrom: peerId.id,
+      from: peerId.id,
+      data: 'hello',
+      seqno: randomSeqno(),
+      topicIDs: ['test-topic']
+    }
+
+    sinon.stub(pubsub, 'strictSigning').value(false)
+
+    const signedMessage = await pubsub._buildMessage(message)
+    expect(pubsub.validate(signedMessage)).to.not.be.rejected()
+  })
+
+  it('validate with strict signing requires a signature', async () => {
+    const message = {
+      receivedFrom: peerId.id,
+      from: peerId.id,
+      data: 'hello',
+      seqno: randomSeqno(),
+      topicIDs: ['test-topic']
+    }
+
+    await expect(pubsub.validate(message)).to.be.rejectedWith(Error, 'Signing required and no signature was present')
+  })
+})

test/pubsub/pubsub.spec.js

@@ -0,0 +1,358 @@
+/* eslint-env mocha */
+/* eslint max-nested-callbacks: ["error", 6] */
+'use strict'
+
+const { expect } = require('aegir/utils/chai')
+const sinon = require('sinon')
+const pWaitFor = require('p-wait-for')
+
+const uint8ArrayFromString = require('uint8arrays/from-string')
+
+const PeerStreams = require('../../src/pubsub/peer-streams')
+const {
+  createPeerId,
+  createMockRegistrar,
+  ConnectionPair,
+  mockRegistrar,
+  PubsubImplementation
+} = require('./utils')
+
+const protocol = '/pubsub/1.0.0'
+const topic = 'test-topic'
+const message = uint8ArrayFromString('hello')
+
+describe('pubsub base implementation', () => {
+  describe('publish', () => {
+    let pubsub
+
+    beforeEach(async () => {
+      const peerId = await createPeerId()
+      pubsub = new PubsubImplementation(protocol, {
+        peerId: peerId,
+        registrar: mockRegistrar
+      })
+    })
+
+    afterEach(() => pubsub.stop())
+
+    it('calls _publish for router to forward messages', async () => {
+      sinon.spy(pubsub, '_publish')
+
+      pubsub.start()
+      await pubsub.publish(topic, message)
+
+      expect(pubsub._publish.callCount).to.eql(1)
+    })
+
+    it('should sign messages on publish', async () => {
+      sinon.spy(pubsub, '_publish')
+
+      pubsub.start()
+      await pubsub.publish(topic, message)
+
+      // Get the first message sent to _publish, and validate it
+      const signedMessage = pubsub._publish.getCall(0).lastArg
+      try {
+        await pubsub.validate(signedMessage)
+      } catch (e) {
+        expect.fail('validation should not throw')
+      }
+    })
+  })
+
+  describe('subscribe', () => {
+    describe('basics', () => {
+      let pubsub
+
+      beforeEach(async () => {
+        const peerId = await createPeerId()
+        pubsub = new PubsubImplementation(protocol, {
+          peerId: peerId,
+          registrar: mockRegistrar
+        })
+        pubsub.start()
+      })
+
+      afterEach(() => pubsub.stop())
+
+      it('should add subscription', () => {
+        pubsub.subscribe(topic)
+
+        expect(pubsub.subscriptions.size).to.eql(1)
+        expect(pubsub.subscriptions.has(topic)).to.be.true()
+      })
+    })
+
+    describe('two nodes', () => {
+      let pubsubA, pubsubB
+      let peerIdA, peerIdB
+      const registrarRecordA = {}
+      const registrarRecordB = {}
+
+      beforeEach(async () => {
+        peerIdA = await createPeerId()
+        peerIdB = await createPeerId()
+
+        pubsubA = new PubsubImplementation(protocol, {
+          peerId: peerIdA,
+          registrar: createMockRegistrar(registrarRecordA)
+        })
+        pubsubB = new PubsubImplementation(protocol, {
+          peerId: peerIdB,
+          registrar: createMockRegistrar(registrarRecordB)
+        })
+      })
+
+      // start pubsub and connect nodes
+      beforeEach(async () => {
+        pubsubA.start()
+        pubsubB.start()
+
+        const onConnectA = registrarRecordA[protocol].onConnect
+        const handlerB = registrarRecordB[protocol].handler
+
+        // Notice peers of connection
+        const [c0, c1] = ConnectionPair()
+
+        await onConnectA(peerIdB, c0)
+        await handlerB({
+          protocol,
+          stream: c1.stream,
+          connection: {
+            remotePeer: peerIdA
+          }
+        })
+      })
+
+      afterEach(() => {
+        pubsubA.stop()
+        pubsubB.stop()
+      })
+
+      it('should send subscribe message to connected peers', async () => {
+        sinon.spy(pubsubA, '_sendSubscriptions')
+        sinon.spy(pubsubB, '_processRpcSubOpt')
+
+        pubsubA.subscribe(topic)
+
+        // Should send subscriptions to a peer
+        expect(pubsubA._sendSubscriptions.callCount).to.eql(1)
+
+        // Other peer should receive subscription message
+        await pWaitFor(() => {
+          const subscribers = pubsubB.getSubscribers(topic)
+
+          return subscribers.length === 1
+        })
+        expect(pubsubB._processRpcSubOpt.callCount).to.eql(1)
+      })
+    })
+  })
+
+  describe('unsubscribe', () => {
+    describe('basics', () => {
+      let pubsub
+
+      beforeEach(async () => {
+        const peerId = await createPeerId()
+        pubsub = new PubsubImplementation(protocol, {
+          peerId: peerId,
+          registrar: mockRegistrar
+        })
+        pubsub.start()
+      })
+
+      afterEach(() => pubsub.stop())
+
+      it('should remove all subscriptions for a topic', () => {
+        pubsub.subscribe(topic, (msg) => {})
+        pubsub.subscribe(topic, (msg) => {})
+
+        expect(pubsub.subscriptions.size).to.eql(1)
+
+        pubsub.unsubscribe(topic)
+
+        expect(pubsub.subscriptions.size).to.eql(0)
+      })
+    })
+
+    describe('two nodes', () => {
+      let pubsubA, pubsubB
+      let peerIdA, peerIdB
+      const registrarRecordA = {}
+      const registrarRecordB = {}
+
+      beforeEach(async () => {
+        peerIdA = await createPeerId()
+        peerIdB = await createPeerId()
+
+        pubsubA = new PubsubImplementation(protocol, {
+          peerId: peerIdA,
+          registrar: createMockRegistrar(registrarRecordA)
+        })
+        pubsubB = new PubsubImplementation(protocol, {
+          peerId: peerIdB,
+          registrar: createMockRegistrar(registrarRecordB)
+        })
+      })
+
+      // start pubsub and connect nodes
+      beforeEach(async () => {
+        pubsubA.start()
+        pubsubB.start()
+
+        const onConnectA = registrarRecordA[protocol].onConnect
+        const handlerB = registrarRecordB[protocol].handler
+
+        // Notice peers of connection
+        const [c0, c1] = ConnectionPair()
+
+        await onConnectA(peerIdB, c0)
+        await handlerB({
+          protocol,
+          stream: c1.stream,
+          connection: {
+            remotePeer: peerIdA
+          }
+        })
+      })
+
+      afterEach(() => {
+        pubsubA.stop()
+        pubsubB.stop()
+      })
+
+      it('should send unsubscribe message to connected peers', async () => {
+        sinon.spy(pubsubA, '_sendSubscriptions')
+        sinon.spy(pubsubB, '_processRpcSubOpt')
+
+        pubsubA.subscribe(topic)
+        // Should send subscriptions to a peer
+        expect(pubsubA._sendSubscriptions.callCount).to.eql(1)
+
+        // Other peer should receive subscription message
+        await pWaitFor(() => {
+          const subscribers = pubsubB.getSubscribers(topic)
+
+          return subscribers.length === 1
+        })
+        expect(pubsubB._processRpcSubOpt.callCount).to.eql(1)
+
+        // Unsubscribe
+        pubsubA.unsubscribe(topic)
+        // Should send subscriptions to a peer
+        expect(pubsubA._sendSubscriptions.callCount).to.eql(2)
+
+        // Other peer should receive subscription message
+        await pWaitFor(() => {
+          const subscribers = pubsubB.getSubscribers(topic)
+
+          return subscribers.length === 0
+        })
+        expect(pubsubB._processRpcSubOpt.callCount).to.eql(2)
+      })
+
+      it('should not send unsubscribe message to connected peers if not subscribed', () => {
+        sinon.spy(pubsubA, '_sendSubscriptions')
+        sinon.spy(pubsubB, '_processRpcSubOpt')
+
+        // Unsubscribe
+        pubsubA.unsubscribe(topic)
+
+        // Should send subscriptions to a peer
+        expect(pubsubA._sendSubscriptions.callCount).to.eql(0)
+      })
+    })
+  })
+
+  describe('getTopics', () => {
+    let peerId
+    let pubsub
+
+    beforeEach(async () => {
+      peerId = await createPeerId()
+      pubsub = new PubsubImplementation(protocol, {
+        peerId: peerId,
+        registrar: mockRegistrar
+      })
+      pubsub.start()
+    })
+
+    afterEach(() => pubsub.stop())
+
+    it('returns the subscribed topics', () => {
+      let subsTopics = pubsub.getTopics()
+      expect(subsTopics).to.have.lengthOf(0)
+
+      pubsub.subscribe(topic)
+
+      subsTopics = pubsub.getTopics()
+      expect(subsTopics).to.have.lengthOf(1)
+      expect(subsTopics[0]).to.eql(topic)
+    })
+  })
+
+  describe('getSubscribers', () => {
+    let peerId
+    let pubsub
+
+    beforeEach(async () => {
+      peerId = await createPeerId()
+      pubsub = new PubsubImplementation(protocol, {
+        peerId: peerId,
+        registrar: mockRegistrar
+      })
+    })
+
+    afterEach(() => pubsub.stop())
+
+    it('should fail if pubsub is not started', () => {
+      const topic = 'topic-test'
+
+      try {
+        pubsub.getSubscribers(topic)
+      } catch (err) {
+        expect(err).to.exist()
+        expect(err.code).to.eql('ERR_NOT_STARTED_YET')
+        return
+      }
+      throw new Error('should fail if pubsub is not started')
+    })
+
+    it('should fail if no topic is provided', () => {
+      // start pubsub
+      pubsub.start()
+
+      try {
+        pubsub.getSubscribers()
+      } catch (err) {
+        expect(err).to.exist()
+        expect(err.code).to.eql('ERR_NOT_VALID_TOPIC')
+        return
+      }
+      throw new Error('should fail if no topic is provided')
+    })
+
+    it('should get peer subscribed to one topic', () => {
+      const topic = 'topic-test'
+
+      // start pubsub
+      pubsub.start()
+
+      let peersSubscribed = pubsub.getSubscribers(topic)
+      expect(peersSubscribed).to.be.empty()
+
+      // Set mock peer subscribed
+      const peer = new PeerStreams({ id: peerId })
+      const id = peer.id.toB58String()
+
+      pubsub.topics.set(topic, new Set([id]))
+      pubsub.peers.set(id, peer)
+
+      peersSubscribed = pubsub.getSubscribers(topic)
+
+      expect(peersSubscribed).to.not.be.empty()
+      expect(peersSubscribed[0]).to.eql(id)
+    })
+  })
+})

test/pubsub/sign.spec.js

@@ -0,0 +1,93 @@
+/* eslint-env mocha */
+/* eslint max-nested-callbacks: ["error", 5] */
+'use strict'
+
+const { expect } = require('aegir/utils/chai')
+const uint8ArrayConcat = require('uint8arrays/concat')
+const uint8ArrayFromString = require('uint8arrays/from-string')
+
+const { Message } = require('../../src/pubsub/message')
+const {
+  signMessage,
+  SignPrefix,
+  verifySignature
+} = require('../../src/pubsub/message/sign')
+const PeerId = require('peer-id')
+const { randomSeqno } = require('../../src/pubsub/utils')
+
+describe('message signing', () => {
+  let peerId
+  before(async () => {
+    peerId = await PeerId.create({
+      bits: 1024
+    })
+  })
+
+  it('should be able to sign and verify a message', async () => {
+    const message = {
+      from: peerId.id,
+      data: uint8ArrayFromString('hello'),
+      seqno: randomSeqno(),
+      topicIDs: ['test-topic']
+    }
+
+    const bytesToSign = uint8ArrayConcat([SignPrefix, Message.encode(message)])
+    const expectedSignature = await peerId.privKey.sign(bytesToSign)
+
+    const signedMessage = await signMessage(peerId, message)
+
+    // Check the signature and public key
+    expect(signedMessage.signature).to.eql(expectedSignature)
+    expect(signedMessage.key).to.eql(peerId.pubKey.bytes)
+
+    // Verify the signature
+    const verified = await verifySignature(signedMessage)
+    expect(verified).to.eql(true)
+  })
+
+  it('should be able to extract the public key from an inlined key', async () => {
+    const secPeerId = await PeerId.create({ keyType: 'secp256k1' })
+
+    const message = {
+      from: secPeerId.id,
+      data: uint8ArrayFromString('hello'),
+      seqno: randomSeqno(),
+      topicIDs: ['test-topic']
+    }
+
+    const bytesToSign = uint8ArrayConcat([SignPrefix, Message.encode(message)])
+    const expectedSignature = await secPeerId.privKey.sign(bytesToSign)
+
+    const signedMessage = await signMessage(secPeerId, message)
+
+    // Check the signature and public key
+    expect(signedMessage.signature).to.eql(expectedSignature)
+    signedMessage.key = undefined
+
+    // Verify the signature
+    const verified = await verifySignature(signedMessage)
+    expect(verified).to.eql(true)
+  })
+
+  it('should be able to extract the public key from the message', async () => {
+    const message = {
+      from: peerId.id,
+      data: uint8ArrayFromString('hello'),
+      seqno: randomSeqno(),
+      topicIDs: ['test-topic']
+    }
+
+    const bytesToSign = uint8ArrayConcat([SignPrefix, Message.encode(message)])
+    const expectedSignature = await peerId.privKey.sign(bytesToSign)
+
+    const signedMessage = await signMessage(peerId, message)
+
+    // Check the signature and public key
+    expect(signedMessage.signature).to.eql(expectedSignature)
+    expect(signedMessage.key).to.eql(peerId.pubKey.bytes)
+
+    // Verify the signature
+    const verified = await verifySignature(signedMessage)
+    expect(verified).to.eql(true)
+  })
+})

test/pubsub/topic-validators.spec.js

@@ -0,0 +1,110 @@
+/* eslint-env mocha */
+'use strict'
+
+const { expect } = require('aegir/utils/chai')
+const sinon = require('sinon')
+const pWaitFor = require('p-wait-for')
+const errCode = require('err-code')
+
+const PeerId = require('peer-id')
+const uint8ArrayEquals = require('uint8arrays/equals')
+const uint8ArrayFromString = require('uint8arrays/from-string')
+
+const { utils } = require('../../src/pubsub')
+const PeerStreams = require('../../src/pubsub/peer-streams')
+
+const {
+  createPeerId,
+  mockRegistrar,
+  PubsubImplementation
+} = require('./utils')
+
+const protocol = '/pubsub/1.0.0'
+
+describe('topic validators', () => {
+  let pubsub
+
+  beforeEach(async () => {
+    const peerId = await createPeerId()
+
+    pubsub = new PubsubImplementation(protocol, {
+      peerId: peerId,
+      registrar: mockRegistrar
+    })
+
+    pubsub.start()
+  })
+
+  afterEach(() => {
+    sinon.restore()
+  })
+
+  it('should filter messages by topic validator', async () => {
+    // use _publish.callCount() to see if a message is valid or not
+    sinon.spy(pubsub, '_publish')
+    // Disable strict signing
+    sinon.stub(pubsub, 'strictSigning').value(false)
+    sinon.stub(pubsub.peers, 'get').returns({})
+    const filteredTopic = 't'
+    const peer = new PeerStreams({ id: await PeerId.create() })
+
+    // Set a trivial topic validator
+    pubsub.topicValidators.set(filteredTopic, (topic, message) => {
+      if (!uint8ArrayEquals(message.data, uint8ArrayFromString('a message'))) {
+        throw errCode(new Error(), 'ERR_TOPIC_VALIDATOR_REJECT')
+      }
+    })
+
+    // valid case
+    const validRpc = {
+      subscriptions: [],
+      msgs: [{
+        from: peer.id.toBytes(),
+        data: uint8ArrayFromString('a message'),
+        seqno: utils.randomSeqno(),
+        topicIDs: [filteredTopic]
+      }]
+    }
+
+    // process valid message
+    pubsub.subscribe(filteredTopic)
+    pubsub._processRpc(peer.id.toB58String(), peer, validRpc)
+
+    await pWaitFor(() => pubsub._publish.callCount === 1)
+
+    // invalid case
+    const invalidRpc = {
+      subscriptions: [],
+      msgs: [{
+        from: peer.id.toBytes(),
+        data: uint8ArrayFromString('a different message'),
+        seqno: utils.randomSeqno(),
+        topicIDs: [filteredTopic]
+      }]
+    }
+
+    // process invalid message
+    pubsub._processRpc(peer.id.toB58String(), peer, invalidRpc)
+    expect(pubsub._publish.callCount).to.eql(1)
+
+    // remove topic validator
+    pubsub.topicValidators.delete(filteredTopic)
+
+    // another invalid case
+    const invalidRpc2 = {
+      subscriptions: [],
+      msgs: [{
+        from: peer.id.toB58String(),
+        data: uint8ArrayFromString('a different message'),
+        seqno: utils.randomSeqno(),
+        topicIDs: [filteredTopic]
+      }]
+    }
+
+    // process previously invalid message, now is valid
+    pubsub._processRpc(peer.id.toB58String(), peer, invalidRpc2)
+    pubsub.unsubscribe(filteredTopic)
+
+    await pWaitFor(() => pubsub._publish.callCount === 2)
+  })
+})

test/pubsub/utils.spec.js

@@ -0,0 +1,78 @@
+/* eslint-env mocha */
+'use strict'
+
+const { expect } = require('aegir/utils/chai')
+const utils = require('../../src/pubsub/utils')
+const uint8ArrayFromString = require('uint8arrays/from-string')
+
+describe('utils', () => {
+  it('randomSeqno', () => {
+    const first = utils.randomSeqno()
+    const second = utils.randomSeqno()
+
+    expect(first).to.have.length(8)
+    expect(second).to.have.length(8)
+    expect(first).to.not.eql(second)
+  })
+
+  it('msgId should not generate same ID for two different Uint8Arrays', () => {
+    const peerId = 'QmPNdSYk5Rfpo5euNqwtyizzmKXMNHdXeLjTQhcN4yfX22'
+    const msgId0 = utils.msgId(peerId, uint8ArrayFromString('15603533e990dfde', 'base16'))
+    const msgId1 = utils.msgId(peerId, uint8ArrayFromString('15603533e990dfe0', 'base16'))
+    expect(msgId0).to.not.deep.equal(msgId1)
+  })
+
+  it('anyMatch', () => {
+    [
+      [[1, 2, 3], [4, 5, 6], false],
+      [[1, 2], [1, 2], true],
+      [[1, 2, 3], [4, 5, 1], true],
+      [[5, 6, 1], [1, 2, 3], true],
+      [[], [], false],
+      [[1], [2], false]
+    ].forEach((test) => {
+      expect(utils.anyMatch(new Set(test[0]), new Set(test[1])))
+        .to.eql(test[2])
+
+      expect(utils.anyMatch(new Set(test[0]), test[1]))
+        .to.eql(test[2])
+    })
+  })
+
+  it('ensureArray', () => {
+    expect(utils.ensureArray('hello')).to.be.eql(['hello'])
+    expect(utils.ensureArray([1, 2])).to.be.eql([1, 2])
+  })
+
+  it('converts an IN msg.from to b58', () => {
+    const binaryId = uint8ArrayFromString('1220e2187eb3e6c4fb3e7ff9ad4658610624a6315e0240fc6f37130eedb661e939cc', 'base16')
+    const stringId = 'QmdZEWgtaWAxBh93fELFT298La1rsZfhiC2pqwMVwy3jZM'
+    const m = [
+      { from: binaryId },
+      { from: stringId }
+    ]
+    const expected = [
+      { from: stringId },
+      { from: stringId }
+    ]
+    for (let i = 0; i < m.length; i++) {
+      expect(utils.normalizeInRpcMessage(m[i])).to.deep.eql(expected[i])
+    }
+  })
+
+  it('converts an OUT msg.from to binary', () => {
+    const binaryId = uint8ArrayFromString('1220e2187eb3e6c4fb3e7ff9ad4658610624a6315e0240fc6f37130eedb661e939cc', 'base16')
+    const stringId = 'QmdZEWgtaWAxBh93fELFT298La1rsZfhiC2pqwMVwy3jZM'
+    const m = [
+      { from: binaryId },
+      { from: stringId }
+    ]
+    const expected = [
+      { from: binaryId },
+      { from: binaryId }
+    ]
+    for (let i = 0; i < m.length; i++) {
+      expect(utils.normalizeOutRpcMessage(m[i])).to.deep.eql(expected[i])
+    }
+  })
+})

test/pubsub/utils/index.js

@@ -0,0 +1,85 @@
+'use strict'
+
+const DuplexPair = require('it-pair/duplex')
+
+const PeerId = require('peer-id')
+
+const PubsubBaseProtocol = require('../../../src/pubsub')
+const { message } = require('../../../src/pubsub')
+
+exports.createPeerId = async () => {
+  const peerId = await PeerId.create({ bits: 1024 })
+
+  return peerId
+}
+
+class PubsubImplementation extends PubsubBaseProtocol {
+  constructor (protocol, libp2p, options = {}) {
+    super({
+      debugName: 'libp2p:pubsub',
+      multicodecs: protocol,
+      libp2p,
+      ...options
+    })
+  }
+
+  _publish (message) {
+    // ...
+  }
+
+  _decodeRpc (bytes) {
+    return message.rpc.RPC.decode(bytes)
+  }
+
+  _encodeRpc (rpc) {
+    return message.rpc.RPC.encode(rpc)
+  }
+}
+
+exports.PubsubImplementation = PubsubImplementation
+
+exports.mockRegistrar = {
+  handle: () => {},
+  register: () => {},
+  unregister: () => {}
+}
+
+exports.createMockRegistrar = (registrarRecord) => ({
+  handle: (multicodecs, handler) => {
+    const rec = registrarRecord[multicodecs[0]] || {}
+
+    registrarRecord[multicodecs[0]] = {
+      ...rec,
+      handler
+    }
+  },
+  register: ({ multicodecs, _onConnect, _onDisconnect }) => {
+    const rec = registrarRecord[multicodecs[0]] || {}
+
+    registrarRecord[multicodecs[0]] = {
+      ...rec,
+      onConnect: _onConnect,
+      onDisconnect: _onDisconnect
+    }
+
+    return multicodecs[0]
+  },
+  unregister: (id) => {
+    delete registrarRecord[id]
+  }
+})
+
+exports.ConnectionPair = () => {
+  const [d0, d1] = DuplexPair()
+
+  return [
+    {
+      stream: d0,
+      newStream: () => Promise.resolve({ stream: d0 })
+    },
+    {
+      stream: d1,
+      newStream: () => Promise.resolve({ stream: d1 })
+    }
+  ]
+}

tsconfig.json

@@ -0,0 +1,18 @@
+{
+  "include": ["src/**/*.js"],
+  "exclude": ["src/**/tests/*", "src/utils"],
+
+  "compilerOptions": {
+    // Tells TypeScript to read JS files, as
+    // normally they are ignored as source files
+    "allowJs": true,
+    // Generate d.ts files
+    "declaration": true,
+    // This compiler run should
+    // only output d.ts files
+    "emitDeclarationOnly": true,
+    "esModuleInterop": true,
+    "rootDir": "./src",
+    "outDir": "./src"
+  }
+}