p2psub

Mesh peer-to-peer JSON Pub/Sub with no external dependencies. Each peer can act as a server and client, just a client, or a simple relay peer. Topics can be subscribed to using a simple string or regex pattern.

Features

• Mesh peer-to-peer network forwards messages to all connected peers, even if they are not directly connected
• Peers can be added and removed on the fly
• PubSub topics can be subscribed using RegExp patterns
• publish() returns a Promise and supports async listeners — await p2p.publish(...)
• Subscribe callbacks receive the originating peer, so misbehaving peers can be identified and ignored
• subscribe() returns an unsubscribe handle for easy cleanup
• Peer-to-peer and pub/sub are two separate importable classes for maximum customization
• Zero runtime dependencies

Installation

npm install p2psub

Requires Node.js >= 18.0.0 (the test suite uses the built-in node:test runner).

TypeScript Support

p2psub ships with TypeScript declarations. Import the classes directly:

import {P2PSub, P2P, PubSub, ps} from 'p2psub';

const p2p = new P2PSub({
    listenPort: 7575,
    peers: ['10.10.10.11:7575']
});

Usage

Standalone PubSub (No P2P Networking)

For simple in-process pub/sub without networking, you can use the PubSub class directly. A singleton instance ps is provided for convenience:

const {ps} = require('p2psub');

// Subscribe to topics. subscribe() returns an unsubscribe handle.
const unsub = ps.subscribe('my-topic', (data, topic) => {
	console.log('Received:', topic, data);
});

// Publish to topics. publish() returns a Promise that resolves once all
// matching listeners (including async ones) have settled.
ps.publish('my-topic', {message: 'Hello World'});

// Later, stop listening:
unsub();

Listeners may be async — publish() awaits any Promise they return:

const {PubSub} = require('p2psub');

const myPubSub = new PubSub();

myPubSub.subscribe('topic', async (data) => {
	await doAsyncWork(data);
});

// Resolves only after the async listener above has finished.
await myPubSub.publish('topic', data);

Alternatively, create your own isolated PubSub instance:

const {PubSub} = require('p2psub');

const myPubSub = new PubSub();
myPubSub.subscribe('topic', callback);
myPubSub.publish('topic', data);

P2PSub (Combined P2P + PubSub)

Instantiate a P2PSub instance:

const {P2PSub} = require('p2psub');

const p2p = new P2PSub({
	listenPort: 7575,
	peers:[
		'10.10.10.11:7575',
		'10.10.10.12:7575',
		'172.16.24.2:8637'
	]
});

Now we can listen for and publish topics across the whole network:

// Local peer

p2p.publish('announcement', {message: 'p2p pubsub is awesome!'});

// Remote peer

p2p.subscribe('announcement', (data)=> console.log(data.message));

# p2p pubsub is awesome!

We can also use regex patterns in our subscribes to catch more, or all topics:

// Local peer

p2p.publish('announcement', {message: 'p2p pubsub is awesome!'});
p2p.publish('announcement-group1', {message: 'Mesh is the future!'});
p2p.publish('resource-block-added', {id: '123', name:'block0'});

// Remote peer

p2p.subscribe(/^announcement/, (data, topic)=> console.log(topic, data.message));

# announcement p2p pubsub is awesome!
# announcement-group1 Mesh is the future!

// Another remote peer

p2p.subscribe(/./, (data, topic)=> console.log(topic, data));

# announcement {message: 'p2p pubsub is awesome!'}
# announcement-group1 {message: 'Mesh is the future!'}
# resource-block-added {id: '123', name:'block0'}

Identifying the Originating Peer

When used through P2PSub, subscribe callbacks receive a third argument — the originating peer's ID — so you can attribute messages and ignore or kick misbehaving peers:

const p2p = new P2PSub({listenPort: 7575, peers: ['10.10.10.11:7575']});

const badPeers = new Set();

p2p.subscribe(/.*/, (data, topic, from) => {
	if(badPeers.has(from)) return; // ignore messages from banned peers

	console.log(`message on '${topic}' from peer ${from}`);
});

// Ban a flooding peer and drop its connection:
// p2p.removePeer('10.10.10.99:7575');

Locally published messages are attributed to the local peer (p2p.p2p.peerID), so a subscriber can tell local and remote messages apart by comparing from against the local peer ID.

P2PSub Instance Options

All options are provided by and passed to the P2P class. The PubSub class takes no instance options.

listenPort (Number or String, optional)

• Sets the incoming TCP port for the local peer to listen on
• If omitted, the local peer will not accept incoming connections
• Default: undefined

peers (Array of Strings, optional)

