Class AMQPLibPlugin

src/plugins/AMQPLibPlugin.ts:28–121

import PluginInstaller from '../core/PluginInstaller';

class AMQPLibPlugin implements SwPlugin {
  readonly module = 'amqplib';
  readonly versions = '*';

  install(installer: PluginInstaller): void {
    const { BaseChannel } = installer.require?.('amqplib/lib/channel') ?? require('amqplib/lib/channel');

    this.interceptProducer(BaseChannel);
    this.interceptConsumer(BaseChannel);
  }

  interceptProducer(BaseChannel: any): void {
    const _sendMessage = BaseChannel.prototype.sendMessage;

    BaseChannel.prototype.sendMessage = function (fields: any, properties: any, content: any) {
      const topic = fields.exchange || '';
      const queue = fields.routingKey || '';
      const peer = `${this.connection.stream.remoteAddress}:${this.connection.stream.remotePort}`;
      const span = ContextManager.current.newExitSpan(
        'RabbitMQ/' + topic + '/' + queue + '/Producer',
        Component.RABBITMQ_PRODUCER,
      );

      span.start();

      try {
        span.inject().items.forEach((item) => {
          fields.headers[item.key] = item.value;
        });

        span.component = Component.RABBITMQ_PRODUCER;
        span.layer = SpanLayer.MQ;
        span.peer = peer;

        span.tag(Tag.mqBroker((this.connection.stream.constructor.name === 'Socket' ? 'amqp://' : 'amqps://') + peer));

        if (topic) span.tag(Tag.mqTopic(topic));

        if (queue) span.tag(Tag.mqQueue(queue));

        const ret = _sendMessage.call(this, fields, properties, content);

        span.stop();

        return ret;
      } catch (e) {
        span.error(e);
        span.stop();

        throw e;
      }
    };
  }

  interceptConsumer(BaseChannel: any): void {
    const _dispatchMessage = BaseChannel.prototype.dispatchMessage;

    BaseChannel.prototype.dispatchMessage = function (fields: any, message: any) {