iambus
minimalist pattern-matching pub-sub message-bus
Installation
Install using npm:
npm install iambusCreating a bus
const Iambus = require('iambus')
const bus = new Iambus()Publishing Messages
Messages should be objects. To publish a message to the bus, use the pub method:
const message = {
topic: 'news',
content: 'Hello, world!'
}
bus.pub(message)Subscribing & Unsubscribing
Subscribers are Readable streamx streams.
Subscribe to a pattern (an object which partially matches against target messages).
To subscribe to messages pass a pattern object to the sub method:
const pattern= { topic: 'news' }
const subscriber = bus.sub(pattern)The pattern object must deeply-equal properties in a published message to match (but does not need to match all properties).
When in async contexts (such as async functions or top-level ESM), a for await...of loop can be used to listen for messages:
for await (const message of subscriber) {
console.log('Received one message:', message)
break
}To unsubscribe, destroy the stream. The usage of break in the for await loop causes the subscriber stream to be destroyed. Exiting the loop via return also causes the subscriber stream to be destroyed.
Here's an equivalent example with the data event and destroy method:
// Listen for messages using 'data' event
subscriber.on('data', (message) => {
console.log('Received one message:', message)
subscriber.destroy()
})A graceful stream close is also possible with the end method:
// Listen for messages using 'data' event
subscriber.on('data', (message) => {
console.log('Received one message:', message)
subscriber.end()
})Subscribing to all messages
An empty pattern object can be used to subscribe to all messages on the bus:
async function msglogger () {
for await (const message of bus.sub({})) console.log('BUS MSG', Date.now(), ' - ', message)
}
msglogger().catch(console.error)Resubscribing
To resubscribe create a new subscriber using bus.sub(pattern).
import Iambus from './index.js'
const bus = new Iambus()
let count = 0
setImmediate(() => {
bus.pub({ match: 'this', and: { also: 'this' }, content: 'Hello, world!' })
setImmediate(() => {
bus.pub({ match: 'this', and: { also: 'this' }, content: 'more content' })
setImmediate(() => {
bus.pub({ match: 'this', and: { also: 'this' }, content: 'even more content' })
})
})
})
for await (const message of bus.sub({ match: 'this', and: { also: 'this' } })) {
console.log('1st subscriber got', message)
if (++count === 2) break // destroy
}
for await (const message of bus.sub({ match: 'this', and: { also: 'this' } })) {
console.log('2nd subscriber got', message)
if (++count === 3) break // destroy
}
console.log('done')This should output:
1st subscriber got { match: 'this', and: { also: 'this' }, content: 'Hello, world!' }
1st subscriber got { match: 'this', and: { also: 'this' }, content: 'more content' }
2nd subscriber got { match: 'this', and: { also: 'this' }, content: 'even more content' }
doneExample
The example.mjs file contains both the resubscribing code and the message logger that uses an empty pattern to subscribe to all messages.
node example.mjs
Should output:
1st subscriber got { match: 'this', and: { also: 'this' }, content: 'Hello, world!' }
1st subscriber got { match: 'this', and: { also: 'this' }, content: 'more content' }
2nd subscriber got { match: 'this', and: { also: 'this' }, content: 'even more content' }
donenode example.mjs --log
Should output similar to:
BUS MSG 2023-06-21T15:25:32.897Z - { match: 'this', and: { also: 'this' }, content: 'Hello, world!' }
1st subscriber got { match: 'this', and: { also: 'this' }, content: 'Hello, world!' }
BUS MSG 2023-06-21T15:25:32.901Z - { something: 'else', whatever: 'that might be' }
1st subscriber got { match: 'this', and: { also: 'this' }, content: 'more content' }
BUS MSG 2023-06-21T15:25:32.901Z - { match: 'this', and: { also: 'this' }, content: 'more content' }
BUS MSG 2023-06-21T15:25:32.902Z - { match: 'this', and: { also: 'this' }, content: 'even more content' }
2nd subscriber got { match: 'this', and: { also: 'this' }, content: 'even more content' }
doneLicense
Apache License 2.0. See the LICENSE file for more details.