bus is always empty object on listen()

[CodisRedding]

Can you help me understand why bus is being set to an empty object in a listen function? It seems that since the bus is passed into the register-handler and being used to attach listen and subscribe, that I should be using that same bus within the handler instead of re-requiring it in the module. I just didn't see the bus being exposed for use in that way.

'use strict';

const bus = require('../bus');
const config = require('../config');
const debug = require('debug')('user-create');

module.exports = {
  ack: true,
  routingKey: 'user.create',
  queueName: config.SERVICE_NAME,
  listen: function(cmd, cb) {
    const user = cmd.data;

    debug('msg', user);
    debug('publishing user.created event', cmd);
    debug('This is also an empty object: {} >>', bus);

    setTimeout(function() {
      debug('This is also an empty object: {} >>', bus);
      bus.publish('user.created', {
        userId: user.id,
        user: user.name
      });
    }, 5000);

    cb();
  }
};

[mateodelnorte]

I'm guessing that's going to depend on what your bus.js module looks like. Are you exporting the bus variable? My guess is that you're not.

Note that this inside a handler is set to this context: https://github.com/mateodelnorte/servicebus-register-handlers/blob/master/index.js#L120-L124

[CodisRedding]

I am exporting my bus, and all is well if I don't use register-handler with that same bus. The interesting thing about this is that the queueName is always undefined

{ queueName: undefined,
  routingKey: 'user.create',
  correlationId: undefined }

[mateodelnorte]

Can you paste your bus.js file contents, as well as your usage of servicebus-register-handers?

[mateodelnorte]

Here's a sample of mine:

bus.js:

const config = require('cconfig')();
const log = require('llog');
const servicebus = require('servicebus');
const retry = require('servicebus-retry');

const bus = servicebus.bus({
  enableConfirms: true,
  prefetch: config.SERVICEBUS_PREFETCH,
  url: config.RABBITMQ_URL
});

bus.use(bus.logger());
bus.use(bus.package());
bus.use(bus.correlate());
bus.use(retry({
  store: new retry.RedisStore({
    host: config.REDIS.HOST,
    port: config.REDIS.PORT
  })
}));

module.exports = bus;

service.js:

const bus = require('./lib/bus');
const config = require('cconfig')();
const log = require('llog');
const registerHandlers = require('servicebus-register-handlers');
const util = require('util');

module.exports.start = (cb) => {

  registerHandlers({
    bus: bus,
    handleError: function handleError (msg, err) {

      log.error('error handling %s: %s. rejecting message w/ cid %s and correlationId %s.', msg.type, err, msg.cid, this.correlationId);

      log.error(err);

      msg.handle.reject(function () {
        throw err;
      });

    },
    path: './lib/handlers',
    queuePrefix: 'nycnode-denormalizer-svc'
  });

  cb();

};

[CodisRedding]

yeah mine is identical, except bus is always {} within the handler. I'l keep debugging and let you know. thanks

'use strict';

const bus = require('../bus');
const config = require('../config');
const debug = require('debug')('user-create');

// debug('BUS:', bus); // logs out the RabbitMQ bus object

module.exports = {
  ack: true,
  routingKey: 'user.create',
  queueName: config.SERVICE_NAME,
  listen: function(cmd, cb) {
    // debug('BUS:', bus); // always {}
    // debug('THIS', this); // always { queueName: undefined, routingKey: 'user.create', correlationId: undefined }
    cb();
  }
};

[mateodelnorte]

Just as a note, I added a console.log(bus) inside one of my handlers from this same service. Here's what I get:

