How to publish to Kafka client in NestJS?

user_1826758 0
Viewed 5370

I have tried to publish to kafka in NestJs 

 async publish<T extends IEvent>(event: T) {
    await this.client.connect();
    await this.client.send('topic', event);
  }

But yet not found the correct method, dispatchEvent is protected.

Edit: using cqrs so this is inside a eventpublisher subscribed to the eventbus.

nestjs
1 Answers

I don't believe you need the connect method. Have you made sure you're subscribing to the message response? The below is a working example with a client controller and server controller:

Client Controller

import {
  Controller,
  Get,
  Inject,
  OnModuleDestroy,
  OnModuleInit,
  UseFilters,
} from '@nestjs/common';
import { ClientKafka } from '@nestjs/microservices';
import { ExceptionFilter } from './exception.filter';

@Controller()
export class KafkaClientController implements OnModuleInit, OnModuleDestroy {
  constructor(@Inject('KAFKA_SERVICE') private readonly kafka: ClientKafka) {}

  async onModuleInit() {
    ['hello', 'error', 'skip'].forEach((key) =>
      this.kafka.subscribeToResponseOf(`say.${key}`),
    );
  }

  onModuleDestroy() {
    this.kafka.close();
  }

  @Get()
  sayHello() {
    return this.kafka.send('say.hello', { ip: '127.0.0.1' });
  }

  @Get('error')
  @UseFilters(ExceptionFilter)
  sayError() {
    return this.kafka.send('say.error', { ip: '127.0.0.1' });
  }

  @Get('skip')
  saySkip() {
    return this.kafka.send('say.skip', { ip: '127.0.0.1' });
  }
}

Server Controller

import { BadRequestException, Controller, UseFilters } from '@nestjs/common';
import { MessagePattern } from '@nestjs/microservices';
import { OgmaSkip } from '@ogma/nestjs-module';
import { AppService } from '../../app.service';
import { ExceptionFilter } from './exception.filter';

@Controller()
export class KafkaServerController {
  constructor(private readonly service: AppService) {}

  @MessagePattern('say.hello')
  sayHello() {
    return this.service.getHello();
  }

  @UseFilters(ExceptionFilter)
  @MessagePattern('say.error')
  sayError() {
    throw new BadRequestException('Borked');
  }

  @OgmaSkip()
  @MessagePattern('say.skip')
  saySkip() {
    return this.service.getHello();
  }
}

The above were used for integration testing of a library I was making. You can check out the full module setups here

user_95761860
Related