Skip to main content

Crate azure_messaging_eventhubs

Crate azure_messaging_eventhubs 

Source
Expand description

§Azure Event Hubs client library for Rust

Azure Event Hubs is a big data streaming platform and event ingestion service from Microsoft. For more information about Event Hubs see this link.

The Azure Event Hubs client library allows you to send single events or batches of events to an event hub and consume events from an event hub.

Source code | Package (crates.io) | API reference documentation | Product documentation

Migrating from the community azeventhubs crate? See the migration guide.

§Getting started

§Install the package

Install the Azure Event Hubs client library for Rust with Cargo:

cargo add azure_messaging_eventhubs

§Prerequisites

If you use the Azure CLI, replace <your-resource-group-name>, <your-eventhubs-namespace-name>, and <your-eventhub-name> with your own, unique names:

Create an Event Hubs Namespace:

az eventhubs namespace create --resource-group <your-resource-group-name> --name <your-eventhubs-namespace-name> --sku Standard

Create an Event Hub Instance:

az eventhubs eventhub create --resource-group <your-resource-group-name> --namespace-name <your-eventhubs-namespace-name> --name <your-eventhub-name>

§Install dependencies

Add the following crates to your project:

cargo add azure_identity tokio

§Authenticate the client

In order to interact with the Azure Event Hubs service, you’ll need to create an instance of the ProducerClient or the ConsumerClient. You need an event hub namespace host URL (which you may see as serviceBusEndpoint in the Azure CLI response when creating the Even Hubs Namespace), an Event Hub name (which you may see as name in the Azure CLI response when crating the Event Hub instance), and credentials to instantiate a client object.

The example shown below uses a DeveloperToolsCredential, which is appropriate for most local development environments. Additionally, we recommend using a managed identity for authentication in production environments. You can find more information on different ways of authenticating and their corresponding credential types in the Azure Identity documentation.

The DeveloperToolsCredential will automatically pick up on an Azure CLI authentication. Ensure you are logged in with the Azure CLI:

az login

Instantiate a DeveloperToolsCredential to pass to the client. The same instance of a token credential can be used with multiple clients if they will be authenticating with the same identity.

§Create an Event Hubs message producer and send an event

use azure_identity::DeveloperToolsCredential;
use azure_messaging_eventhubs::ProducerClient;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let host = "<EVENTHUBS_HOST>";
    let eventhub = "<EVENTHUB_NAME>";

    // Create new credential
    let credential = DeveloperToolsCredential::new(None)?;

    // Create and open a new ProducerClient
    let producer = ProducerClient::builder()
        .open(host, eventhub, credential.clone())
        .await?;

    producer.send_event(vec![1, 2, 3, 4], None).await?;

    Ok(())
}

§Key concepts

An Event Hub namespace can have multiple Event Hub instances. Each Event Hub instance, in turn, contains partitions which store events.

Events are published to an Event Hub instance using an event publisher. In this package, the event publisher is the ProducerClient

Events can be consumed from an Event Hub instance using an event consumer.

Consuming events is done using an EventReceiver, which can be opened from the ConsumerClient. This is useful if you already known which partitions you want to receive from.

More information about Event Hubs features and terminology can be found in the Event Hubs features documentation.

§Examples

Additional examples for various scenarios can be found on in the examples directory in our GitHub repo for Event Hubs.

§Open an Event Hubs message producer on an Event Hub instance

use azure_identity::DeveloperToolsCredential;
use azure_messaging_eventhubs::ProducerClient;

async fn open_producer_client() -> Result<ProducerClient, Box<dyn std::error::Error>> {
    let host = "<EVENTHUBS_HOST>";
    let eventhub = "<EVENTHUB_NAME>";

    let credential = DeveloperToolsCredential::new(None)?;

    let producer = ProducerClient::builder()
        .open(host, eventhub, credential.clone())
        .await?;

    Ok(producer)
}

§Send events

There are two mechanisms used to send events to an Event Hub instance. The first directly sends individual messages to the Event Hub, the second uses a “batch” operation to send multiple messages in a single network request to the service.

§Send events directly to the Event Hub
use azure_messaging_eventhubs::ProducerClient;

async fn send_events(producer: &ProducerClient) -> Result<(), Box<dyn std::error::Error>> {
    producer.send_event(vec![1, 2, 3, 4], None).await?;

    Ok(())
}
§Send events using a batch operation
use azure_messaging_eventhubs::ProducerClient;

async fn send_events(producer: &ProducerClient) -> Result<(), Box<dyn std::error::Error>> {
    let batch = producer.create_batch(None).await?;
    assert_eq!(batch.len(), 0);
    assert!(batch.try_add_event_data(vec![1, 2, 3, 4], None)?);

    let res = producer.send_batch(batch, None).await;
    assert!(res.is_ok());

    Ok(())
}

§Send events with the buffered producer

BufferedProducerClient accepts single events and publishes them in the background. The client groups the events into batches for each partition, and one worker for each partition sends them. This gives a higher throughput than ProducerClient, because the caller does not wait for each send.