denormalizer_1      | !!!! RabbitMQBus {
denormalizer_1      |   assertQueuesOnFirstSend: true,
denormalizer_1      |   channels:
denormalizer_1      |    [ Channel {
denormalizer_1      |        domain: null,
denormalizer_1      |        _events: [Object],
denormalizer_1      |        _eventsCount: 5,
denormalizer_1      |        _maxListeners: undefined,
denormalizer_1      |        connection: [Object],
denormalizer_1      |        reply: null,
denormalizer_1      |        pending: [],
denormalizer_1      |        lwm: 1,
denormalizer_1      |        unconfirmed: [],
denormalizer_1      |        handleMessage: [Function: acceptDeliveryOrReturn],
denormalizer_1      |        consumers: {},
denormalizer_1      |        ch: 1 },
denormalizer_1      |      Channel {
denormalizer_1      |        domain: null,
...

[mateodelnorte]

Part of your problem may be magic coming from the way require works. Instead of exporting an object, try using multiple module.exports statements – one for each property/function. this, in the case of your strangely undefined queueName is probably happening because at the time your function is created, this is the object literal you're defining.

I would think that the fn.call(context, ... method of calling it would have given it our context, but at least we can rule it out.

[mateodelnorte]

Also, be sure that the first and only thing requiring this module is your servicebus-register-handlers invocation.

[CodisRedding]

so moving the functions directly to the module object module.exports.ack... This is the first time that I've run into an issue with this, in fact I make it a point to never put function directly on it and always make a single source of truth in my modules. now that I'm thinking about it, I wonder if module.exports = export = { ack: true ... } will do the trick. I'll try it out and let you know. Thanks for the help

[CodisRedding]

just an fyi, this worked:

exports = module.exports = {
  ack: true,
  routingKey: 'user.created',
  subscribe: function(event, cb) {
    const user = event.data;
    debug('msg', user);
    cb();
  }
};

[CodisRedding]

oh, and no matter which method I use. I still get { queueName: undefined,
routingKey: 'user.create',
correlationId: undefined }

[mateodelnorte]

The queueName being undefined may be a bug. All of these fields are taken directly off the header properties placed on the message by Rabbitmq (and subsequently amqplib). I remember queueName being one of the properties, but don't see it in their docs now. Perhaps it's moved and now needs to be taken from local state.

correlationId is a header property, just like described above, and will get passed down if you provide it. You can do so like the following:

const bus = require('../lib/bus');

bus.publish('subscribe.test', { test: 'params' }, { correlationId: '00001' });

or

const bus = require('../lib/bus');

bus.send('listen.test', { test: 'params' }, { ack: true, correlationId: '00001' });

[mateodelnorte]

Other than the above, very strange you were seeing bus as undefined. I've written more handlers that reference bus (to subsequently publish or send after having received an event) than I have written handlers that don't.

Any more findings on that part?

See #11 for a fix on queueName. Will push fix if everything checks out.

[CodisRedding]

I really don't know what was happening. I was trying several different approaches to wrapping the register-handlers module but none of them seemed to work until I moved the handlers function call out of my ./lib/bus.js module. Either way it's working atm, but when/if I have more time I will see if I can't debug more since I would like to wrap the handlers func into my bus wrapper as well.

[CodisRedding]

#11 fixed

[CodisRedding]

so the correlation id is still undefined but that might be because I'm logging it out to console before one is assigned? I do the id once it reaches the user.created subscriber though

THIS { queueName: 'user.create',
  routingKey: 'user.create',
  correlationId: undefined }

// command

exports = module.exports = {
  ack: true,
  routingKey: 'user.create',
  queueName: 'user',
  listen: function(cmd, cb) {
    const user = cmd.data;
    debug('THIS', this); // correlationId: undefined
    cb();
  }
};

// event

exports = module.exports = {
  ack: true,
  routingKey: 'user.created',
  subscribe: function(evt, cb) {
    const user = evt.data;
    debug('received event', evt); // cid: '75c07874-fd96-4724-bd39-dd1032898a02'
};

[mateodelnorte]

You need to provide the correlationId in your send or publish command, as an options property. Are you doing that for the send?

cid and correlationId are different. cid is placed on the message via the correlate middleware. correlationId is just a property that rabbitmq and amqplib know to store on the message.header.properties.

[CodisRedding]

ahh that was a misunderstanding on my part. thank you