peer to peer message queue
With this program, we will set up a peer to listen and one to connect and send messages. These roles aren't static: either peer could listen and either could send messages.
var peermq = require('peermq')
var path = require('path')
var { Transform } = require('readable-stream')
var minimist = require('minimist')
var argv = minimist(process.argv.slice(2), {
alias: { m: 'message', d: 'datadir' }
})
require('mkdirp').sync(argv.datadir)
var mq = peermq({
network: require('peermq/network'),
storage: function (name) {
return path.join(argv.datadir, name)
}
})
if (argv._[0] === 'id') {
mq.getId(function (err, id) {
if (err) console.error(err)
else console.log(id.toString('hex'))
})
} else if (argv._[0] === 'add-peer') {
mq.addPeer(argv._[1], function (err) {
if (err) console.error(err)
})
} else if (argv._[0] === 'listen') {
mq.listen()
mq.createReadStream('unread', { live: true }).pipe(new Transform({
objectMode: true,
transform: function ({ from, seq, data }, enc, next) {
console.log(`MESSAGE ${from}@${seq} ${data.toString()}`)
mq.archive({ from, seq }, next)
}
}))
} else if (argv._[0] === 'send') {
mq.send({ to: argv.to, message: argv.message }, function (err) {
if (err) console.error(err)
})
} else if (argv._[0] === 'connect') {
mq.connect(argv._[1])
}
create two new databases and print their IDs:
$ A=$(node mq.js -d /tmp/a id); echo $A
41f541aa50bbd9948e794b3a5d55b4267f731a94c1ee4053470df9f514762ac4
$ B=$(node mq.js -d /tmp/b id); echo $B
19eee824db6456a97f7e90a75dfde8e6ad3a4b34eee65373823a50276f33872e
allow incoming connections from node B on node A:
$ node mq.js -d /tmp/a add-peer $B
create some messages on node B to send to node A:
$ node mq.js -d /tmp/b send --to $A --message wow
$ node mq.js -d /tmp/b send --to $A --message cool
$ node mq.js -d /tmp/b send --to $A --message hiiiiiiii
connect to node A from node B:
$ node mq.js -d /tmp/b connect $A
and in another terminal, listen on node A. Once a connection is made, you should see messages arrive:
$ node mq.js -d /tmp/a listen
MESSAGE 19eee824db6456a97f7e90a75dfde8e6ad3a4b34eee65373823a50276f33872e@0 wow
MESSAGE 19eee824db6456a97f7e90a75dfde8e6ad3a4b34eee65373823a50276f33872e@1 cool
MESSAGE 19eee824db6456a97f7e90a75dfde8e6ad3a4b34eee65373823a50276f33872e@2 hiiiiiiii
on node B, we can disconnect (ctrl+c), queue more messages to send, then reconnect:
$ node mq.js -d /tmp/b connect $A
^C
$ node mq.js -d /tmp/b send --to $A --message beep
$ node mq.js -d /tmp/b send --to $A --message boop
$ node mq.js -d /tmp/b connect $A
back on node A, we now see the messages arrive:
MESSAGE 19eee824db6456a97f7e90a75dfde8e6ad3a4b34eee65373823a50276f33872e@3 beep
MESSAGE 19eee824db6456a97f7e90a75dfde8e6ad3a4b34eee65373823a50276f33872e@4 boop
Because the listener marks messages as "read" once it receives them (with
mq.archive()
), we will see each message only once. This is a useful property
for building job queues or other batch processing tools.
The sender could also listen for 'ack'
events from the listener to know when
it's safe to delete messages that have been received. 'ack'
messages are sent
over an authenticated channel where the network session keys (noise) are signed
by the hypercore key.
var peermq = require('peermq')
Create a peermq instance mq
from:
opts.network
- a network adapter (userequire('peermq/network')
)opts.storage(name)
- function that returns a random-access adaptor or a string path for aname
string
Get the id of this node as a Buffer in cb(err, id)
.
Authorize a peer to connect by its publicKey
(Buffer or hex string).
Remove a peer by its publicKey
(Buffer or hex string).
Get the list of peers
, an array of hex string public keys as cb(err, peers)
.
Listen for incoming connections. Get access to the network adaptor's server
instance (from network.createServer()
) as cb(err, server)
.
Connect to a publicKey to
. connection
emits 'ack'
events for the
underlying replication stream.
Write a message
addressed to a remote public key to
. These messages will be
synchronized when a connection is available.
Create a new objectMode stream
to read records for a stream type
.
This stream pulls from all peer streams in live mode. As you pull messages from this stream, documents are downloaded from the remote database in sparse mode.
Right now type
must be 'unread'
. Soon other categories and user-created
categories will be available.
Set the document in the public key from
at the sequence number seq
to
"read" and copy this value into the "archive" set.
Remove the document in the public key from
at the sequence number seq
.
license zero parity and apache 2.0 (contributions)