|
| 1 | +/** |
| 2 | + * The main app entry |
| 3 | + */ |
| 4 | + |
| 5 | +import config from 'config'; |
| 6 | +import amqp from 'amqplib'; |
| 7 | +import _ from 'lodash'; |
| 8 | +import logger from './common/logger'; |
| 9 | +import ConsumerService from './services/ConsumerService'; |
| 10 | +import { EVENT } from '../config/constants'; |
| 11 | + |
| 12 | +const debug = require('debug')('app:worker'); |
| 13 | + |
| 14 | +const FETCH_LIMIT = 10; |
| 15 | + |
| 16 | +let connection; |
| 17 | +process.once('SIGINT', () => { |
| 18 | + debug('Received SIGINT...closing connection...') |
| 19 | + try { |
| 20 | + connection.close(); |
| 21 | + } catch (ignore) { // eslint-ignore-line |
| 22 | + logger.logFullError(ignore) |
| 23 | + } |
| 24 | + process.exit(); |
| 25 | +}); |
| 26 | + |
| 27 | +let EVENT_HANDLERS = { |
| 28 | + [EVENT.ROUTING_KEY.PROJECT_DRAFT_CREATED]: ConsumerService.processProjectCreated |
| 29 | + // [EVENT.ROUTING_KEY.PROJECT_UPDATED]: ConsumerService.processProjectUpdated |
| 30 | +} |
| 31 | + |
| 32 | +function close() { |
| 33 | + console.log('closing self after processing messages...') |
| 34 | + try { |
| 35 | + setTimeout(connection.close.bind(connection), 30000); |
| 36 | + } catch (ignore) { // eslint-ignore-line |
| 37 | + logger.logFullError(ignore) |
| 38 | + } |
| 39 | +} |
| 40 | + |
| 41 | +export function initHandlers(handlers) { |
| 42 | + EVENT_HANDLERS = handlers; |
| 43 | +} |
| 44 | + |
| 45 | +/** |
| 46 | + * Processes the given message and acks/nacks the channel |
| 47 | + * @param {Object} channel the target channel |
| 48 | + * @param {Object} msg the message to be processed |
| 49 | + */ |
| 50 | +export function processMessage(channel, msg) { |
| 51 | + return new Promise((resolve, reject) => { |
| 52 | + if (!msg) { |
| 53 | + reject(new Error('Empty message. Ignoring')); |
| 54 | + return; |
| 55 | + } |
| 56 | + debug(`Consuming message in \n${msg.content}`); |
| 57 | + const key = _.get(msg, 'fields.routingKey'); |
| 58 | + debug('Received Message', key, msg.fields); |
| 59 | + |
| 60 | + let handler; |
| 61 | + let data; |
| 62 | + try { |
| 63 | + handler = EVENT_HANDLERS[key]; |
| 64 | + if (!_.isFunction(handler)) { |
| 65 | + logger.error(`Unknown message type: ${key}, NACKing... `); |
| 66 | + reject(new Error(`Unknown message type: ${key}`)); |
| 67 | + return; |
| 68 | + } |
| 69 | + data = JSON.parse(msg.content.toString()); |
| 70 | + } catch (ignore) { |
| 71 | + logger.info(ignore); |
| 72 | + logger.error('Invalid message. Ignoring'); |
| 73 | + resolve('Invalid message. Ignoring'); |
| 74 | + return; |
| 75 | + } |
| 76 | + return handler(logger, data).then(() => { |
| 77 | + resolve(msg); |
| 78 | + return; |
| 79 | + }) |
| 80 | + .catch((e) => { |
| 81 | + // logger.logFullError(e, `Error processing message`); |
| 82 | + if (e.shouldAck) { |
| 83 | + debug("Resolving for Unprocessable Error in handler..."); |
| 84 | + resolve(msg); |
| 85 | + } else { |
| 86 | + debug("Rejecting promise for error in msg processing...") |
| 87 | + reject(new Error('Error processing message')); |
| 88 | + } |
| 89 | + }); |
| 90 | + }) |
| 91 | +} |
| 92 | + |
| 93 | +function assertExchangeQueues(channel, exchangeName, queue) { |
| 94 | + channel.assertExchange(exchangeName, 'topic', { durable: true }); |
| 95 | + channel.assertQueue(queue, { durable: true }); |
| 96 | + const bindings = _.keys(EVENT_HANDLERS); |
| 97 | + const bindingPromises = _.map(bindings, rk => |
| 98 | + channel.bindQueue(queue, exchangeName, rk)); |
| 99 | + debug('binding queue ' + queue + ' to exchange: ' + exchangeName); |
| 100 | + return Promise.all(bindingPromises); |
| 101 | +} |
| 102 | + |
| 103 | +/** |
| 104 | + * Start the worker |
| 105 | + */ |
| 106 | +export async function start() { |
| 107 | + try { |
| 108 | + console.log("Scheduled Worker Connecting to RabbitMQ: " + config.rabbitmqURL.substr(-5)); |
| 109 | + connection = await amqp.connect(config.rabbitmqURL); |
| 110 | + connection.on('error', (e) => { |
| 111 | + logger.logFullError(e, `ERROR IN CONNECTION`); |
| 112 | + }) |
| 113 | + connection.on('close', () => { |
| 114 | + debug('Before closing connection...') |
| 115 | + }) |
| 116 | + debug('created connection successfully with URL: ' + config.rabbitmqURL); |
| 117 | + const connect2sfChannel = await connection.createConfirmChannel(); |
| 118 | + debug('Channel created for consuming failed messages ...'); |
| 119 | + connect2sfChannel.prefetch(FETCH_LIMIT); |
| 120 | + assertExchangeQueues( |
| 121 | + connect2sfChannel, |
| 122 | + config.rabbitmq.connect2sfExchange, |
| 123 | + config.rabbitmq.queues.connect2sf |
| 124 | + ).then(() => { |
| 125 | + debug('Asserted all required exchanges and queues'); |
| 126 | + let counter = 0; |
| 127 | + _.range(1, 11).forEach(() => { |
| 128 | + return connect2sfChannel.get(config.rabbitmq.queues.connect2sf). |
| 129 | + then((msg) => { |
| 130 | + if (msg) { |
| 131 | + return processMessage( |
| 132 | + connect2sfChannel, |
| 133 | + msg |
| 134 | + ).then((responses) => { |
| 135 | + counter++; |
| 136 | + debug('Processed message'); |
| 137 | + connect2sfChannel.ack(msg); |
| 138 | + if (counter >= FETCH_LIMIT) { |
| 139 | + close(); |
| 140 | + } |
| 141 | + }).catch((e) => { |
| 142 | + counter++; |
| 143 | + debug('Processed message with Error'); |
| 144 | + connect2sfChannel.nack(msg); |
| 145 | + logger.logFullError(e, `Unable to process one of the messages`); |
| 146 | + if (counter >= FETCH_LIMIT) { |
| 147 | + close(); |
| 148 | + } |
| 149 | + }) |
| 150 | + } else { |
| 151 | + counter++; |
| 152 | + debug('Processed Empty message'); |
| 153 | + if (counter >= FETCH_LIMIT) { |
| 154 | + close(); |
| 155 | + } |
| 156 | + } |
| 157 | + }).catch(() => { |
| 158 | + console.log('get failed to consume') |
| 159 | + }) |
| 160 | + }) |
| 161 | + }) |
| 162 | + |
| 163 | + } catch (e) { |
| 164 | + logger.logFullError(e, `Unable to connect to RabbitMQ`); |
| 165 | + } |
| 166 | +} |
| 167 | + |
| 168 | +if (!module.parent) { |
| 169 | + start(); |
| 170 | +} |
0 commit comments