command run doesn't close when functionality contains observable calls via @nestjs/microservices clientProxy.send()

[ziazon]

Is there an existing issue for this?

• I have searched the existing issues

Current behavior

When calling the logic via an HTTP request, it works as expected and request completes.

When calling the logic via a nest-commander command execution, the command hangs after successfully completing all logic.

given the following microservice client:

import { firstValueFrom } from 'rxjs';

import { Injectable } from '@nestjs/common';
import { ClientProxy, ClientProxyFactory } from '@nestjs/microservices';

import { MessagingConfigService } from './messaging.config';

@Injectable()
export class MessagingService {
  private eventsClient: ClientProxy;
  constructor(private readonly configService: MessagingConfigService) {
    this.eventsClient = ClientProxyFactory.create(this.configService.eventsConfig);
  }

  send<R, I = unknown>(pattern: string, data: I) {
    return this.eventsClient.send<R, I>(pattern, data);
  }

  async sendAsync<R, I = unknown>(pattern: string, data: I) {
    return await firstValueFrom(this.send<R, I>(pattern, data));
  }

  emit<R, I = unknown>(pattern: string, data: I) {
    return this.eventsClient.emit<R, I>(pattern, data);
  }

  async emitAsync<R, I = unknown>(pattern: string, data: I) {
    return await firstValueFrom(this.emit<R, I>(pattern, data));
  }
}

calling await this.messagingService.sendAsync('message-pattern', payload) results in the command hanging even though this call returns values. Mind you, running the logic from an HTTP request works as expected, just not from the cli.

Also note that I have tested removing the microservice call and instead hard coded the response value, and it works fine in that instance.

Minimum reproduction code

@Command({
  name: 'test',
  arguments: '<action>',
  description: 'test command'
})
export class TestCommand implements CommandRunner {
  constructor(
    private readonly messagingService: MessagingService
  ) {}

  run(): Promise<void> {
     return this.messagingService.sendAsync('test-subject', {foo: 'bar'})
  }

Expected behavior

Command execution closes on response from sendAsync call.

Package

• nest-commander
• nest-commander-schematics
• nest-commander-testing

Package version

2.3.5

Node.js version

14.18.1

In which operating systems have you tested?

• macOS
• Windows
• Linux

Other

No response

[jmcdo29]

Please provide a minimum reproduction repository. Most likely, the ClientProxy isn't being closed, but I can take a look

[ziazon]

Please provide a minimum reproduction repository. Most likely, the ClientProxy isn't being closed, but I can take a look

sure thing! see: https://github.com/jubairsaidi/nest-examples

Notes:
The command file path: src/api/commands/example.command.ts.

It's covered in the readme, but you'll need to run nats via docker (or installing it on your machine) to test the microservice requests, and then run npm run start:dev to get the initial dist folder created which you need for npm run cli commands.

The screenshot of the example requests. note that the last one I had to ctrl-c out get it to terminate

Also note that requests to http requests which make microservice requests (via nats) do work properly and terminate as expected when the response gets back. (I can add this example if it's helpful)

[jmcdo29]

Thanks for the reproduction. I was able to reproduce with your code (woohoo), and I have a fix for it. But first, a little background as to what's happening:

When Nest creates a microservice client (your EventsClient) it's created lazily. It's not connected off the bat, and will only connect on manually calling connect as shown in the Nest docs (scroll up just slightly) or when the first event is sent. Because you don't end up calling sendAsync or send in your MessagingService code unless you use cli examples nats, the client is never connected and you never end up having the hang. But when you do use cli examples nats you connect the client, and it won't disconnect until told to.

There are two easy solutions here:

1. Change your send or sendAsync to include a close() call for this.eventsClient. This works okay, but you'll have to remember it each time, and if you end up making multiple sends you'll end up opening and closing connections which is slow

2. Add an OnModuleDestory or OnApplicationShutdown hook and call await this.eventsClient.close(). The top of the service could look like this:

@Injectable()
export class MessagingService implements OnModuleDestroy {
  private eventsClient: ClientProxy;
  private reqClient: AxiosInstance;
  constructor(private readonly configService: MessagingConfigService) {
    this.eventsClient = ClientProxyFactory.create(this.configService.eventsConfig);
    this.reqClient = axios.create();
  }

  async onModuleDestroy() {
    await this.eventsClient.close();
  }

With this, the event was sent, response was logged, and the command exited correctly.

[ziazon]

this helped! thanks!!

Slight modification to get it working when making multiple calls though:

@Injectable()
export class MessagingService implements OnApplicationShutdown, OnApplicationBootstrap {
  private eventsClient: ClientProxy;
  constructor(private readonly configService: MessagingConfigService) {
    this.eventsClient = ClientProxyFactory.create(this.configService.eventsConfig);
  }

  async onApplicationBootstrap() {
    await this.eventsClient.connect();
  }

  async onApplicationShutdown() {
    await this.eventsClient.close();
  }
}

Seems we have to create a connection for this client (I assume that otherwise it creates a new connection on each request automatically?)

When omitting the onApplicationBootstrap I noticed onApplicationShutdown did get fired but the connection was still hanging when I'm making multiple calls in a given command execution.

[jmcdo29]

Seems so, thanks for adding the modification! Glad I could help