feat(rabbitmq): wIP rabbitmq adapter

Author: juanmarin-co

package-lock.json

@@ -91,6 +91,7 @@
     "@types/amqplib": "^0.5.13",
     "@types/jest": "^24.0.17",
     "@types/node": "^12.6.8",
+    "@types/uuid": "^3.4.5",
     "colors": "^1.3.3",
     "commitizen": "^4.0.3",
     "coveralls": "^3.0.5",

package.json

@@ -18,7 +18,8 @@ export default {
   // Indicate here external modules you don't wanna include in your bundle (i.e.: 'lodash')
   external: [
     'amqplib',
-    'amqp-connection-manager'
+    'amqp-connection-manager',
+    'uuid'
   ],
   watch: {
     include: 'src/**'

rollup.config.ts

@@ -0,0 +1,64 @@
+import { connect, AmqpConnectionManager, ChannelWrapper } from 'amqp-connection-manager'
+import { ConnectionOptions } from 'tls'
+import { URL } from 'url'
+
+import { Broker, BrokerSender, BrokerReceiver, Connection } from '../../domain/model/broker.model'
+import { AmqpSender } from './amqp-sender'
+import { AmqpReceiver } from './amqp-receiver'
+
+export class AmqpBroker implements Broker {
+  connectionManager?: AmqpConnectionManager
+  channel?: ChannelWrapper
+  sender?: BrokerSender
+  receiver?: BrokerReceiver
+
+  constructor(private url: string, private options?: ConnectionOptions) {}
+
+  connect(): Connection {
+    if (this.connectionManager) {
+      throw new Error('AMQP connection already created')
+    }
+
+    const connection = connect(
+      [this.url],
+      { connectionOptions: this.options }
+    )
+
+    const channel = connection.createChannel({
+      json: true
+    })
+
+    this.connectionManager = connection
+    this.channel = channel
+    this.sender = new AmqpSender(channel)
+    this.receiver = new AmqpReceiver(channel)
+
+    return {
+      onConnect: listener => {
+        connection.on('connect', connected => {
+          const url = new URL(connected.url)
+          listener({ host: url.host })
+        })
+      },
+      onDisconnect: listener => {
+        connection.on('disconnect', err => listener(err.err))
+      }
+    }
+  }
+
+  getSender(): AmqpSender {
+    if (!this.connectionManager) {
+      throw new Error('AMQP sender must be accessed after connection creation')
+    }
+
+    return this.sender as AmqpSender
+  }
+
+  getReceiver(): AmqpReceiver {
+    if (!this.connectionManager) {
+      throw new Error('AMQP receiver must be accessed after connection creation')
+    }
+
+    return this.receiver as AmqpReceiver
+  }
+}

src/adapter/amqp/amqp-adapter.ts

@@ -0,0 +1,22 @@
+import { ChannelWrapper } from 'amqp-connection-manager'
+import { ConfirmChannel } from 'amqplib'
+
+import { BrokerReceiver, MessageConsumer } from '../../domain/model/broker.model'
+
+export class AmqpReceiver implements BrokerReceiver {
+  constructor(private channel: ChannelWrapper) {}
+
+  async consume(queueName: string, consumer: MessageConsumer): Promise<void> {
+    await this.channel.addSetup((channel: ConfirmChannel) => {
+      return channel.consume(queueName, message => {
+        if (message) {
+          consumer(JSON.parse(message.content.toString()))
+            .then(() => channel.ack(message))
+            .catch(() => {
+              channel.nack(message)
+            })
+        }
+      })
+    })
+  }
+}

src/adapter/amqp/amqp-receiver.ts

@@ -0,0 +1,33 @@
+import { ChannelWrapper } from 'amqp-connection-manager'
+import { ConfirmChannel } from 'amqplib'
+
+import {
+  BrokerSender,
+  QueueSpecitication,
+  ExchangeSpecification,
+  BindingSpecification
+} from '../../domain/model/broker.model'
+
+export class AmqpSender implements BrokerSender {
+  constructor(private channel: ChannelWrapper) {}
+
+  declareQueue({ name, ...rest }: QueueSpecitication): Promise<void> {
+    return this.channel.addSetup((channel: ConfirmChannel) => channel.assertQueue(name, rest))
+  }
+
+  declareTopic({ name }: ExchangeSpecification): Promise<void> {
+    return this.channel.addSetup((channel: ConfirmChannel) => channel.assertExchange(name, 'topic'))
+  }
+
+  bind({ queue, topic, routingKey }: BindingSpecification): Promise<void> {
+    return this.channel.addSetup((channel: ConfirmChannel) =>
+      channel.bindQueue(queue, topic, routingKey)
+    )
+  }
+
+  publish(topic: string, routingKey: string, content: object) {
+    return this.channel.publish(topic, routingKey, content as Buffer, {
+      deliveryMode: 2
+    })
+  }
+}

src/adapter/amqp/amqp-sender.ts

@@ -0,0 +1,38 @@
+import {
+  CommandHandler,
+  HandlerRegistryRoutes,
+  QueryHandler,
+  EventListener
+} from './model/message-handlers.model'
+
+export class HandlerRegistry {
+  commandHandlers: HandlerRegistryRoutes<CommandHandler<any>> = {}
+  queryHandlers: HandlerRegistryRoutes<QueryHandler<any, any>> = {}
+  eventListeners: HandlerRegistryRoutes<EventListener<any>> = {}
+
+  private constructor() {}
+
+  public static register() {
+    return new HandlerRegistry()
+  }
+
+  public handleCommand<M>(commandName: string, handler: CommandHandler<M>): void {
+    this.push(this.commandHandlers, commandName, handler)
+  }
+
+  public serverQuery<M, R>(queryName: string, handler: QueryHandler<M, R>): void {
+    this.push(this.queryHandlers, queryName, handler)
+  }
+
+  public listenEvent<M>(eventName: string, handler: EventListener<M>): void {
+    this.push(this.eventListeners, eventName, handler)
+  }
+
+  private push<H>(routes: HandlerRegistryRoutes<H>, path: string, handler: H) {
+    if (!routes[path]) {
+      routes[path] = []
+    }
+
+    routes[path].push(handler)
+  }
+}

src/domain/handler-registry.ts

@@ -0,0 +1,39 @@
+export interface QueueSpecitication {
+  name: string
+  exclusive: boolean
+}
+
+export interface ExchangeSpecification {
+  name: string
+}
+
+export interface BindingSpecification {
+  queue: string
+  topic: string
+  routingKey: string
+}
+
+export interface Connection {
+  onConnect(listener: (event: { host: string }) => void): void
+  onDisconnect(listener: (err: Error) => void): void
+}
+
+export type MessageConsumer = (messageContent: object) => Promise<void>
+
+export interface BrokerSender {
+  declareQueue(specification: QueueSpecitication): Promise<void>
+
+  declareTopic(specification: ExchangeSpecification): Promise<void>
+
+  bind(specification: BindingSpecification): Promise<void>
+}
+
+export interface BrokerReceiver {
+  consume(queueName: string, consumer: MessageConsumer): Promise<void>
+}
+
+export interface Broker {
+  connect(): Connection
+  getSender(): BrokerSender
+  getReceiver(): BrokerReceiver
+}

src/domain/model/broker.model.ts

@@ -0,0 +1,7 @@
+export type CommandHandler<M> = (message: M) => Promise<void>
+export type QueryHandler<M, R> = (message: M) => Promise<R>
+export type EventListener<M> = (message: M) => Promise<void>
+
+export type HandlerRegistryRoutes<H> = {
+  [path: string]: H[]
+}

src/domain/model/message-handlers.model.ts

@@ -0,0 +1,105 @@
+import { connect, AmqpConnectionManager, ChannelWrapper } from 'amqp-connection-manager'
+import { ConnectionOptions } from 'tls'
+
+import { AmqpBroker } from '../../src/adapter/amqp/amqp-adapter'
+import { AmqpSender } from '../../src/adapter/amqp/amqp-sender'
+import { AmqpReceiver } from '../../src/adapter/amqp/amqp-receiver'
+
+jest.mock('amqp-connection-manager', () => ({
+  _esModule: true,
+  connect: jest.fn(() => ({
+    createChannel: jest.fn(),
+    on: jest.fn()
+  }))
+}))
+
+const mockedConnect = connect as jest.MockedFunction<typeof connect>
+const url = 'amqp://myrabbitserver'
+const options: ConnectionOptions = { timeout: 10 }
+
+describe(`AMQP Broker`, () => {
+  let broker: AmqpBroker
+
+  beforeEach(() => {
+    jest.clearAllMocks()
+    broker = new AmqpBroker(url, options)
+  })
+
+  it(`connects using connection manager`, () => {
+    broker.connect()
+
+    const connectionManager = mockedConnect.mock.results[0].value as AmqpConnectionManager
+
+    expect(mockedConnect).toHaveBeenCalledTimes(1)
+    expect(mockedConnect.mock.calls[0][0]).toStrictEqual([url])
+    expect(mockedConnect.mock.calls[0][1]).toStrictEqual({ connectionOptions: options })
+    expect(broker.connectionManager).toBe(connectionManager)
+
+    const mockedChannel = connectionManager.createChannel as jest.MockedFunction<
+      typeof connectionManager.createChannel
+    >
+
+    expect(mockedChannel).toHaveBeenCalledTimes(1)
+    expect(mockedChannel.mock.calls[0][0]).toStrictEqual({ json: true })
+  })
+
+  it(`connect can only be called once`, () => {
+    broker.connect()
+    expect(() => broker.connect()).toThrow('AMQP connection already created')
+  })
+
+  it(`must create an instance of AmqpSender`, () => {
+    broker.connect()
+
+    expect(broker.getSender()).toBeInstanceOf(AmqpSender)
+  })
+
+  it(`must create an instance of AmqpReceiver`, () => {
+    broker.connect()
+
+    expect(broker.getReceiver()).toBeInstanceOf(AmqpReceiver)
+  })
+
+  it(`AmqpSender can only be accessed after connect`, () => {
+    expect(() => broker.getSender()).toThrow(
+      'AMQP sender must be accessed after connection creation'
+    )
+  })
+
+  it(`AmqpReceiver can only be accessed after connect`, () => {
+    expect(() => broker.getReceiver()).toThrow(
+      'AMQP receiver must be accessed after connection creation'
+    )
+  })
+
+  it(`should call onConnect callback when connectionManager emits it`, () => {
+    const connection = broker.connect()
+    const connectionManager = mockedConnect.mock.results[0].value as AmqpConnectionManager
+    const mockedOn = connectionManager.on as jest.MockedFunction<typeof connectionManager.on>
+
+    const onConnect = jest.fn()
+    connection.onConnect(onConnect)
+
+    const listener = mockedOn.mock.calls[0][1] as (arg: object) => void
+    listener({ url })
+
+    expect(mockedOn.mock.calls[0][0]).toBe('connect')
+    expect(onConnect.mock.calls[0][0]).toStrictEqual({ host: 'myrabbitserver' })
+  })
+
+  it(`should call onDisconnect callback when connectionManager emits it`, () => {
+    const connection = broker.connect()
+    const connectionManager = mockedConnect.mock.results[0].value as AmqpConnectionManager
+    const mockedOn = connectionManager.on as jest.MockedFunction<typeof connectionManager.on>
+    const error = Error('something went wrong')
+
+    const onDisconnect = jest.fn()
+    connection.onDisconnect(onDisconnect)
+
+    const listener = mockedOn.mock.calls[0][1]
+    listener({ err: error })
+
+    expect(mockedOn.mock.calls[0][0]).toBe('disconnect')
+    expect(onDisconnect.mock.calls[0][0]).toBe(error)
+  })
+})