randomize163 / menashmq Goto Github PK

Usage:
const queue = menash.queue('my-queue-name');
const exchange = menash.exchange('my-exchange-name');

menash.queue('my-queue-name').send(...)

Hi,

When connection fails an MenashMQ logs to user "[RabbitMQ Client] Starting reinitialize...". This does not allow monitoring of logs with their logging libraries.

Possible solution:
Add possibility of listening to "reinitialize event" so user can decide how to log the error.
maybe for simplifying, keep the default behaviour, because many clients will forget to listen to that event.

Example syntax

import menash from 'menashmq';
menash.on('errror', (err) => logger.log(err));

I noticed if I want to use delay plugin see here, then I cant because menash expects ExchangeType (direct, fanout, headers, topic)

Currently menash.declareTopology() defines queues, exchanges and bindings.

I suggest adding consumers to the list:
{
consumers: [
{ queue: string | Queue, consume: (msg: ConsumerMessage)=> {}, options: ConsumeOptions},
],
}

Channel::isInitialized() is overridden by Queue and Exchange.
When parent class Channel checks if it initialized (by calling isInitialzed()) it calls childs function isInitialized() and gets the wrong result causing connection handle to leak.

Currently, it seems there is no way to know whether or not there already is a connection to rabbit, which complicates the users of the package.
An isConnected function could fix that.

I would love to see type support for

1. which options are avaiable with declareQueue & exchangeQueue
2. and what is the default for each one

maybe even add a short description of each option

Add option to manage a connection pool of different, unrelated, rabbitmq instances.

possible solution:
Same solution as mongoose (see here), that creates a connection and returns an instance of an object exposing the same API as exposed by the default library.

When user forgets to specify the exchange type, menashmq should throw an error explaining what types of exchanges are supported.

I think there should be no default value for exchange. User should understand the differences and choose the best option for his use case.

Hi,

In the Typescript-Support paragraph

import menash from 'menashmq';

await menash.connect('amqp://localhost');
await menash.declateQueue('menash-queue');

Need to change from declateQueue to declareQueue.

Thank you so much for this awsome package @Randomize163

Need to add support for RPC Client and Server

Usage could look like this:

Client:
const result = await menash.rpc('server-queue-name', { info: "some request" });

Server:
await menash.activateRpcServer('server-queue-name', (msg: ConsumerMessage) => { msg.reply({ info: "my reply" }); })

Hi,

if my consumer throws after I already acked/nacked message, MenashMQ throws the following error:

(node:46) UnhandledPromiseRejectionWarning: Error: [BUG] message already processed ack: false nack: true
at ConsumerMessage.nack (/usr/src/app/node_modules/menashmq/dist/lib/message.js:53:19)
at consume (/usr/src/app/node_modules/menashmq/dist/lib/queue.js:106:29)
at processTicksAndRejections (internal/process/task_queues.js:97:5)

This happens because MenashMQ nack-ing message for the consumer because it assumes consumer throwed before ack-ing.

see in MenashMQ consumer wrapper:

Will create pull request soon that checks message.acked and message.nacked fields before nacking.