Nestjs, External EventBus

Hello friends! In this article, I wanna share one solution to implementing external EventBus. if you haven't been acquainted with EventBus yet, you can visit this link.
By default, Nestjs allows calling event handler only locally. It's a problem when we want to publish events via external EventBus, so I suggest the following solution. Bellow, I'll show a publisher which is based on redis-pub. Also, we can find a publisher for RabbitMQ.

RedisPublisher.

import {Injectable} from '@nestjs/common';
import {Client, ClientProxy, Transport} from '@nestjs/microservices';
import {AbstractPublisher} from '../abstract.publisher';


@Injectable()
export class RedisPublisher extends AbstractPublisher {
    TRANSPORT = Transport.REDIS;
    PATTERN = 'event_bus';

    @Client({
        transport: Transport.REDIS,
        options: {
            url: 'redis://:password123@redis:6379',
        },
    })
    client: ClientProxy;

    protected async send(pattern: any, data: any) {
        try {
            await this.client.send(pattern, data).toPromise();
        } catch (e) {
            this.log.error(e);
        }
    }
}

AbstractPublisher

import {IEvent, IEventPublisher} from '@nestjs/cqrs';
import {Transport} from './transport.enum';
import {Injectable, Logger} from '@nestjs/common';

@Injectable()
export abstract class AbstractPublisher implements IEventPublisher {
    abstract TRANSPORT: Transport;
    abstract PATTERN: string;

    constructor(
        protected readonly log: Logger,
    ) {

    }

    publish<T extends IEvent>(event: T): void {

        const data = {
            payload: event,
            event: event.constructor.name,
        };

        this.send(this.PATTERN, data);
    }

    protected abstract send(pattern: any, data: any): any;
}

As you can see, RedisPublisher extends AbstractPublisher where is required to implement the send method. For a particular type of transport, we should implement own send method and add transport into the event. In common case, it's enough to start sending an event via transport.

ContrivedEvent

export class ContrivedEvent implements IEventWithTransport {
    TRANSPORTS = [Transport.RMQ, Transport.DEF, Transport.REDIS];

    constructor(
    ) {}
}
...
import {IEvent} from '@nestjs/cqrs';
import {Transport} from '../transport.enum';

export interface IEventWithTransport extends IEvent {
    TRANSPORTS: Transport[];
}
...

export enum Transport {
    TCP = 0,
    REDIS = 1,
    NATS = 2,
    MQTT = 3,
    GRPC = 4,
    RMQ = 5,
    DEF = 6,
}

As you can see, there were used three kinds of transports:

RabitMQ
Redis
Local

Using EventBusTransport:

import {Controller, Get} from '@nestjs/common';
import {EventBusTransport} from '../event-bus-transport/event.bus.transport';
import {ContrivedEvent} from '../events/contrived/contrived.event';

@Controller('/')
export class ContrivedController {

    constructor(
        readonly eventBusTransport: EventBusTransport,
    ) {}

    @Get('/contrived')
     contrived() {
        this.eventBusTransport.publish(new ContrivedEvent());
    }

}

Blog

Nestjs, External EventBus

Sergey Telpuk

Join Our Newsletter. No Spam, Only the good stuff.

Related