• List of peers this peer will attempt to connect with
• Peers should be specified as {host}:{port}
• The host can be an IP address or a hostname the system can resolve
• Default: []

logLevel (Array of Strings or false, optional)

• Controls logging output to STDOUT/STDERR
• Options: 'info', 'warn', 'error'
• Messages will be printed if their level is included in the array
• Set to false or leave blank to suppress all messages except critical errors (e.g., port already in use)
• Default: []

connectInterval (Number, optional)

• How often, in milliseconds, to attempt connections to unconnected wanted peers and send heartbeats
• The reconnection timer is unref'd, so it does not keep the process alive on its own (a listening server or your own code will)
• Mainly useful for shortening the interval in tests
• Default: 1000

preBroadcast(data, topic) (Function, optional)

• Function called before a topic is published across the network
• Receives the message body Object as the first argument and the topic as the second
• Return the Object to be broadcast across the network, or false to prevent propagation
• Useful for filtering sensitive data before network transmission
• Only used on P2PSub class instances

Example of stripping data from a message:

const p2p = new P2PSub({
	listenPort: 7575,
	preBroadcast: function(data, topic){
		let thisData = {...data} // copy the object or all local subscriptions will lose the data too
		delete thisData.sensitiveInformation;

		return thisData
	}
});

CLI Usage

A simple relay peer can be set up using just the CLI, no code required. This peer will only relay messages to all its connected peers. The logging level is automatically set to info.

If the package is installed globally or run with npx:

npx p2psub 7575 10.1.0.1:7575 10.2.0.1:7575 10.3.0.1:7575 ...

Or, when running from the repository directly:

./app.js 7575 10.1.0.1:7575 10.2.0.1:7575 10.3.0.1:7575 ...

The first argument is the listening port, optionally followed by a space-separated list of peers to connect with. Run with help (or omit the port) to print usage instructions.

API Reference

Exports

P2PSub - Combined P2P + PubSub class P2P - Peer-to-peer networking class PubSub - Standalone pub/sub class ps - Singleton PubSub instance for convenience

Methods are provided by either the P2P or PubSub classes. The P2PSub class combines both and exposes subscribe(), publish(), addPeer(), removePeer(), and destroy().

PubSub Class Methods

subscribe(topic, callback)

• Sets a callback function to be called when topic is published
• topic can be a String (exact match) or RegExp (pattern match)
• Callback receives (data, topic, from) as arguments, where from is the originating peer ID when used through P2PSub (otherwise undefined)
• Returns: an unsubscribe Function — call it to remove the callback

publish(topic, data, [from])

• Executes all callbacks subscribed to the given topic
• Returns a Promise that resolves once every matching listener has settled. Async listeners (those returning a Promise) are awaited
• from is an opaque origin identifier forwarded to listeners; P2PSub.publish sets it to the local peer ID automatically
• To prevent propagation across the network, set __local = true in the data object
• Returns: Promise (resolves to undefined)

topics (Object)

• Instance attribute holding all topics and their bound callbacks
• Not exposed by the P2PSub class

P2P Class Methods

addPeer(peer)

• Adds a remote peer to the local peer's connections
• peer can be a single String ('remotehost.com:7575') or an Array of Strings
• Duplicate and already-connected peers are ignored
• Returns: void

removePeer(peer)

• Disconnects the specified peer and removes it from the auto-reconnect list
• peer is a String with hostname and port ('remotehost.com:7575')
• Returns: void

onData(callback)

• Registers a listener to be called when this peer receives a message
• The message is passed to the callback as a parsed JSON object
• Not exposed by the P2PSub class (use subscribe instead)
• Returns: void

broadcast(message, [excludes])

• Sends a JSON message to all connected peers
• message is a JSON object that will be stringified
• excludes (optional) is an Array of peerIDs to skip (for internal use)
• Not exposed by the P2PSub class (use publish instead)
• Returns: void

destroy()

• Stops the reconnection timer, closes the listening server, and destroys every active peer socket
• Available on both the P2P and P2PSub classes; safe to call more than once
• Returns: void

peerID (String)

• Randomly generated unique identifier for the local peer
• For internal use only
• Not exposed by the P2PSub class

Testing

Run the test suite using Node.js built-in test runner:

npm test

Tests cover:

• PubSub functionality (subscriptions, publications, regex patterns, async publish, unsubscribe handles)
• P2P networking (connections, message forwarding, peer management, message framing, lifecycle/destroy)
• P2PSub integration (cross-network pub/sub, filtering, preBroadcast hooks, peer attribution)

Contributing

This project is in active development. Please report issues and request features via GitHub Issues. Pull requests are welcome.

License

MIT