feat(transport): add mqtt

package.json

@@ -36,13 +36,15 @@
     "@japa/expect-type": "^2.0.2",
     "@japa/runner": "^3.1.4",
     "@swc/core": "^1.5.7",
+    "@testcontainers/hivemq": "^10.9.0",
     "@testcontainers/redis": "^10.9.0",
     "@types/node": "^20.12.12",
     "@types/object-hash": "^3.0.6",
     "c8": "^9.1.0",
     "del-cli": "^5.1.0",
     "eslint": "^8.57.0",
     "ioredis": "^5.4.1",
+    "mqtt": "^5.6.1",
     "prettier": "^3.2.5",
     "release-it": "^17.2.1",
     "ts-node": "^10.9.2",

src/transports/mqtt.ts

@@ -0,0 +1,95 @@
+/**
+ * @boringnode/bus
+ *
+ * @license MIT
+ * @copyright Boring Node
+ */
+
+import Mqtt from 'mqtt'
+import { assert } from '@poppinss/utils/assert'
+
+import debug from '../debug.js'
+import {
+  Transport,
+  TransportEncoder,
+  TransportMessage,
+  Serializable,
+  SubscribeHandler,
+  MqttProtocol,
+  MqttTransportConfig,
+} from '../types/main.js'
+import { JsonEncoder } from '../encoders/json_encoder.js'
+
+export function mqtt(config: MqttTransportConfig) {
+  return () => new MqttTransport(config)
+}
+
+export class MqttTransport implements Transport {
+  #id: string | undefined
+  #client: any
+  #url: string
+  readonly #encoder: TransportEncoder
+
+  constructor(options: MqttTransportConfig, encoder?: TransportEncoder) {
+    this.#encoder = encoder ?? new JsonEncoder()
+    this.#url = `${options.protocol || MqttProtocol.MQTT}://${options.host}${options.port ? `:${options.port}` : ''}`
+
+    this.#client = Mqtt.connect(this.#url)
+  }
+
+  setId(id: string): Transport {
+    this.#id = id
+
+    return this
+  }
+
+  async disconnect(): Promise<void> {
+    await this.#client.endAsync()
+  }
+
+  async publish(channel: string, message: any): Promise<void> {
+    assert(this.#id, 'You must set an id before publishing a message')
+
+    const encoded = this.#encoder.encode({ payload: message, busId: this.#id })
+
+    await this.#client.publishAsync(channel, encoded)
+  }
+
+  async subscribe<T extends Serializable>(
+    channel: string,
+    handler: SubscribeHandler<T>
+  ): Promise<void> {
+    this.#client.subscribe(channel, (err) => {
+      if (err) {
+        throw err
+      }
+    })
+
+    this.#client.on('message', (receivedChannel: string, message: Buffer | string) => {
+      if (channel !== receivedChannel) return
+
+      debug('received message for channel "%s"', channel)
+
+      const data = this.#encoder.decode<TransportMessage<T>>(message)
+
+      /**
+       * Ignore messages published by this bus instance
+       */
+      if (data.busId === this.#id) {
+        debug('ignoring message published by the same bus instance')
+        return
+      }
+
+      // @ts-expect-error - TODO: Weird typing issue
+      handler(data.payload)
+    })
+  }
+
+  onReconnect(): void {
+    this.#client.reconnect(this.#url)
+  }
+
+  async unsubscribe(channel: string): Promise<void> {
+    await this.#client.unsubscribeAsync(channel)
+  }
+}

src/types/main.ts

@@ -36,6 +36,24 @@ export interface RedisTransportConfig extends RedisOptions {
   useMessageBuffer?: boolean
 }
 
