bus   

Note: bus is deprecated and no longer maintained! Use Eshu instead.

bus provides a high-level way to work with MQTT. Under the hood it uses the excellent MQTT.js module.

The module is written in TypeScript and has the required declaration files included. It can also be used in vanilla JS.

bus is currently in active development and breaking changes may be introduced without warning! This will remain true at least as long as the module is in alpha (e.g. 0.0.1-alpha.1)

• Install
• Example
• API
• Inspired by
• License

Install

bus can be installed using npm:

npm install @art-of-coding/bus

Example

This basic example shows how to use the basic functionality.

import { Bus } from '@art-of-coding/bus'

// Create a new Bus instance with the given client ID and broker URL
const bus = Bus.create('my-client-id', 'mqtt://localhost')

// Set up a topic pattern identified by the label `configUpdated`
// `+deviceId` and '#keys' are parameterized wildcards
bus.setPattern('configUpdated', 'devices/+deviceId/config/#keys')

// Add a listener for `configUpdated`
// This function is called each time a publish packet is received on the
// topic identified by the label
bus.on('configUpdated', function (packet) {
  // We can use packet.params to access the pattern parameters
  console.log(`Device ${packet.params.deviceId} config updated:
  ${packet.params.keys.join('/')} is now ${packet.payload})`)
})

// Now, connect to the broker
bus.connect().then(() => {
  console.log('connected to broker')

  // Subscribe to the topic with label `configUpdated`
  return bus.subscribe('configUpdated')
}).then(() => {
  console.log('subscribed to configUpdated')

  // Publish a config update on the same pattern
  // This will publish the word 'localhost' on topic devices/my-device/config/http/host
  return bus.publish('configUpdated', { deviceId: 'my-device', keys: [ 'http', 'host' ] }, 'localhost')
}).then(() => {
  console.log('update published')

  // We're done, so we'll close the connection
  return bus.end()
}).then(() => {
  console.log('disconnected from broker')
}).catch(err => {
  // One of the steps above threw an error
  console.error('Got an error:')
  console.error(err)
})

API

• Bus.create()
• bus.getId()
• bus.isAvailable()
• bus.getStatus()
• bus.getStatusError()
• bus.onStatusChange()
• bus.setPattern()
• bus.removePattern()
• bus.on()
• bus.once()
• bus.removeListener()
• bus.removeAllListeners()
• bus.connect()
• bus.end()
• bus.subscribe()
• bus.unsubscribe()
• bus.publish()

Bus.create (clientId: string, url: string, opts?: IBusOptions): Bus

Creates a new Bus instance.

