feature(@nestjs/microservices) RabbitMQ support #887

AlariCode · 2018-07-17T16:27:48Z

PR Checklist

Please check if your PR fulfills the following requirements:

The commit message follows our guidelines: https://github.com/nestjs/nest/blob/master/CONTRIBUTING.md
Tests for the changes have been added (for bug fixes / features)
Docs have been added / updated (for bug fixes / features)

PR Type

What kind of change does this PR introduce?

[ ] Bugfix
[x] Feature
[ ] Code style update (formatting, local variables)
[ ] Refactoring (no functional changes, no api changes)
[ ] Build related changes
[ ] CI related changes
[ ] Other... Please describe:

What is the current behavior?

No RabbitMQ transport.

What is the new behavior?

Added RabbitMQ transport.

Does this PR introduce a breaking change?

[ ] Yes
[x] No

Other information

I ❤️ NestJS and use it in our company in production. But we use RabbitMQ. It is one of popular message queue engine. It would be great if RQM transport will become part of NestJS. It uses RPC pattern, in which we have one main queue which server (one or more instance) listens. When client sends message to main queue, it creates temporary queue in which it listens for an answer. This allow many clients communicate with server at ones.

You had old RQM example in your docs (as custom Strategy) that uses two queue, but we will run into issues if we have two different clients with different messages.

I covered code with test. Next step I'll write docs if pull request is useful for NestJS project.

…ure/rmq-transport

shekohex · 2018-07-17T20:11:48Z

package.json

@@ -35,6 +35,8 @@
    "@nestjs/microservices": "5.0.0",
    "@nestjs/testing": "5.0.0",
    "@nestjs/websockets": "5.0.0",
+    "@types/amqplib": "^0.5.7",


Please, can you move this line to dev dependencies?

Yes, my mistake. Moved it to dev dependencies.

shekohex · 2018-07-17T20:14:17Z

package.json

@@ -35,6 +35,8 @@
    "@nestjs/microservices": "5.0.0",
    "@nestjs/testing": "5.0.0",
    "@nestjs/websockets": "5.0.0",
+    "@types/amqplib": "^0.5.7",
+    "amqplib": "^0.5.2",


and also we are forced to download this package if we don't even will not use RMQ !
could you make it as optional if possible 😃 !