+export enum MqttProtocol {
+  MQTT = 'mqtt',
+  MQTTS = 'mqtts',
+  TCP = 'tcp',
+  TLS = 'tls',
+  WS = 'ws',
+  WSS = 'wss',
+  WXS = 'wxs',
+  ALIS = 'alis',
+}
+
+export interface MqttTransportConfig {
+  host: string
+  port?: number
+  protocol?: MqttProtocol
+  qos?: number
+}
+
 export interface Transport {
   setId: (id: string) => Transport
   onReconnect: (callback: () => void) => void

tests/drivers/mqtt_transport.spec.ts

@@ -0,0 +1,134 @@
+/**
+ * @boringnode/bus
+ *
+ * @license MIT
+ * @copyright Boring Node
+ */
+
+import { setTimeout } from 'node:timers/promises'
+import { test } from '@japa/runner'
+import { HiveMQContainer, StartedHiveMQContainer } from '@testcontainers/hivemq'
+import { MqttTransport } from '../../src/transports/mqtt.js'
+import { JsonEncoder } from '../../src/encoders/json_encoder.js'
+import { TransportEncoder, TransportMessage } from '../../src/types/main.js'
+
+test.group('Mqtt Transport', (group) => {
+  let container: StartedHiveMQContainer
+
+  group.setup(async () => {
+    container = await new HiveMQContainer().start()
+
+    return async () => {
+      await container.stop()
+    }
+  })
+
+  test('transport should not receive message emitted by itself', async ({ assert, cleanup }) => {
+    const transport = new MqttTransport({
+      host: container.getHost(),
+      port: container.getPort(),
+    }).setId('bus')
+    cleanup(() => transport.disconnect())
+
+    await transport.subscribe('testing-channel', () => {
+      assert.fail('Bus should not receive message emitted by itself')
+    })
+
+    await transport.publish('testing-channel', 'test')
+    await setTimeout(1000)
+  }).disableTimeout()
+
+  test('transport should receive message emitted by another bus', async ({
+    assert,
+    cleanup,
+  }, done) => {
+    assert.plan(1)
+
+    const transport1 = new MqttTransport({
+      host: container.getHost(),
+      port: container.getPort(),
+    }).setId('bus1')
+    const transport2 = new MqttTransport({
+      host: container.getHost(),
+      port: container.getPort(),
+    }).setId('bus2')
+
+    cleanup(async () => {
+      await transport1.disconnect()
+      await transport2.disconnect()
+    })
+
+    await transport1.subscribe('testing-channel', (payload) => {
+      assert.equal(payload, 'test')
+      done()
+    })
+
+    await setTimeout(200)
+
+    await transport2.publish('testing-channel', 'test')
+  }).waitForDone()
+
+  test('message should be encoded and decoded correctly when using JSON encoder', async ({
+    assert,
+    cleanup,
+  }) => {
+    const transport = new MqttTransport(
+      {
+        host: container.getHost(),
+        port: container.getPort(),
+      },
+      new JsonEncoder()
+    ).setId('bus')
+    cleanup(() => transport.disconnect())
+
+    const data = { test: 'test' }
+
+    await transport.subscribe('testing-channel', (payload) => {
+      assert.deepEqual(payload, data)
+    })
+
+    await setTimeout(200)
+
+    await transport.publish('testing-channel', data)
+  })
+
+  test('send binary data', async ({ assert, cleanup }, done) => {
+    assert.plan(1)
+
+    class BinaryEncoder implements TransportEncoder {
+      encode(message: TransportMessage<any>) {
+        return Buffer.from(JSON.stringify(message))
+      }
+
+      decode(data: string | Buffer) {
+        const buffer = Buffer.isBuffer(data) ? data : Buffer.from(data, 'binary')
+        return JSON.parse(buffer.toString())
+      }
+    }
+
+    const transport1 = new MqttTransport(
+      { host: container.getHost(), port: container.getMappedPort(1883) },
+      new BinaryEncoder()
+    ).setId('bus1')
+
+    const transport2 = new MqttTransport(
+      { host: container.getHost(), port: container.getMappedPort(1883) },
+      new BinaryEncoder()
+    ).setId('bus2')
+
+    cleanup(() => {
+      transport1.disconnect()
+      transport2.disconnect()
+    })
+
+    const data = ['foo', '👍']
+
+    await transport1.subscribe('testing-channel', (payload) => {
+      assert.deepEqual(payload, data)
+      done()
+    })
+
+    await setTimeout(200)
+    await transport2.publish('testing-channel', data)
+  }).waitForDone()
+})