Message Broker for NestJs

The idea behind this module is to connect it to a message broker that enables the permanent distribution of messages.

Install

$ npm i @cubiles/nestjs-message-broker

| Support | Message-Broker | Docker-Image | |---------------|----------------|---------------------------------------------| | ✅ Supported | RabbitMQ | heidiks/rabbitmq-delayed-message-exchange | | ✴️ In Process | Redis | |

Features

• ✅ simple Usage without much configuration
• ✅ support the native routing pattern
• ✅ auto-retry of messages with a lot of strategies
• ✅ group messages in Namespaces and Scopes

Example

Code

@Module({
  imports: [
    MessageBrokerModule.register(RabbitMQBroker, {
      broker: {
        password: 'guest',
        user: 'guest',
        host: 'localhost',
        port: 5672,
      },
      name: 'my-broker',
      namespace: 'user-service',
      delimiter: '.',
      wildcards: '*',
      multiLevelWildcards: '#',
    }),
  ],
  providers: [AppService],
  exports: [AppService],
})
export class AppModule {
}

@MessageEvent('user')
export class UserEvent implements IMessageEvent {
  userId: string;

  constructor(userId: string) {
    this.userId = userId;
  }
}

@MessageEvent('created')
export class UserCreatedEvent extends UserEvent {
}

@Injectable()
export class AppService {
  constructor(
    @InjectMessageBroker()
    private readonly messageBroker: RabbitMQBroker,
  ) {
  }

  @OnMessageEvent(UserEvent, { exact: false })
  async handleUser(user) {
    console.log('On user', user);
  }

  @OnMessageEvent('user.created')
  async handleCreatedUser(user) {
    console.log('New user', user);
  }

  async sendMessages() {
    await this.messageBroker.emit(new UserCreatedEvent('1024'));
  }
}

Auto-build of Message-Brokers

Exchanges

• my-broker.user-services.messages
• my-broker.user-services.retries

Queues

• my-broker.user-services.retries
• my-broker.user-services.app-service.handle-created-user
• my-broker.user-services.app-service.handle-user

Flow-Chart

1. Publish in my-broker.messages
2. Route to
   1. my-broker.user-services.app-service.handle-created-user
   2. my-broker.user-services.app-service.handle-user
3. Process message

What happens if the processing fails

1. Reroute to my-broker.user-services.retries
2. Calculate the delay of the next try
3. Publish again in my-broker.user-services.messages and wait the delay
4. Route only to the subscriber of failed process

Retry-Strategy

| function | default | 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9 | 10 | ... | |------------|-----------------------------------------|--------|--------|---------|---------|---------|---------|---------|----------|----------|----------|--------| | constant | delay=5min | 5min | 5min | 5min | 5min | 5min | 5min | 5min | 5min | 5min | 5min | 5min |
| linear | pitch=1min min=0.5s max=6h | 0.5s | 5min | 10min | 15min | 20min | 25min | 30min | 35min | 40min | 45min | 6h |
| cube | pitch=1s min=0.5s max=6h base=3 | 0.5s | 1.5s | 4.5s | ~13s | ~40s | ~2min | ~6min | ~18min | ~54min | ~2.75h | 6h |
| steps | steps=[...] | 1s | 5s | 30s | 3min | 10min | 1h | 6h | 6h | 6h | 6h | 6h |

Api

Coming soon

Types

export interface OnMessageEventOptions {
  /**
   * Custom name of queue
   *
   * Default ${class-name}/${methode-key}
   */
  queue?: string;

  /**
   * Print the errors int the console
   *
   * Default true
   */
  suppressErrors?: string;

  /**
   * Maximal count of parallel processing
   *
   * Default 1
   */
  prefetch?: number;

  /**
   * Priority of queue
   *
   * Default 0
   */
  priority?: number;

  /**
   * Listen only in these scopes for new messages.
   * The global scope is always active.
   *
   * Default []
   */
  scope?: string | Array<string>;
}

export interface MessageBrokerEmitOption {
  /**
   * Delay of consume the messages
   *
   * Default 0
   */
  delay?: number;

  /**
   * Messages priority
   *
   * Default 0
   */
  priority?: number;

  /**
   * Emit the message in the scope
   *
   * Default global
   */
  scope?: string;
}

export interface MessageBrokerOptions<T> {
  /**
   * Is the module global?
   *
   * Default false
   */
  global?: boolean;

  /**
   * Custom serializer methode of string to buffer
   *
   * Default JSON.parse() and JSON.stringify()
   */
  serializer?: MessageBrokerSerializer;

  /**
   * Calculate the delay of a retry if messages failed
   *
   * Default MessageBrokerRetryStrategy.cube()
   */
  retryStrategy?: IMessageBrokerRetryStrategy;

  /**
   * Option of the usages messages broker
   */
  broker: T;

  /**
   * Change the default scope of messages
   *
   * Default global
   */
  defaultScope?: string;

  /**
   * Namespace of queue and exchanges
   */
  namespace?: string | null;
}