The client reports the outcome of each batch through handlers. A handler for failed batches is required, because a send failure arrives after the enqueue call already returned.

use azure_identity::DeveloperToolsCredential;
use azure_messaging_eventhubs::BufferedProducerClient;

async fn buffered_publish() -> Result<(), Box<dyn std::error::Error>> {
    let host = "<EVENTHUBS_HOST>";
    let eventhub = "<EVENTHUB_NAME>";
    let credential = DeveloperToolsCredential::new(None)?;

    let producer = BufferedProducerClient::builder()
        .with_on_send_succeeded(|context| async move {
            println!(
                "The service accepted {} events on partition {}.",
                context.events.len(),
                context.partition_id
            );
        })
        .with_on_send_failed(|context| async move {
            eprintln!(
                "{} events failed on partition {}: {}",
                context.events.len(),
                context.partition_id,
                context.error
            );
        })
        .open(host, eventhub, credential.clone())
        .await?;

    for index in 0..1000 {
        producer.enqueue_event(format!("event {index}"), None).await?;
    }

    producer.close().await?;
    Ok(())
}
§Route events to a partition

Give a partition ID to send an event to one partition. Give a partition key to send every event with that key to the same partition. Set at most one of the two; the client rejects a request that sets both. When you set neither, the client assigns the partitions in round-robin order.

use azure_messaging_eventhubs::{BufferedProducerClient, EnqueueEventOptions};

async fn route_events(
    producer: &BufferedProducerClient,
) -> Result<(), Box<dyn std::error::Error>> {
    producer
        .enqueue_event(
            "to partition 0",
            Some(EnqueueEventOptions {
                partition_id: Some("0".to_string()),
                ..Default::default()
            }),
        )
        .await?;

    producer
        .enqueue_event(
            "grouped by key",
            Some(EnqueueEventOptions {
                partition_key: Some("customer-17".to_string()),
                ..Default::default()
            }),
        )
        .await?;

    Ok(())
}
§Flush and shut down

flush sets a barrier. It completes once every event that the client accepted before the call reaches a terminal outcome. An event that arrives after the barrier does not delay the call.

close sends the buffered events and then shuts the client down. abort shuts the client down at once and abandons the buffered events.

use azure_messaging_eventhubs::BufferedProducerClient;

async fn flush_and_close(
    producer: &BufferedProducerClient,
) -> Result<(), Box<dyn std::error::Error>> {
    producer.enqueue_event("an event", None).await?;

    // Wait for the events that the client already accepted.
    producer.flush().await?;
    println!("{} events are still buffered.", producer.total_buffered_event_count());

    // Send what is left, then shut down.
    producer.close().await?;
    Ok(())
}
§Trade-offs of buffered publishing
  • A successful enqueue means only that the local buffer accepted the event. It does not mean that Event Hubs accepted the event.
  • The process loses the buffered events if it stops before a flush or a close. Call flush or close when the delivery of the buffered events matters.
  • A send failure arrives after the enqueue call already returned, through the failure handler.
  • Buffering gives a higher throughput, but the latency of one event is less predictable.
  • Use ProducerClient when the application needs the result of each send.

§Open an Event Hubs message consumer on an Event Hub instance

use azure_identity::DeveloperToolsCredential;
use azure_messaging_eventhubs::ConsumerClient;

async fn open_consumer_client() -> Result<ConsumerClient, Box<dyn std::error::Error>> {
    let host = "<EVENTHUBS_HOST>".to_string();
    let eventhub = "<EVENTHUB_NAME>".to_string();

    let credential = DeveloperToolsCredential::new(None)?;

    let consumer = azure_messaging_eventhubs::ConsumerClient::builder()
        .open(&host, eventhub, credential.clone())
        .await?;

    Ok(consumer)
}

§Receive events

The following example shows how to receive events from partition 0 on an Event Hubs instance.

It assumes that the caller has provided a consumer client which will be used to receive events.

Each message receiver can only receive messages from a single Event Hubs partition

use futures::stream::StreamExt;
use azure_messaging_eventhubs::{
    ConsumerClient, OpenReceiverOptions, StartLocation, StartPosition,
};

// By default, an event receiver only receives new events from the event hub. To receive events from earlier, specify
// a `start_position` which represents the position from which to start receiving events.
// In this example, events are received from the start of the partition.
async fn receive_events(client: &ConsumerClient) -> Result<(), Box<dyn std::error::Error>> {
    let message_receiver = client
        .open_receiver_on_partition(
            "0".to_string(),
            Some(OpenReceiverOptions {
                start_position: Some(StartPosition {
                    location: StartLocation::Earliest,
                    ..Default::default()
                }),
                ..Default::default()
            }),
        )
        .await?;

    let mut event_stream = message_receiver.stream_events();

    while let Some(event_result) = event_stream.next().await {
        match event_result {
            Ok(event) => {
                // Process the received event
                println!("Received event: {:?}", event);
            }
            Err(err) => {
                // Handle the error
                eprintln!("Error receiving event: {:?}", err);
            }
        }
    }

    Ok(())
}

