Undefined dependencies and private fields in rabbitmq subscriber

[Scrib3r]

Hello everyone. ✋ I have one strange issue with RabbitSubscribe. There is module with the exchange consumer that handle files in base64 format. Module:

import { Module } from '@nestjs/common';
import { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';
import { FileConsumer } from './file.consumer';
import { FILE_SERVICE } from './di.constants';
import { FileService } from './file.service';
import { rabbitExchangesConfig } from '../config/rabbitmq';

@Module({
  imports: [
    RabbitMQModule.forRoot(RabbitMQModule, {
      uri: rabbitExchangesConfig.fileSaver.uri,
      exchanges: [
        {
          type: 'fanout',
          name: rabbitExchangesConfig.fileSaver.exchange,
        },
      ],
    }),
  ],
  providers: [
    FileConsumer,
    {
      provide: FILE_SERVICE,
      useClass: FileService,
    },
  ],
})
export class FileModule {}

Consumer:

import { Inject, Injectable } from '@nestjs/common';
import { Nack, RabbitSubscribe } from '@golevelup/nestjs-rabbitmq';
import { LoggerFactory } from '@company/logger-winston';
import { FILE_SERVICE } from './di.constants';
import { IFileService } from './file.service';
import { rabbitExchangesConfig } from '../config/rabbitmq';
import { FileSaveData } from './types';
import { FileWasNotFoundError } from './errors';

@Injectable()
export class FileConsumer {
  private readonly logger = LoggerFactory.getLogger(FileConsumer.name);

  constructor(@Inject(FILE_SERVICE) private readonly fileService: IFileService) {}

  @RabbitSubscribe({
    exchange: rabbitExchangesConfig.fileSaver.exchange,
    queue: rabbitExchangesConfig.fileSaver.queue,
    routingKey: rabbitExchangesConfig.fileSaver.routingKey,
  })
  public async uploadFile(msg: FileSaveData): Promise<null | Nack> {
    if (!msg.fileId || !msg.rawData) {
      this.logger.error('Incoming message is broken', { msg });
      return new Nack(false);
    }
    this.logger.info(`Income new message: ${msg}`, { fileId: msg.fileId });
    try {
      await this.fileService.uploadFile(msg);
      return null;
    } catch (err) {
      this.logger.error('Failed processing data', err, { fileId: msg.fileId });
      switch (true) {
        case err instanceof FileWasNotFoundError:
          return new Nack(false);
        default:
          return new Nack(true);
      }
    }
  }
}

main.ts:

import { NestFactory } from '@nestjs/core';
import { NestExpressApplication } from '@nestjs/platform-express';
import { ValidationPipe } from '@nestjs/common';
import { DocumentBuilder, SwaggerModule } from '@nestjs/swagger';
import * as bodyParser from 'body-parser';
import { AppModule } from './app.module';
import { AppConfigService } from './config';
import { WinstonLogger } from './logger';

async function bootstrap() {
  const app = await NestFactory.create<NestExpressApplication>(AppModule, {
    // logger: false,
  });
  const appConfig = app.get<AppConfigService>('AppConfigService');
  const logger = new WinstonLogger('APP');
  app.useLogger(logger);
  app.useGlobalPipes(new ValidationPipe());

  app.setGlobalPrefix('api');
  app.use(bodyParser.json({ limit: '10mb' }));
  app.use(bodyParser.urlencoded({ limit: '10mb', extended: true }));
  const options = new DocumentBuilder()
    .setTitle('Core service')
    .setDescription('Core API description')
    .setVersion('1.0')
    .build();
  const document = SwaggerModule.createDocument(app, options);
  SwaggerModule.setup('api-doc', app, document);

  return app.listen(appConfig.port);
}
bootstrap();

I see in my logs that subscriber has connected to my exchange with set queue and routing key:

2021-03-15 21:11:30 [APP] info: Initializing RabbitMQ Handlers { env: 'local', server_env: 'local', pid: 308 }
2021-03-15 21:11:30 [APP] info: Registering rabbitmq handlers from FileConsumer { env: 'local', server_env: 'local', pid: 308 }
2021-03-15 21:11:30 [APP] info: FileConsumer.uploadFile {subscribe} -> rc-oc.file_saver.spool.exchange::rc-oc.file_saver::rc-oc.file_saver.spool { env: 'local', server_env: 'local', pid: 308 }
2021-03-15 21:11:30 [APP] info: Successfully connected a RabbitMQ channel { env: 'local', server_env: 'local', pid: 308 }
2021-03-15 21:11:30 [APP] info: Nest application successfully started { env: 'local', server_env: 'local', pid: 308 }

But I have a problem with accessing to my private fields and injected dependencies because after receiving message from exchange my logger and injected service are undefined. I've tested an invocation of constructor of FileConsumer, and saw that It wasn't called before handling message from exchange. 🤯

I also have a http controllers in another modules and don't have any problems with injection. Can anyone help me? It's really strange bug. 🤔

Version of @golevelup/nestjs-rabbitmq is 1.16.0.

[Scrib3r]

I realised that I have Scope.Request dependencies inside IFileService and it was a reason of undefined properties. But I need them. Are there any solutions for injecting request dependencies per RabbitSubscriber or I should use only dependencies with default scope?

[WonderPanda]

@Scrib3r Right now there is no support for Request scoped providers since they're tied to the HTTP request lifecycle in NestJS so it doesn't map well with an event driven component like RabbitMQ that uses a persistent connection.

What is the nature of the request scoped dependency that your IFileService is depending on?

[Scrib3r]

@WonderPanda It's really disappointed 😞 I have a custom solution of pattern unity of work for transactions per every request cross all code components based on Scope.Request and TypeORM. It's work in http but not in RabbitMQ. Do you have any plans for supporting Request scope inside rabbit consumers?

[WonderPanda]

@Scrib3r I can see how Unit of Work would be a desirable pattern to be able to implement when processing messages from RabbitMQ. I'm not sure how much work is involved with being able to leverage Scope.Request outside the context of HTTP or if it is even possible but I will take a look and see what is possible

[silentroach]

+1 for request scoped consumers like in nest

[gallak87]

also +1 for request scoped if possible

[aitormunoz]

+1 for this request. At least if it's not supported, it would be fine to show an error...

[aitormunoz]

@Scrib3r For now I solved it like this.

Inside uploadFile method:

const fileService = await this.moduleRef.resolve(FILE_SERVICE);

Hopefully help you

[Scrib3r]

@aitormunoz Thank you! But I've started to use without Unit of work pattern and prayed to God for consistent saving data without transactions. 😞

[WonderPanda]

@aitormunoz Thank you! But I've started to use without Unit of work pattern and prayed to God for consistent saving data without transactions.

What ORM are you using for UoW? Does the service that handles the RabbitMQ message call out to a bunch of other services that all make changes or could you just not open up a new UoW and then flush it manually?