EventHubsStore — Azure Event Hubs

EventHubsStore — Azure Event Hubs Overview Azure Event Hubs is a managed, partitioned event-ingestion / event-streaming service — Azure's Kafka-like offering. There is no query language and no SQL-shaped access pattern: producers send event batches to a hub, and consumers receive them through a consumer group. Because none of that fits the SQL-shaped Dialect interface, EventHubsStore implements the minimal NoSqlStore marker interface (src/nosql/store.ts) — connection lifecycle plus a getClient() escape hatch — and exposes Event Hubs's own operations directly rather than forcing them into a query(sql) shape. Identity: Property | Value | --------- | --------------------- | name | 'eventhubs' | library | '@azure/event-hubs' | The store is built on @azure/event-hubs: an EventHubProducerClient sends events (sendBatch) and an EventHubConsumerClient receives them (subscribe). Unlike the AWS stores, there is no command-object pattern — the driver exposes client methods directly. Lazy loading — not a hard dependency @azure/event-hubs is not a hard dependency of this package. The driver is loaded lazily via require() inside connect(), rather than a top-level import. Importing this module therefore does not require the driver to be installed — it is only needed when an Event Hubs store is actually connected. Injected client (module) EventHubsStoreOptions accepts a pre-built client — a module-shaped object exposing the two client constructors (EventHubProducerClient, EventHubConsumerClient), typed as EventHubsModuleLike. When provided, connect() uses it directly and does not require the driver. This is how the test suite injects mocks (no driver, no network). Message encoding publish() accepts a string or Buffer (used as the event body unchanged) or any other JSON-serializable value (JSON.stringifyd to a string body). Event Hubs itself AMQP-encodes the body on the wire. No createTopic Event Hubs namespaces / hubs are provisioned via the Azure management plane (ARM / portal / CLI), not the data-plane SDK, so this store intentionally does not expose a createTopic (or any create method). Connection Build a store from connection options and call connect(): All connection options are optional: Option | Type | Purpose | ------------------ | --------------------- | ------------------------------------------------------------------------------------------------- | connectionString | string | Event Hubs namespace connection string. | eventHubName | string | Default event hub (entity) name used when publish / subscribe omit one. When set, connect() eagerly builds the default producer. | consumerGroup | string | Consumer group used by subscribe(). Defaults to '$Default'. | client | EventHubsModuleLike | A pre-built module (or mock) exposing the two client constructors. When set, the driver is not required. | Injected-client form Supply your own module (or a mock) to bypass driver loading: Methods The store caches one EventHubProducerClient per event hub name (lazily, on first use) and constructs a fresh EventHubConsumerClient per subscribe(). Driver failures are wrapped in a DatabaseError (preserving the original error); using a method before connect() (or after disconnect()) throws a ConnectionError. Lifecycle Method | Signature | Behavior | ------------- | ---------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------- | connect | connect(): Promise<void> | Idempotent (returns early if already connected). Uses an injected module if provided, otherwise lazy-requires @azure/event-hubs. Builds the default producer when eventHubName is set. Wraps failures in ConnectionError. | disconnect | disconnect(): Promise<void> | Closes all tracked consumer subscriptions and all cached producers, then clears state. | isConnected | isConnected(): boolean | true only when connected and the module is present. | getClient | getClient(): EventHubProducerLike| Returns the default EventHubProducerClient (for the store's eventHubName). Throws ConnectionError if not connected. | Producing Method | Signature | Driver call | Behavior | --------- | ----------------------------------------------------------------- | ---------------------------------- | ---------------------------------------------------------------------------------------------------------------- | publish | publish(eventHubName: string, message: EventHubsMessage): Promise<void> | producer.sendBatch([{ body }]) | Resolves (and caches) a producer for eventHubName, encodes message to a body, and sends it as a single-event batch. Returns nothing. | Consuming Method | Signature | Driver call | Behavior | ----------- | ------------------------------------------------------------------------------------------------------------------- | ---------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | subscribe | subscribe(eventHubName: string, handler: (event: any) => void \| Promise<void>, options?: EventHubsSubscribeOptions): Promise<EventHubsSubscription> | consumer.subscribe() | Constructs a new EventHubConsumerClient(consumerGroup, connectionString, eventHubName) and registers a processEvents handler that invokes handler once per received event (processError is swallowed). Returns a subscription whose stop() closes both the subscription handle and the consumer client. Tracked so disconnect() closes it. | Example Verification status Unit / mock-verified only. The tests in tests/nosql/eventhubs.test.ts are fully mock-driven: a fake @azure/event-hubs module (exposing EventHubProducerClient and EventHubConsumerClient constructors that build jest-spy instances) is injected via EventHubsStoreOptions.client. The real driver is not installed, and there is no live Event Hubs access and no network in the test run. What this proves: connect() builds the default producer once when eventHubName is set, and getClient() returns a producer exposing sendBatch. publish sends a single-event batch with a string body, and JSON-encodes object messages. subscribe constructs an EventHubConsumerClient(consumerGroup, connectionString, eventHubName), wires the handler through processEvents (invoked once per event), and defaults the consumer group to '$Default'. disconnect() closes both producers and consumer subscriptions. Error handling: sendBatch failures wrapped in DatabaseError; ConnectionError from getClient() before connect(). What this does not prove: live execution against real Azure Event Hubs. The client construction and method-call shape are verified against the driver's documented contract, but end-to-end execution over the wire has not been exercised here. Related reading All data stores — the full catalogue, grouped by purpose Database types — where this sits among the 22 categories