Как вызвать микросервис NestJs из функции прослушивателя EventEmitter? - PullRequest
Новая
       21

Как вызвать микросервис NestJs из функции прослушивателя EventEmitter?

0 голосов
/ 23 мая 2019

Я разрабатываю два микро-сервиса с NestJS и машинописью для использования и публикации сообщений в двух разных очередях. Сценарий выглядит следующим образом

  1. Микросервис потребителя читает сообщения из раздела или очереди Kafka
  2. Затем он отправляет это сообщение второму микросервису.

Проблема появляется на втором этапе, сообщение никогда не отправляется микросервисом потребителя.

Использование сообщений реализовано с помощью Consumer из библиотеки kafka-node. Сообщения используются правильно и могут быть зарегистрированы.

Связь между микро-сервисами осуществляется с помощью RedisClient из пакета @nestjs/microservices и работает при выполнении извне метода слушателя.

Когда сообщения между микро-сервисами отправляются откуда-либо еще, я вижу их на redis-cli MONITOR, но при отправке из функции слушателя ничего не появляется.

код

Клиент Redis и инициализация клиента Kafka: 

private readonly kafkaClient: kafka.KafkaClient;
private readonly consumer: kafka.Consumer;
private readonly client: ClientRedis;

constructor() {
    this.kafkaClient = new kafka.KafkaClient(KAFKA_HOST);
    this.consumer = new kafka.Consumer(this.kafkaClient, TOPICS, OPTIONS);
    this.client = new RedisClient(REDIS_URL);
}

Функция прослушивания для приема сообщений: 

async onApplicationBootstrap() {
    this.consumer.on('message', message => {
        // Sends the message to Redis
        this.client.send<Message>(PATTERN, message);
    });
}

Тогда второй микросервис должен получить сообщение благодаря декоратору @MessagePattern(PATTERN) NestJS. Но он даже не публикуется на Redis, как я уже говорил ранее.

1 Ответ

1 голос
/ 23 мая 2019

Хорошо, поэтому проблема была в Observable, возвращенном методом ClientRedis::send().Просто преобразование Observable в Promise решило проблему: 

  this.client.send<Message>(PATTERN, message).toPromise()
...