I think that if microservices is required to run, it has to be installed, whether you like it or not. (could lead to can't read property of undefined error at runtime). In the end, I believe that adapters should be in separate packages to decrease installed bundle size, but that was a mistake in structuring from the beginning.

Yeah, but do you know, sometimes mistakes are useful.

AlariCode · 2018-07-18T09:42:43Z

I updated merge request. I moved "@types/amqplib" to dev dependencies and rewrote client and server with correlationId like official documentation suggested: https://www.rabbitmq.com/tutorials/tutorial-six-javascript.html. To match requests and responses I used EventEmitter. So now there is only one temporary queue for each client. No additional queue created on publish().

wbhob · 2018-07-18T16:41:45Z

I don't personally use microservices, so I'm reluctant to leave an approval, but otherwise this looks good to me

weeco · 2018-07-19T19:39:53Z

Maybe I am missing it, but is it possible to specify rabbitmq options like

Setting the prefetch count (global and consumerspecific)
Setting queue options (e. g. exclusive channel, exchange type, etc)
Consumer options like autoack

Also is there a chance you could offer a generic function for handleMessage so that you can pass a TypeScript interface for the expected message type? (See my rabbitmq lib here: https://github.com/weeco/rabbitmq-plus#rpc-sample-usage ).

Nice to see more RabbitMQ users here, I asked Kamil several months ago if he would accept a pull request for it as I wanted to commit that feature as well, but I never got a response. I hope he will accept the feature.

AlariCode · 2018-07-20T06:03:25Z

@weeco, thank's for feedback and ideas! I dived into documentation:

Prefetch is really useful on high loaded projects, I'll add it. I never actually used this option in production.
Yes, it will add some flexibility.
Could you please give some use cases for this option? Why consumer will need manual acknowledgements and how will he use it?

As for generic functions, you are right. In send method you can already pass Interface public send<TResult = any, TInput = any>.

weeco · 2018-07-20T06:41:28Z

@AlariCode Sure, my use case is a "request worker" which is consuming a rate limited REST Api. Each worker is allowed to perform x requests every second and I am using a prefetch of 100 there. I always want to use the full rate limit. For this purpose I introduced a second channel which publishes token at the API's rate limit (400 tokens published every second, equally distributed over a second).

So what I am doing is:

Prefetch 100 job mesages (request x from endpoint y) without autoack
Wait for a token from the token queue
Once there is a token available process the oldest job (requesting the resource from the rest api)
Acknowledge the job message and send the response using the replyTo channel back to the publisher

In general I don't see any issue with providing the user all original amqplib options? Why wouldn't you make all options available to him, it shouldn't be more implementation effort either.

AlariCode · 2018-07-20T09:23:20Z

Thank’s! I’ll add configs to my pull request.

AlariCode · 2018-07-20T15:08:12Z

I updated request and added some options. Everything seams fine when I run client with ClientProxyFactory like this:

  this.client = ClientProxyFactory.create({
      transport: Transport.RMQ,
      options: {
        url: `amqp://${conf.amqp.login}:${conf.amqp.password}@${conf.amqp.host}`,
        queue: 'test',
        queueOptions: { durable: false }
      }
    });

But when I try using @Client decorator:

  @Client({
    transport: Transport.RMQ,
    options: {
      url: `amqp://${conf.amqp.login}:${conf.amqp.password}@${conf.amqp.host}`,
      queue: 'test',
      queueOptions: { durable: false }
    }
  })

I got and error after send method:

Error: connect ECONNREFUSED 127.0.0.1:3000
    at TCPConnectWrap.afterConnect [as oncomplete] (net.js:1161:14)

Can anyone help me with decorator? It seams like it is trying t use TCP transport.

kamilmysliwiec · 2018-07-25T17:58:12Z

Hi @AlariCode,
Such an amazing contribution! Thank you so much 🙂I'll take a look at your issues. In the meantime, could you please create a PR to the docs with the new transport method?

AlariCode · 2018-07-26T09:03:11Z

Sure, I'll add PR to docs :)

kamilmysliwiec · 2018-07-26T12:49:10Z

Another point - we need to cover RabbitMQ in the integration tests

ryanswart · 2018-08-05T04:32:31Z

@AlariCode Did you try increasing the timeouts for tests? Integration tests often have these timing issues - see https://jestjs.io/docs/en/troubleshooting

AlariCode · 2018-08-11T09:17:11Z

I looked deeper into the problem and found some confusing behavior:

   // Timeout error (no answer)
   @Get('a')
    a(): Observable<string> {
        let msg = 'test ' + Math.random();
        console.log('Client sent: ' + msg);
        return this.client.send<string, string>({ cmd: 'test' }, msg);
    }

    // Works fine!
    @Get('b')
    b(): Promise<string> {
        let msg = 'test ' + Math.random();
        console.log('Client sent: ' + msg);
        return this.client.send<string, string>({ cmd: 'test' }, msg).pipe(take(1)).toPromise();
    }

Strange difference in Promise and Observable. That's why integration tests failed with RMQ transport. Any suggestions what could be wrong?

AlariCode · 2018-08-11T12:36:46Z

Fixed return Observable. The problem was with 'responseEmitter.once'. It was not waiting for 'isDisposed: true'. Changed it to 'responseEmitter.on' so it will wait for all events.

kamilmysliwiec · 2018-08-16T12:13:21Z

It seems that few conflicts have appeared recently. Would you like to take charge of it?

AlariCode · 2018-08-16T15:25:26Z

@kamilmysliwiec merged master into feature/rmq-transport and resolved conflicts.

ruscon · 2018-08-21T12:23:53Z

@AlariCode, @kamilmysliwiec

I suggest you also look

Auto-reconnect and round robin support for amqplib - https://github.com/benbria/node-amqp-connection-manager
Support for clustering (many hosts, supported by node-amqp-connection-manager)
Example http://take.ms/jXVFp
Support amqps protocol, amqplib has issue ERROR: unable to verify the first certificate amqp-node/amqplib#331 (comment)
We also rely on this article http://skillachie.com/2014/06/27/rabbitmq-exchange-to-exchange-bindings-ampq/
Ideally there should be an opportunity to specify an arbitrary configuration when running

Without this, it's impossible to work in a large project

These are my thoughts on this topic :)

Example

import * as amqp from 'amqplib';
/** @types/amqp-connection-manager has old types, so for now I can do only this */
import AmqpManager = require('amqp-connection-manager');

export interface LoggerInterface {
    log(message: string): any;
    info(message: string): any;
    warn(message: string): any;
    error(message: string, trace?: string): any;
}

/**
 * How to work this implementation ?
 * @link http://skillachie.com/2014/06/27/rabbitmq-exchange-to-exchange-bindings-ampq/
 *
 * What is routingKey ?
 * @link https://www.cloudamqp.com/blog/2015-09-03-part4-rabbitmq-for-beginners-exchanges-routing-keys-bindings.html
 * @link https://www.rabbitmq.com/tutorials/tutorial-five-javascript.html
 */
export class EventDispatcher {
    private readonly connection: AmqpManager.AmqpConnectionManager;
    private publisherChannelWrapper: AmqpManager.ChannelWrapper;

    /**
     * @param {string} currentPlatformName
     * @param {string[]} hosts
     * @param {LoggerInterface} logger - any logger, for example console
     */
    constructor(private readonly currentPlatformName: string, private readonly hosts: string[], private readonly logger: LoggerInterface) {
        this.connection = AmqpManager.connect(this.hosts, {
            /** @see https://github.com/squaremo/amqp.node/issues/331 */
            // servername: parsedURI.hostname,
            heartbeatIntervalInSeconds: 5,
        });

        this.connection.on('connect', () => {
            this.logger.info(`amqp connected to the platform "${this.currentPlatformName}"`);
        });
        this.connection.on('disconnect', params => {
            this.logger.error(`amqp disconnected from the "${this.currentPlatformName}"`, params.err.stack);
        });
    }

    /**
     * Dispatch event with a message
     *
     * @param {string} eventName
     * @param {object} message
     */
    async dispatch<T>(eventName: string, message: T): Promise<void> {
        const channelWrapper = await this.getPublisherChannelWrapper();
        const fullEventName = `${this.currentPlatformName}/${eventName}`;

        try {
            await channelWrapper.publish('events_gateway', `${fullEventName}`, message, {
                deliveryMode: 2,
                contentType: 'application/json',
            });
            // TODO need to implement
            // @link https://github.com/squaremo/amqp.node/issues/103
            // await channel.waitForConfirms();
            this.logger.info(`amqp event published "${fullEventName}" with payload ${JSON.stringify(message)}`);
        } catch (error) {
            this.logger.error(`amqp message was rejected for event "${fullEventName}"`, error.stack);
            channelWrapper.close();
            this.connection.close();
        }
    }

    /**
     * TODO add retry configuration https://felipeelias.github.io/rabbitmq/2016/02/22/rabbitmq-exponential-backoff.html
     * Subscribe to the events by routingKey
     *
     * @param {string} toPlatformName - the name of the platform from which to take the routingKey
     * @param {string} routingKey - read this https://www.cloudamqp.com/blog/2015-09-03-part4-rabbitmq-for-beginners-exchanges-routing-keys-bindings.html
     * @param {function} callback
     */
    async subscribe<T>(
        toPlatformName,
        routingKey: string,
        callback: (content: T, message: amqp.Message, channel: amqp.ConfirmChannel) => Promise<boolean>,
    ): Promise<void> {
        const fullRoutingKey = `${this.currentPlatformName}/${toPlatformName}/${routingKey}`;

        const channelWrapper = this.connection.createChannel({
            json: true,
            setup: async (channel: amqp.ConfirmChannel) => {
                await this.commonChannelSetup(channel);

                const assertQueue: amqp.Replies.AssertQueue = await channel.assertQueue(fullRoutingKey, { durable: true, autoDelete: false });
                return Promise.all([
                    channel.bindQueue(assertQueue.queue, 'events_distributor', `${toPlatformName}/${routingKey}`),
                    channel.prefetch(1),
                    channel.consume(
                        assertQueue.queue,
                        async (message: amqp.Message | null) => {
                            if (!message) {
                                this.logger.error(`No message`);
                                return;
                            }

                            this.logger.info(`amqp received event "${message.fields.routingKey}" with payload ${message.content.toString()}`);

                            const content: T = JSON.parse(message.content.toString());
                            /* const isAcknowledged = */ await callback(content, message, channel);
                            // channelWrapper.ack(message);
                        },
                        { noAck: true },
                    ),
                ]);
            },
        });

        await channelWrapper.waitForConnect();

        this.logger.info(`Listening for messages from ${fullRoutingKey}`);
    }

    /**
     * General channel configuration
     *
     * @param {amqp.ConfirmChannel} channel
     */
    private async commonChannelSetup(channel: amqp.ConfirmChannel) {
        await Promise.all([
            channel.assertExchange('events_gateway', 'fanout', { durable: true, autoDelete: false }),
            channel.assertExchange('events_distributor', 'topic', { durable: true, autoDelete: false }),
            channel.bindExchange('events_distributor', 'events_gateway', '#'),
        ]);
    }

    private async getPublisherChannelWrapper(): Promise<AmqpManager.ChannelWrapper> {
        if (!this.publisherChannelWrapper) {
            this.publisherChannelWrapper = await this.connection.createChannel({
                json: true,
                setup: async (channel: amqp.ConfirmChannel) => {
                    await this.commonChannelSetup(channel);
                },
            });
        }

        return this.publisherChannelWrapper;
    }
}

luanxuechao · 2018-08-29T07:09:30Z

good job!

kamilmysliwiec · 2018-08-30T21:16:16Z

Thanks for sharing your ideas @ruscon. I believe that node-amqp-connection-manager integration might be useful. We should heavily consider to implement it beforehand.

AlariCode · 2018-09-07T10:48:03Z

@kamilmysliwiec ok, I'll add soon in my PR.

AlariCode · 2018-09-09T06:58:07Z

@kamilmysliwiec аdded amqp-connection-manager support. Now you can specify more then one url on client or server:

client = new ClientRMQ({
    urls: [`amqp://login:password@host`],
    queue: 'test',
    queueOptions: { durable: false }
});

Client and Server now reconnects if disconnected.

kamilmysliwiec · 2018-09-14T11:08:27Z

Sounds great @AlariCode, i'll review soon

juristr · 2018-09-25T13:28:06Z

@kamilmysliwiec Hey, interested in this one as we might use it at work. Do you have any estimate when this will be ready? or even a pre-release to play around with 😉

kamilmysliwiec · 2018-09-25T13:53:53Z

hey @juristr,
ETA is next minor release 5.4.0 🙂(likely to land in October)

juristr · 2018-09-25T13:55:23Z

ETA is next minor release 5.4.0 🙂(likely to land in October)

Alright. Do you plan to do prerelease versions? Or what would be the best way to integrate with RabbitMQ. We're currently having a simple node script for doing so, but I'd like to try it with NestJS as it's for sure more future proof.

AlariCode · 2018-09-25T15:55:28Z

@juristr you can play with this package https://github.com/AlariCode/nestjs-rmq. This is the same implementation (even the same code) but as a Strategy. We use it in our production projects before 5.4.0.

juristr · 2018-09-26T06:02:04Z

@AlariCode great, gonna try it 👍

rychkog · 2018-10-04T08:59:32Z

@AlariCode How about persistence option support?
Currently I'm working on a system where we are going to use Sagas pattern for distributed transactions and messages delivery should be guaranteed there (This is just as a usecase example).

Also why is noAck defaults to true? What if the message is not delivered properly?

This is a great contribution! I'm sure lots of community members were waiting for this for a long time.

kamilmysliwiec · 2018-10-05T20:49:06Z

@rychkog if you use any persistent message broker (like STAN or Kafka streams), I would recommend to not use it as a microservice strategy, but rather together with the Execution Context feature (+ manually setup all subscriptions)

@AlariCode
I have created an updated pull request which contains all your commits here: #1170

fixed client unit tests
added server unit tests
fixed integration tests
adjust code style

you can see all diffs here:

rychkog · 2018-10-08T07:58:13Z

@kamilmysliwiec Could you, please, elaborate on this more - why the persistent message broker in the microservice strategy is not recommended?

kamilmysliwiec · 2018-10-08T08:06:19Z

In general, Nest supports a bunch of built-in transporters. They are based on request-response paradigm and a whole communication logic is hidden behind an abstraction layer. Thanks to that you can easily switch between transporters without changing any line of your code. We don't support streaming platforms with log based persistance, such as Kafka or NATS streaming because they have been created to solve a different range of issues. However, you can still use them with Nest by making use of execution context feature.

https://docs.nestjs.com/microservices/basics

Generally, Nest emits additional messages to provide an efficient request-response cycle, for example, it sends a message to indicate whether stream completes already and "client side" should be notified or not. Eventually, it would create a redundancy in your persistent logs by adding either not really meaningful isDisposed messages or associated with messages correlational ids. Execution Context (as described in docs) should be a more preferable option because you have more control over the message broker, and thus it will persist only substantial events for your application.

rychkog · 2018-10-08T09:04:41Z

Thanks, I see!
@kamilmysliwiec I have one more question that I've asked here #758
Maybe you could answer it? I haven't found the answer yet
Thanks in advance!

lock · 2019-09-24T08:12:53Z

This thread has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related bugs.

AlariCode added 4 commits July 7, 2018 23:33

Added RMQ Client and Server

3da993e

feature(@nestjs/microservices) added rmq transport support

685dc33

Merge remote-tracking branch 'origin/feature/rmq-transport' into feat…

ffe4529

…ure/rmq-transport

test(@nestjs/microservices) tests for rqm transport

0ddc86e

AlariCode changed the title ~~Feature/rmq transport~~ feature(@nestjs/core) RabbitMQ support Jul 17, 2018

AlariCode changed the title ~~feature(@nestjs/core) RabbitMQ support~~ feature(@nestjs/microservices) RabbitMQ support Jul 17, 2018

AlariCode changed the title ~~feature(@nestjs/microservices) RabbitMQ support~~ feature(@nestjs/core) RabbitMQ support Jul 17, 2018

AlariCode changed the title ~~feature(@nestjs/core) RabbitMQ support~~ feature(@nestjs/microservices) RabbitMQ support Jul 17, 2018

shekohex suggested changes Jul 17, 2018

View reviewed changes

feature(@nestjs/microservices) boost performance with correlationID

bcb6c5f

shekohex approved these changes Jul 18, 2018

View reviewed changes

AlariCode added 6 commits July 19, 2018 19:20

feature(@nestjs/microservices) revert lock file to resolve conflicts

6ce0e38

feature(@nestjs/microservices) updated package-lock

e7e4bbd

Merge branch 'feature/rmq-transport'

adfa769

Merge branch 'master' into feature/rmq-transport

75e16b4

Merge updates

76fea94

Merge branch 'master' into feature/rmq-transport

bb73598

feature(@nestjs/microservices) Added additional options and fixes

e0d1364

kamilmysliwiec added type: feature 🎉 scope: microservices labels Jul 21, 2018

feature(@nestjs/microservices) fixed tests

f3a80ab

Merge branch 'master' into feature/rmq-transport

c5a752c

AlariCode added 2 commits September 8, 2018 08:48

Merge branch 'master' into feature/rmq-transport

6099f9e

feature(microservices) added amqp-connection-manager

17c5044

kamilmysliwiec changed the base branch from 5.2.0 to 5.4.0 October 3, 2018 20:28

kamilmysliwiec mentioned this pull request Oct 5, 2018

feature(microservices) RabbitMQ support #1170

Merged

3 tasks

kamilmysliwiec closed this Oct 5, 2018

lock bot locked as resolved and limited conversation to collaborators Sep 24, 2019

feature(@nestjs/microservices) RabbitMQ support #887

feature(@nestjs/microservices) RabbitMQ support #887

Conversation

AlariCode commented Jul 17, 2018 • edited Loading

PR Checklist

PR Type

What is the current behavior?

What is the new behavior?

Does this PR introduce a breaking change?

Other information

shekohex Jul 17, 2018

Choose a reason for hiding this comment

AlariCode Jul 18, 2018

Choose a reason for hiding this comment

shekohex Jul 17, 2018

Choose a reason for hiding this comment

wbhob Jul 17, 2018

Choose a reason for hiding this comment

shekohex Jul 18, 2018

Choose a reason for hiding this comment

AlariCode commented Jul 18, 2018 • edited Loading

wbhob commented Jul 18, 2018

weeco commented Jul 19, 2018

AlariCode commented Jul 20, 2018

weeco commented Jul 20, 2018

AlariCode commented Jul 20, 2018

AlariCode commented Jul 20, 2018

kamilmysliwiec commented Jul 25, 2018

AlariCode commented Jul 26, 2018

kamilmysliwiec commented Jul 26, 2018

ryanswart commented Aug 5, 2018

AlariCode commented Aug 11, 2018

AlariCode commented Aug 11, 2018

kamilmysliwiec commented Aug 16, 2018

AlariCode commented Aug 16, 2018

ruscon commented Aug 21, 2018 • edited Loading

luanxuechao commented Aug 29, 2018

kamilmysliwiec commented Aug 30, 2018

AlariCode commented Sep 7, 2018

AlariCode commented Sep 9, 2018

kamilmysliwiec commented Sep 14, 2018

juristr commented Sep 25, 2018

kamilmysliwiec commented Sep 25, 2018

juristr commented Sep 25, 2018

AlariCode commented Sep 25, 2018 • edited Loading

juristr commented Sep 26, 2018

rychkog commented Oct 4, 2018 • edited Loading

kamilmysliwiec commented Oct 5, 2018 • edited Loading

rychkog commented Oct 8, 2018

kamilmysliwiec commented Oct 8, 2018

rychkog commented Oct 8, 2018

lock bot commented Sep 24, 2019

AlariCode commented Jul 17, 2018 •

edited

Loading

AlariCode commented Jul 18, 2018 •

edited

Loading

ruscon commented Aug 21, 2018 •

edited

Loading

AlariCode commented Sep 25, 2018 •

edited

Loading

rychkog commented Oct 4, 2018 •

edited

Loading

kamilmysliwiec commented Oct 5, 2018 •

edited

Loading