MQEmitter

To use MQEmitter in your project, install it using npm:

npm install mqemitter

When initializing MQEmitter, you can provide the following options:

• concurrency <number>: Maximum number of concurrent messages that can be on concurrent delivery. Default: 0
• wildcardOne <string>: A character to use for matching exactly one non-empty level word. Default: +
• wildcardSome <string>: A character to use for matching multiple level wildcards. Default: #
• matchEmptyLevels <boolean>: If true, wildcardOne also matches an empty word. Default: true
• separator <string>: A separator character to use for separating words. Default: /

The concurrency option in the MQEmitter constructor controls how many messages are processed simultaneously.

• If concurrency is 0 (default), messages are processed immediately.
• If concurrency > 0, and the number of active messages reaches this limit, new messages are added to an internal queue (_messageQueue) and processed sequentially as previous ones complete.

You can create a new MQEmitter instance by calling the exported function or using the new keyword. You can pass an optional configuration object to customize behavior like concurrency and wildcard characters.

const mq = require('mqemitter')
const emitter = mq({ concurrency: 5 })

Use emitter.emit(message, callback) to send a message. The message object must contain a topic property. The callback is invoked when the emission process is complete. Note that the emitter will not return an error.

const message = { topic: 'hello/world', payload: 'data' }

emitter.emit(message, function (err) {
  // err will be null
})

Use emitter.on(topic, listener, [callback]) to subscribe to a topic or a wildcard pattern.

Important Constraints:

• The listener function must be (message, done) => void.
• The listener must never throw an error.
• The done callback passed to the listener must not be called with an error object (e.g., done(err) is invalid; use done() instead).
• The optional third argument callback is executed when the subscription is successfully established.

emitter.on('hello/world', function (message, done) {
  console.log(message.payload)
  done()
}, function () {
  console.log('Subscribed!')
})

Create a new MQEmitter instance to handle message emission with topic-based matching. You can configure the separator and wildcard characters used for topic matching via the opts object.

const MQEmitter = require('mqemitter');

const emitter = new MQEmitter({
  separator: '/',      // Default: '/'
  wildcardOne: '+',    // Default: '+' (matches exactly one level)
  wildcardSome: '#',   // Default: '#' (matches zero or more levels)
  matchEmptyLevels: true, // Default: true
  concurrency: 0       // Default: 0 (0 means no limit on concurrent processing)
});

You can manage subscriptions using two methods:

1. removeListener(topic, notify, [done]): Removes a specific listener function for a specific topic.
2. removeAllListeners(topic, [done]): Removes all listeners associated with a topic pattern.

Both methods execute the removal asynchronously via setImmediate and can accept an optional done callback.

const notify = (message, cb) => { /* ... */ };

// Remove a specific listener
emitter.removeListener('sensors/temperature', notify, () => {
  console.log('Listener removed');
});

// Remove all listeners for a topic pattern
emitter.removeAllListeners('sensors/+', () => {
  console.log('All sensors listeners removed');
});

Use emitter.on(topic, notify, [done]) to register a listener for a specific topic. The notify function will be called when a message matching the topic is emitted. If a done callback is provided, it will be executed asynchronously via setImmediate after the listener is registered.

// Register a listener for a specific topic
emitter.on('sensors/temperature', (message, cb) => {
  console.log('Received:', message.topic, message.payload);
  cb(); // Call the callback to signal completion
});

// Register a listener with a completion callback
emitter.on('sensors/+', (message, cb) => {
  // ...
}, () => {
  console.log('Listener registered');
});