app.eventHub.js

/*
 * Microsoft Sample Code - Copyright (c) 2020 - Licensed MIT
 */

const { EventHubConsumerClient } = require('@azure/event-hubs')
const { convertIotHubToEventHubsConnectionString } = require('./app.eventHub.connectionString')

class EventHubReader {
  constructor (iotHubConnectionString, consumerGroup) {
    this.iotHubConnectionString = iotHubConnectionString
    this.consumerGroup = consumerGroup
  }

  async startReadMessage (startReadMessageCallback) {
    try {
      const eventHubConnectionString = await convertIotHubToEventHubsConnectionString(this.iotHubConnectionString)
      const consumerClient = new EventHubConsumerClient(this.consumerGroup, eventHubConnectionString)
      consumerClient.subscribe({
        processEvents: (events, context) => {
          for (let i = 0; i < events.length; ++i) {
            startReadMessageCallback(
              events[i].body,
              events[i].enqueuedTimeUtc,
              events[i].systemProperties['iothub-connection-device-id'])
          }
        },
        processError: (err, context) => {
          console.error(err.message || err)
        }
      })
    } catch (ex) {
      console.error(ex.message || ex)
    }
  }

  // Close connection to Event Hub.
  async stopReadMessage () {
    const disposeHandlers = []
    this.receiveHandlers.forEach((receiveHandler) => {
      disposeHandlers.push(receiveHandler.stop())
    })
    await Promise.all(disposeHandlers)

    this.consumerClient.close()
  }
}

module.exports = EventHubReader