Azure Event Hubs

Subscribes to an Azure Event Hub and forwards each event to the pipeline. Any publisher speaking the Event Hubs protocol works, including Microsoft Fabric Real-Time Intelligence eventstreams through their Event Hubs-compatible endpoint.

When to use it

• Your telemetry already lands in Azure — from Azure IoT Hub, which exposes an Event Hubs-compatible endpoint, from a Fabric eventstream, or from producers writing to a hub directly.
• You want the buffering and fan-out an event broker gives you between the devices and the scene.

Install

npm install @azure/event-hubs

Configuration

sources: [
    {
        kind: "azure-event-hub",
        config: {
            connection_string: process.env.EVENT_HUB_CONNECTION_STRING!,
            consumer_group: "livelink",
        },
    },
],

Option	Type	Default	Description
connection_string	string	—	Required. Usually embeds an EntityPath= segment
consumer_group	string	$Default	The consumer group to read with
event_hub_name	string	—	Only needed when the connection string does not embed EntityPath=

A connection string looks like this:

Endpoint=sb://xxxx.servicebus.windows.net/;SharedAccessKeyName=key_xxxx;SharedAccessKey=xxxx;EntityPath=xxxx

It is a credential — keep it in the environment, never in the source. The key is redacted before the transport logs anything.

Use a dedicated consumer group

Consumer groups are independent views of the same stream. Giving the ingestion its own means it never competes with whatever else already reads the hub, and each reader keeps its own position.

What arrives

{
    channel: "3",                              // the PARTITION ID — see below
    payload: { deviceId: "sensor-14", temperature: 21.4 },
    received_at: Date,
    source_timestamp: Date,                    // enqueuedTimeUtc
    metadata: {
        transport: "azure-event-hub",
        partition_id: "3",
        offset: "12345",
        sequence_number: 987,
    },
}

Event Hubs delivers bodies already deserialized. Non-object bodies (a raw string, a number) are logged and skipped, so the pipeline always sees a stable shape.

Never select on channel here

This is the one thing that will otherwise cost you a day.

Event Hubs has no per-message routing key. Unlike an MQTT topic, a partition is a load-balancing artifact: the same logical device can land on any partition, and the assignment can change. A mapping that selects channel: "3" works until it silently stops.

Select on the payload instead:

const temperatureMapping: EventMapping = {
    // NOT channel — the payload says what this event is.
    when: event => (event.payload as { temperature?: unknown }).temperature !== undefined,

    entities: { byName: "sensor-{id}" },

    updates: event => {
        const { deviceId, temperature } = event.payload as { deviceId: string; temperature: number };
        return { id: deviceId, update: { local_transform: { scale: [1, temperature / 50, 1] } } };
    },
};

The partition id, offset and sequence number remain available in metadata — useful for diagnostics, not for routing.

Positions and checkpointing

The transport reads from the latest position and does not checkpoint. It picks up events that arrive after it connects, and a restart does not replay what it missed.

For driving a live scene that is usually what you want: on reconnecting you care about the current state of the plant, not the backlog. It does mean the transport is not a fit for anything that must not miss an event.

If you need durable positions, own the client yourself and point it at the ingestion — it is an EventSink:

import { EventHubConsumerClient } from "@azure/event-hubs";
import { BlobCheckpointStore } from "@azure/eventhubs-checkpointstore-blob";

const checkpoint_store = new BlobCheckpointStore(containerClient);
const client = new EventHubConsumerClient(consumer_group, connection_string, checkpoint_store);

client.subscribe({
    processEvents: async (events, context) => {
        for (const event of events) {
            await ingestion.ingest({
                channel: context.partitionId,
                payload: event.body,
                source_timestamp: event.enqueuedTimeUtc,
            });
        }
        await context.updateCheckpoint(events[events.length - 1]);
    },
    processError: async error => console.error(error),
});

The sequence_number and offset carried in each event's metadata are exactly what such a store records. See writing your own transport for the general shape of this.

Ordering

Event Hubs guarantees ordering within a partition, not across them. Two events about the same device are ordered only if the producer gives them the same partition key.

For scene ingestion this rarely matters — you are writing current state, and the pipeline skips redundant writes anyway. It does matter if a mapping accumulates rather than overwrites (integrating a rate into an angle, say). Key such a stream by device at the producer.