Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
},
err => {
if (!err) {
return msg.finish()
}
msg.requeue()
}
)
})
readers.set(readerKey, reader)
}
// only one writer for this service
const writer = new Nsq.Writer(
opts.nsqWriter.host,
opts.nsqWriter.port,
opts.nsqWriter.options
)
writer.connect()
writer.on('error', err => {
hemera.log.error(err, 'NSQ Writer error')
hemera.fatal() // Let it crash and restart
})
writer.on('closed', () => {
hemera.log.warn('NSQ Writer closed')
})
OutputNsq.prototype.start = function(callback) {
if (!this.topic||!this.dataUrl) { logger.error('Critical Error! Missing config!'); return; }
logger.info('Initializing NSQ Publisher...');
const w = new nsq.Writer(this.dataUrl, this.dataTcpPort);
var count = 0;
w.connect();
w.on('ready', () => {
logger.info('NSQ Writer Ready!');
this.nsqw = w;
});
w.on('closed', () => {
logger.error('NSQ Writer closed!');
if (count < this.retries) {
logger.info('Reconnecting in '+this.timeout+'');
setTimeout(function(){ w.connect(); }, this.timeout);
} else {
logger.error('Giving up! Attempted:',this.retries);
}