docker run -d --name rabbit -p 5672:5672 -e rabbitmq:3.7.15-management-alpine

// Интерфейс товара заказа
export interface Product {
  id: string;
  name: string;
  price: number;
}
 
// Общий интерфейс заказа
export interface Order {
  clientName: string;
  city: string;
  email?: string;
  phone?: string;
  products: Product[];
  totalSum: number;
  deliveryAddress?: string;
}
 
// Интерфейс заказа у которого есть номер телефона клиента
export interface OrderWithPhone extends Order {
  phone: string;
}
// Интерфейс заказа у которого есть адрес доставки
export interface OrderWithDeliveryAddress extends Order {
  deliveryAddress: string;
}
 
// Types Guard'ы для определения какой заказ к нам пришел
export const isOrderWithPhone = (order: Order): order is OrderWithPhone => Boolean(order.phone);
 
export const isOrderWithDeliveryAddress = (order: Order): order is OrderWithDeliveryAddress =>
  Boolean(order.deliveryAddress);
 
// Сообщение в рамках системы.
export interface Message<O extends Order> {
  errors: string[];
  retry: number;
  order: O;
 
// Интерфейс сообщения которое будет отправленно в хранилище ошибок
export interface FailOrder extends Message<Order> {
  exchange: string;
  routingKey: string;
}

import { Types, ExchangeTypes } from '../constants';
import { Options } from 'amqplib';
 
// Типы объектов RabbitMQ которые мы будем использовать для конфигурации
export enum Types {
  QUEUE = 'queue',
  EXCHANGE = 'exchange',
}
 
// Типы обменников которые будем использовать
export enum ExchangeTypes {
  TOPIC = 'topic',
}
 
// Интерфейс описания очереди
export interface Queue {
  name: string;
  options: Options.AssertQueue;
}
 
// Интерфейс описания обменника
export interface Exchange {
  name: string;
  type: ExchangeTypes;
}
 
// Интерфейс описания привязки
export interface Binding {
  type: Types;
  destination: string;
  source: string;
  routingKey: string;
}
 
// Интерфейс конфигурации RabbitMQ
export interface PipelineConfig {
  queues: Queue[];
  exchanges: Exchange[];
  bindings: Binding[];
}

export default [
  // Очередь для сообщений для которых нужно сгенерировать routingKey
  {
    name: 'generateRoutingKey',
    options: {
      durable: true,
    },
  },
  // Очередь отправки sms
  {
    name: 'sendSms',
    options: {
      durable: true,
    },
  },
  // Очередь интеграции со службой доставки
  {
    name: 'delivery',
    options: {
      durable: true,
    },
  },
  // Отложенная очередь для сообщений которые ожидают повторной отправки sms
  {
    name: 'sendSmsHold',
    options: {
      durable: true,
      deadLetterExchange: 'notify',
      deadLetterRoutingKey: 'sendSms',
      messageTtl: 60000,
    },
  },
  // Отложенная очередь для сообщений которые ожидают повторной отправки в службу доставки
  {
    name: 'deliveryHold',
    options: {
      durable: true,
      deadLetterExchange: 'integrates',
      deadLetterRoutingKey: 'delivery',
      messageTtl: 60000,
    },
  },
];

import { ExchangeTypes } from '../constants';
 
export default [
  {
    name: 'postprocessing',
    type: ExchangeTypes.TOPIC,
  },
  {
    name: 'notify',
    type: ExchangeTypes.TOPIC,
  },
  {
    name: 'integrates',
    type: ExchangeTypes.TOPIC,
  },
];

import { Types } from '../constants';
 
export default [
  {
    type: Types.EXCHANGE,
    destination: 'notify',
    source: 'postprocessing',
    routingKey: '#.notify.#',
  },
  {
    type: Types.EXCHANGE,
    destination: 'integrates',
    source: 'postprocessing',
    routingKey: '#.integrates.#',
  },
  {
    type: Types.QUEUE,
    destination: 'generateRoutingKey',
    source: 'postprocessing',
    routingKey: 'generateRoutingKey',
  },
  {
    type: Types.QUEUE,
    destination: 'sendSms',
    source: 'notify',
    routingKey: '#.sendSms.#',
  },
  {
    type: Types.QUEUE,
    destination: 'delivery',
    source: 'integrates',
    routingKey: '#.delivery.#',
  },
  {
    type: Types.QUEUE,
    destination: 'sendSmsHold',
    source: 'notify',
    routingKey: 'sendSmsHold',
  },
  {
    type: Types.QUEUE,
    destination: 'deliveryHold',
    source: 'integrates',
    routingKey: 'deliveryHold',
  },
];

import { PipelineConfig } from '../interfaces';
import exchanges from './exchanges';
import queues from './queues';
import bindings from './bindigs';
 
export const pipelineConfig: PipelineConfig = {
  exchanges,
  queues,
  bindings,
};

import { connect, Connection, Channel } from 'amqplib';
 
export class RabbitConnect {
  private _uri: string;
  private _connection: Connection;
  private _chanel: Channel;
  constructor() {
    // Строка подключения к rabbit будет браться из окружения
    this._uri = process.env.RABBIT_URI || 'amqp://localhost';
  }
  protected async connect() {
    this._connection = await connect(this._uri);
    this._chanel = await this._connection.createChannel();
  }
  protected async disconnect() {
    await this._chanel.close();
    return this._connection.close();
  }
  protected get chanel() {
    return this._chanel;
  }
}

import { RabbitConnect } from './RabbitConnect';
import { PipelineConfig } from './interfaces';
import { Types } from './constants';
 
export class Pipeline extends RabbitConnect {
  private _pipeline: PipelineConfig;
  constructor(pipelineConfig: PipelineConfig) {
    super();
    this._pipeline = pipelineConfig;
  }
  public async create() {
    try {
      await this.connect();
 
      // Создаем очереди
      const createQueues = this._pipeline.queues.map(queue =>
        this.chanel.assertQueue(queue.name, queue.options),
      ) as PromiseLike<any>[];
      // Создаём обменники
      const createExchanges = this._pipeline.exchanges.map(exchange =>
        this.chanel.assertExchange(exchange.name, exchange.type),
      ) as PromiseLike<any>[];
 
      await Promise.all([...createQueues, ...createExchanges]);
      // После создания необходимых компонентов создаём биндинги
      const createBindings = this._pipeline.bindings.map(binding => {
        if (binding.type === Types.QUEUE) {
          return this.chanel.bindQueue(binding.destination, binding.source, binding.routingKey);
        }
        return this.chanel.bindExchange(binding.destination, binding.source, binding.routingKey);
      });
 
      await Promise.all(createBindings);
      return this.disconnect();
    } catch (error) {
      console.error(error);
      throw new Error(error);
    }
  }
}

import { RabbitConnect } from './RabbitConnect';
import { Message, Order, FailOrder } from './interfaces';
import { ConsumeMessage } from 'amqplib';
 
export interface WorkerParams {
  maxRetry?: number; // Максимальное количество повторов обработки
  active: string; // Имя активной очереди
  exchange: string; // Имя обменника из которого пришло сообщение
  holdKey?: string; // Ключ маршрутизации для отложенной очереди
}
 
export abstract class Worker<M extends Order> extends RabbitConnect {
  private _maxRetry: number;
  private _active: string;
  private _holdKey: string | undefined;
  protected exchange: string;
  private _currentMessage: Message<M>;
  private _currentConsumeMessage: ConsumeMessage;
  constructor({ active, holdKey, exchange, maxRetry }: WorkerParams) {
    super();
    this._maxRetry = maxRetry || 0;
    this._active = active;
    this._holdKey = holdKey;
    this.exchange = exchange;
  }
  public async subscribe() {
    await this.connect();
    this.chanel.consume(this._active, this.checkMessage.bind(this));
  }
  // Метод проверки для сообщений у которых превышен лимит повторов
  private async checkMessage(message: ConsumeMessage) {
    this._currentConsumeMessage = message;
    const orderMessage: Message<M> = JSON.parse(message.content.toString());
    if (orderMessage.retry >= this._maxRetry) {
      await this.sendToErrorStorage('Превышен лимит попыток');
    }
    this._currentMessage = orderMessage;
    // Если количество попыток не превышено вызываем метод с бизнес логикой
    await this.handler(orderMessage.order || orderMessage);
  }
  // Метод отправки сообщения в хранилище ошибок
  protected async sendToErrorStorage(error: string) {
    const message: FailOrder = {
      order: this._currentMessage.order,
      errors: [...this._currentMessage.errors, error],
      retry: this._currentMessage.retry + 1,
      exchange: this.exchange,
      routingKey: this._active
    };
    console.log('Отправка в хранилище ошибок', message);
    this.ack();
  }
  // Метод отправки сообщения в отложенную очередь
  protected async hold(error: string) {
    if (!this._holdKey) {
      return;
    }
    const orderMessage = {
      order: this._currentMessage.order,
      errors: [...this._currentMessage.errors, error],
      retry: this._currentMessage.retry + 1,
    };
    const orderData = Buffer.from(JSON.stringify(orderMessage));
    return this.chanel.publish(this.exchange, this._holdKey, orderData);
  }
  // Метод подтверждения удачной обработки сообщения
  protected async ack() {
    return this.chanel.ack(this._currentConsumeMessage);
  }
  protected abstract handler(message: M): void;
}

import { Worker } from '../Worker';
import {
  isOrderWithPhone,
  isOrderWithDeliveryAddress,
  Order,
  Message,
} from '../interfaces';
import { Keys } from '../constants';
 
export class GenerateRoutingKey extends Worker<Order> {
  constructor() {
    super({
      active: 'generateRoutingKey',
      exchange: 'postprocessing',
    });
  }
  protected async handler(order: Order) {
    try {
      const routingKey: string[] = [];
      if (isOrderWithPhone(order)) {
        routingKey.push(Keys.SEND_SMS);
      }
      if (isOrderWithDeliveryAddress(order)) {
        routingKey.push(Keys.SEND_TO_DELIVERY);
      }
      const message: Message<Order> = {
        retry: 0,
        errors: [],
        order,
      };
      await this.chanel.publish(
        this.exchange,
        routingKey.join('.'),
        Buffer.from(JSON.stringify(message)),
      );
      await this.ack();
    } catch (error) {
      console.error(error);
      await this.sendToErrorStorage(error);
    }
  }
}

import { Worker } from '../Worker';
import { OrderWithPhone } from '../interfaces';
 
export class SendSms extends Worker<OrderWithPhone> {
  constructor() {
    super({
      active: 'sendSms',
      exchange: 'notify',
      holdKey: 'sendSmsHold',
      maxRetry: process.env.MAX_RETRY ? parseInt(process.env.MAX_RETRY) : 5,
    });
  }
  protected async handler(message: OrderWithPhone) {
    try {
      console.log('Отправка sms на номер: ', message.phone);
      this.ack();
    } catch (error) {
      console.error(error);
      await this.hold(error);
    }
  }
}

import { Worker } from '../Worker';
import { OrderWithDeliveryAddress } from '../interfaces';
 
export class Delivery extends Worker<OrderWithDeliveryAddress> {
  constructor() {
    super({
      active: 'delivery',
      exchange: 'interates',
      holdKey: 'deliveryHold',
      maxRetry: process.env.MAX_RETRY ? parseInt(process.env.MAX_RETRY) : 5,
    });
  }
  protected async handler(message: OrderWithDeliveryAddress) {
    try {
      console.log('Отправка заказа в службу доставки на адрес: ', message.deliveryAddress);
      this.ack();
    } catch (error) {
      console.error(error);
      await this.hold(error);
    }
  }
}

import { Pipeline } from './Pipeline';
import { pipelineConfig } from './pipeline';
import { GenerateRoutingKey } from './workers/GenerateRoutingKey';
import { SendSms } from './workers/SendSms';
import { Delivery } from './workers/Delivery';
 
(async () => {
  try {
    const pipeline = new Pipeline(pipelineConfig);
    const generateRoutingKey = new GenerateRoutingKey();
    const sendSms = new SendSms();
    const delivery = new Delivery();
    await pipeline.create();
    await Promise.all([generateRoutingKey.subscribe(), sendSms.subscribe(), delivery.subscribe()]);
  } catch (error) {
    console.error(error);
    process.exit(1);
  }
})();