• clientId: The instance's client ID
• url: The broker URL (e.g. mqtt://localhost:3306)
• opts: MQTT.js options (see here)
  • clientId is always overwritten by the Id given as the first parameter

import { Bus } from '@art-of-coding/bus'
// or
const Bus = require('@art-of-coding/bus').Bus

// create a Bus instance
const bus = Bus.create('my-client-id', 'mqtt://localhost')

bus.getId (): string

Get the client ID for the instance.

bus.isAvailable (): boolean

Check to see if the bus is available and can be used.

bus.getStatus (): Status

Get the instance status. The return value is an integer and corresponds to the Status enum:

| Status | Code | |--------------|------| | READY | 0 | | CONNECTING | 1 | | CONNECTED | 2 | | RECONNECTING | 3 | | OFFLINE | 4 | | CLOSED | 5 | | ERROR | 6 |

bus.getStatusError (): Error|null

Get the last error, or null if no error occured yet. Is usually only available when bus.getStatus() returns ERROR.

bus.onStatusChange (fn: (status: Status, error: Error) => void): void

Set a method which will be called each time the bus's status changes.

• fn: The method to call

bus.onStatusChange(function (status: Status, error: Error) {
  const withOrWithout = error ? 'with' : 'without'
  console.log(`Status changed to ${status} (${withOrWithout} error)`)
})

bus.setPattern (label: string, pattern: string): Bus

Set a topic pattern with the given label. When set, a pattern can be referenced by using its label. For more on topic patterns, see the mqtt-regex readme.

• label: The pattern's label
• pattern: The topic pattern

Notes:

• This method will throw if the label already exists or the pattern is invalid
• This method returns the bus instance, so calls can be chained

// set a topic with a single named wildcard - `one`
bus.setPattern('myLabel1', 'my/topic/+one')

// set a topic with a multi named wildcard - `keys`
bus.setPattern('myLabel2', 'my/topic/#keys')

bus.removePattern (label: string): Bus

Removes a topic pattern with the given label.

• label: The label to remove

Notes:

• This method will throw if the label doesn't exist
• This method returns the bus instance, so calls can be chained

bus.on (label: string, fn: (packet: IPacket) => void): Bus

Add a listener for the given label. This listener is called each time a message is received on a topic that matches the pattern accompanying the label.

• label: The label
• fn: The method to call

Notes:

• bus is not an EventEmitter! Therefore, not all methods for regular event emitters will work!
• This method will throw if the label doesn't exist
• This method returns the bus instance, so calls can be chained

bus.once (label: string, fn: (packet: IPacket) => void): Bus

Adds a once listener for the given label. The listener is called the next time a message that matches the topic pattern is received, then it is removed.

• label: The label
• fn: The method to call

Notes:

• bus is not an EventEmitter! Therefore, not all methods for regular event emitters will work!
• This method will throw if the label doesn't exist
• This method returns the bus instance, so calls can be chained

bus.removeListener (label: string, fn: (packet: IPacket) => void): Bus

Removes the listener method for the given label. The method will no longer be called.

• label: The label
• fn: The method to call

Notes:

• bus is not an EventEmitter! Therefore, not all methods for regular event emitters will work!
• This method will throw if the label doesn't exist
• This method returns the bus instance, so calls can be chained

bus.removeAllListeners (label?: string): Bus

Removes all listeners. When label is set, only that label's listeners will be removed. Otherwise, all listeners (for all labels) will be removed.

• label: The label

Notes:

• bus is not an EventEmitter! Therefore, not all methods for regular event emitters will work!
• This method will throw if the label doesn't exist
• This method returns the bus instance, so calls can be chained

bus.connect(): Promise

Connect to the broker.

bus.connect().then(() => {
  console.log('connected!')
}).catch(err => {
  console.error(err)
})

bus.end (force: boolean = false): Promise

Closes the connection to the broker. If force is true, the connection will be closed without waiting for in-flight packages fo be acked.

• force: Don't wait for in-flight messages to be acked (default false)

bus.subscribe (label: string, opts?: IClientSubscribeOptions): Promise

Subscribe to the topic pattern identified by label.

• label: The label to subscribe to
• opts:
  • qos: qos subscription level, (default 0)

bus.unsubscribe (label: string, removeListeners: boolean = false): Promise

Unsubscribes from the topic pattern identified by label. If removeListeners is true, all added listeners for the label will be removed as well.

• label: The label to unsubscribe from
• removeListeners: Remove all listeners for the label (default false)

bus.publish (label: string, params: any, payload: any, opts?: IClientPublishOptions): Promise

Published the payload on the topic pattern identified by label, with the parameters params. If you have no parameters, use an empty object ({}) instead.

• label: The label to publish
• params: The topic parameter values
• payload: The payload to publish
• opts:
  • qos: QoS level (default 0)
  • retain: retain flag (default false)
  • dup: mark as duplicate (default false)

Notes:

• params will usually be an object with keys corresponding to wildcard names
• payload automatically stringifies JSON and converts numbers to string

// Set a pattern `myLabel` with one parameter (`+name`)
bus.setPattern('myLabel', 'topics/+name')

// Make the parameters object
const params = { name: 'my-topic' }
// Set the payload
const payload = 'Hello, topic!'

// Publish the payload on the label with the given parameters
bus.publish('myLabel', params, payload)

Inspired by

• MQTT.js, which this module uses under the hood
• mqtt-regex, whose support for parameterized topics is integrated

License

Copyright 2017 Michiel van der Velde.

This software is licensed under the MIT License.