使用 RabbitMq 从 Nodejs 微服务与 NestJs 微服务进行通信

Question

我正在尝试使用 RabbitMq 从 NestJs 微服务调用特定函数，但无法使用 nodejs 命中特定消息模式。

我尝试从控制器文件访问的消息模式

@MessagePattern('signin')
  async signin(
    @Payload(ValidationPipe) signInCredentialsDto: SignInCredentialsDto,
    @Ctx() context: RmqContext,
  ): Promise<{ accessToken: string }> {
    this.logger.verbose('Signing in ' + JSON.stringify(signInCredentialsDto));
    // const result = await this.authService.userSignin(signInCredentialsDto);
    // const channel = context.getChannelRef();
    // const originalMsg = context.getMessage();

    // channel.ack(originalMsg);
    return {
      accessToken: 'works',
    };
  }

我的 NestJsrabbitmq 连接详细信息

async function bootstrap() {
  const logger = new Logger('Main');
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(
    AppModule,
    {
      transport: Transport.RMQ,
      options: {
        urls: ['amqp://localhost:5672'],
        queue: 'rebus_queue',
        queueOptions: {
          durable: false,
        },
      },
    },
  );
  app.listen(() => {
    logger.log('Auth Microservice Started Successfully');
  });
}
bootstrap();

我的nodejs客户端代码：

amqp.connect("amqp://localhost:5672", (error0, connection) => {
  if (error0) throw error0;

  connection.createChannel((error1, channel) => {
    if (error1) throw error1;

    channel.assertQueue("rebus_queue", {
      durable: false,
    });

    app.get("/", (req: express.Request, res: express.Response) => {});

    app.post(
      "/auth/signin",
      async (req: express.Request, res: express.Response) => {
        channel.sendToQueue(
          "rebus_queue",
          Buffer.from(JSON.stringify(req.body))
        );
      }
    );
  });
});

当我点击 /auth/signin 路线时，我成功点击了 Nestjs 微服务，但我不知道如何访问登录消息模式。

就目前情况而言，点击nodejs http端点会导致nestjs终端出现以下错误

[Server] There is no matching event handler defined in the remote service. Event pattern: undefined. +50026ms

Answer 1

我有一个 typecrit/express 发布者和一个 NestJS 订阅者；我认为这工作正常。

import amqplib from 'amqplib';

const payload = Buffer.from(JSON.stringify({
  pattern: 'queue name',
  data: 'your actual payload',
}));

...
channel.sendToQueue('my_queue_name', payload);