§Troubleshooting

§General

When you interact with the Azure Event Hubs client library using the Rust SDK, errors returned by the service are returned as azure_core::Error values using ErrorKind::Other which are azure_messaging_eventhubs::Error values.

§Logging

The Event Hubs SDK client uses the tracing package to enable diagnostics.

The crate does not set custom tracing target= values. Events are emitted on the standard tracing module-path targets, which match the module that produced them (for example, azure_messaging_eventhubs::common::recoverable::connection). You can filter events by module path with RUST_LOG or an EnvFilter. For example, RUST_LOG=azure_messaging_eventhubs=debug enables debug-and-above for the whole crate, while RUST_LOG=azure_messaging_eventhubs::common::recoverable=trace narrows tracing to the connection recovery path.

Diagnostic values are attached as structured fields (connection_id, partition_id, url, and similar) rather than being interpolated into the message text, so they can be captured and queried by structured subscribers. Credentials (tokens, shared-access keys, and connection strings) are never logged. Event payloads and message bodies are redacted: the only site that logs a message does so at trace, and its body and application properties are stripped by SafeDebug. Enabling the azure_core debug cargo feature turns that redaction off, so avoid it in production when event contents are sensitive.

Events follow a consistent level policy so you can pick the verbosity you need:

  • error - terminal or fatal failures that abort an operation, plus the exit of a long-lived background task.
  • warn - recoverable or anomalous-but-handled conditions, such as a send being rejected, modified, or released, attach failures, retry exhaustion, a recovery action being required, an etag mismatch, a missing management key, or an unauthorized fast-fail.
  • info - lifecycle success milestones, such as a connection or link opening, a link attaching, a receiver attaching on a partition, recovery completing, or partition ownership being claimed.
  • debug - per-operation bookkeeping, error classification decisions, retry chatter, and internal map updates.
  • trace - very-high-frequency or per-message detail, including the hot send path.

§Contributing

See the CONTRIBUTING.md for details on building, testing, and contributing to these libraries.

This project welcomes contributions and suggestions. Most contributions require you to agree to a Contributor License Agreement (CLA) declaring that you have the right to, and actually do, grant us the rights to use your contribution. For details, visit https://opensource.microsoft.com/cla/.

When you submit a pull request, a CLA-bot will automatically determine whether you need to provide a CLA and decorate the PR appropriately (e.g., label, comment). Simply follow the instructions provided by the bot. You will only need to do this once across all repos using our CLA.

This project has adopted the Microsoft Open Source Code of Conduct. For more information see the Code of Conduct FAQ or contact opencode@microsoft.com with any additional questions or comments.

§Reporting security issues and security bugs

Security issues and bugs should be reported privately, via email, to the Microsoft Security Response Center (MSRC) secure@microsoft.com. You should receive a response within 24 hours. If for some reason you do not, please follow up via email to ensure we received your original message. Further information, including the MSRC PGP key, can be found in the Security TechCenter.

§License

Azure SDK for Rust is licensed under the MIT license.

Re-exports§

pub use error::EventHubsError;
pub use error::Result;

Modules§

builders
Builders for producer client and consumer client.
error
Error types for the Event Hubs service.
models
Types sent to and received from the Event Hubs service.
processor
Event Hubs processor related types.

Structs§

BufferedProducerClient
A producer client that buffers events and publishes them in the background.
ConnectionString
A parsed Event Hubs connection string.
ConsumerClient
A client that can be used to receive events from an Event Hub.
EnqueueEventOptions
Options for BufferedProducerClient::enqueue_event and BufferedProducerClient::enqueue_events.
EventDataBatch
Represents a collections of event data that can be sent to an Event Hubs instance in one operation.
EventDataBatchOptions
Represents the options that can be set when creating an EventDataBatch. The options include the maximum size of the batch, the partition key, and the partition ID.
EventProcessor
Represents the event processor responsible for processing events from Event Hub partitions.
EventReceiver
A message receiver that can be used to receive messages from an Event Hub.
OpenReceiverOptions
Represents the options for receiving events from an Event Hub.
ProducerClient
A client that can be used to send events to an Event Hubs instance.
RetryOptions
Options for configuring exponential backoff retry behavior.
SendBatchFailedContext
Reports that the service did not durably accept a batch of events.
SendBatchOptions
Represents the options that can be set when submitting a batch of event data.
SendBatchSucceededContext
Reports that the service accepted a batch of events.
SendEventOptions
Options used when sending an event to an Event Hub.
SendMessageOptions
Options used when sending an AMQP message to an Event Hub. The SendMessageOptions can be used to specify the partition to which the message should be sent. If the partition is not specified, the Event Hub will automatically select a partition.
StartPosition
Represents the starting position of a consumer when receiving events from an Event Hub.

Enums§

ProcessorStrategy
Represents the strategy for load balancing event processing.
StartLocation
Represents the starting position of a consumer when receiving events from an Event Hub.

Traits§

CheckpointStore
Trait representing a checkpoint store.