node package manager

mqstreams

MQ pub/sub as streams - based on mqemitter

mqstreams  

Publish-Subscribe node streams style, based on mqemitter.

$ npm install mqemitter mqstreams --save

'use strict'
 
var mqemitter = require('mqemitter')
var mqstreams = require('mqstreams')
var emitter = mqstreams(mqemitter())
var through = require('through2')
var input = emitter.writable()
var output = emitter.readable('output/#')
 
emitter
  .readable('some/+')
  .pipe(through.obj(function (msg, enc, callback) {
    msg.topic = 'output/' + msg.topic
    this.push(msg)
    callback()
  }))
  .pipe(emitter.writable())
 
input.write({ topic: 'some/food', type: 'greek' })
input.write({ topic: 'some/startup', type: 'instasomething' })
input.end({ topic: 'some/dev', type: 'matteo' })
 
output.on('data', function (msg) {
  console.log(msg)
 
  // OUTPUT: 
  // { topic: 'output/some/food', type: 'greek' } 
  // { topic: 'output/some/startup', type: 'instasomething' } 
  // { topic: 'output/some/dev', type: 'matteo' } 
})

Extends the MQEmitter with the readable() and writable() methods.


Return a Readable stream in object mode that will include all emitter messages that match the given topic. The opts parameter is passed through to the Stream constructor. This stream fully respect the Stream3 interface.

The topic parameter is passed to the emitter.on method.

the returned object has the following method added: subscribe(), unsubscribe(), destroy().

Subscribe to the given topic, which can also be an array of topics.

Unsubscribe from the given topic, which can also be an array of topics.

Close the stream, unsubscribing from all the topics. This is aliased to close() for backwards compatibility.


Return a Writable stream in object mode that will pass any message to the emitter.emit method. This stream fully respect the Stream3 interface.

MIT