Messaging

A complete implementation of synchronous and asynchronous messaging for use in microservices architecture.

Available:

• Kafka

• Fake

Methods:

Emit: It is used to send a message in a topic. It is dynamically typed and accepts any data structure.

this.messagingProvider.emit('TOPIC', {
  user: { age: 18, name: 'John', surname: 'Doe' },
});

Listen: It waits for a message and, upon receiving it, executes a callback (or an array of callbacks, it has support for behaving like Express middlewares).

const userController = new UserController();

messagingProvider.listen('CREATE-USER', isAuthenticated, userValidator, userController.create);

SubscribeFrom: It is used to sign a reply to a topic (it is used together with the 'send' method).

class Controller {
  public constructor(
    private readonly messagingProvider: IMessagingProvider,
  ) {
    // It will listen to 'SEND-DATA.reply'
    this.messagingProvider.subscribeFrom('SEND-DATA');
  }
}

Send: It is used to send an indexed message and wait for the message to return in a defined partition (timeout set by observerTimeout), and should be used together with subscribeFrom.

class RequestController {
  public constructor(
    private readonly messagingProvider: IMessagingProvider,
  ) {
    this.messagingProvider.subscribeFrom('SEND-MESSAGE');
  }

  public async getMessage(): Promise<IMessageDTO> {
    const message: IMessageDTO = await this.messagingProvider.send(
      'SEND-MESSAGE',
      {
        message: 'Hello, send me a message!',
      },
    );

    return message;
  }
}

class ReplyController {
  public constructor(
    private readonly messagingProvider: IMessagingProvider,
  ) {}

  public async sendMessage(data: IBaseMessageDTO<IMessageDTO>): Promise<void> {
    this.messagingProvider.emit(
      data.replyTopic,
      {
        reply: 'Hello, take your message!',
      },
      {
        replyId: data.replyId,
        replyPartition: data.replyPartition,
      },
    );
  }
}

Close: It is used to close the connection with the messaging.

this.messagingProvider.close();

---