Implementing Event Hubs SDK over AMQP 1.0

Cargo.toml

@@ -3,4 +3,7 @@ members = [
   "sdk/*",
   "./eng/test/mock_transport"
 ]
-resolver = "2"
+
+exclude = [
+  "sdk/messaging_eventhubs", # excluded because of wasm support
+]

sdk/messaging_eventhubs/CHANGELOG.md

@@ -0,0 +1,70 @@
+# Change log
+
+## 0.14.4
+
+- Removed the internal buffer from `EventStream`
+- Deprecates `cache_event_count` and `max_wait_time` fields in `ReadEventOptions`
+
+## 0.14.3
+
+- Reworked `EventHubConnection` to have more fine-grained locks
+- Changed `EventStream` error type to `azure_core::Error`
+- Added type alias `SingleConsumerEventStream` and `MultiConsumerEventStream`
+
+## 0.14.2
+
+- Exposed `TryAddError` to public
+
+## 0.14.1
+
+- Renamed the following constructor methods and marked the old methods as deprecated
+  - `from_connection_string()` -> `new_from_connection_string()`
+  - `from_namespace_and_credential()` -> `new_from_credential()`
+  - `from_namespace_and_named_key_credential()` -> `new_from_named_key_credential()`
+  - `from_namespace_and_sas_credential()` -> `new_from_sas_credential()`
+- Removed generic parameter `C` from `EventHubConnection`
+
+## 0.14.0
+
+- Fixed problem with `azure_identity` credentials
+- Added example showcasing how to work with `azure_identity` credentials
+
+## 0.14.0-alpha
+
+- Changed version number to follow that of `azure_core`
+- Changing visibility of the following to public
+  - `BasicRetryPolicy`,
+  - `mod authorization`,
+  - `EventHubTokenCredential`
+  - `SharedAccessCredential`
+  - `AzureNamedKeyCredential`
+  - `AzureSasCredential`
+- Added helper function `crate::authorization::build_connection_signature_authorization_resource()`
+- Added following convenience constructor methods to `EventHubConnection`, `EventHubProducerClient`, and `EventHubConsumerClient`
+  - `from_namespace_and_named_key_credential()`
+  - `from_namespace_and_sas_credential()`
+
+## 0.1.2
+
+### 0.1.2-beta
+
+- Changed visibility of `IntoAzureCoreError` to `pub(crate)` and restricted its impl to only foreign
+  error types. All other error type natively implements `Into<azure_core::error::Error>`
+
+### 0.1.2-alpha
+
+- Fixed a bug where `EventStream` is not `Send` because `ClosingBoxedFuture` misses `Send` in its
+  trait bounds
+- Changed visibility of struct `EventStream` to public
+- Changed visibility of trait `IntoAzureCoreError` to public
+- Updated dependency `azure_core` to `"0.13"`
+- Updated dependency `time`'s version to `"<=0.3.23"`, which is the latest version that supports
+  rust version 1.65
+
+## 0.1.1
+
+- Fixed wrong crate name in the example
+
+## 0.1.0
+
+- Initial release

sdk/messaging_eventhubs/Cargo.toml

@@ -0,0 +1,91 @@
+[package]
+name = "azeventhubs"
+version = "0.15.0"
+edition = "2021"
+description = "An unofficial AMQP 1.0 rust client for Azure Event Hubs"
+readme = "README.md"
+authors = ["Minghua Wu"]
+license = "MIT"
+repository = "https://github.com/minghuaw/azure-sdk-for-rust/tree/separate_eventhubs_crate/sdk/messaging_eventhubs"
+homepage = "https://github.com/minghuaw/azure-sdk-for-rust/tree/separate_eventhubs_crate/sdk/messaging_eventhubs"
+documentation = "https://docs.rs/azeventhubs"
+
+categories = ["api-bindings"]
+keywords = ["sdk", "azure", "eventhubs", "amqp", "cloud"]
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+time = { version = "0.3", features = ["macros"] }
+url = "2"
+uuid = { version = "1", features = ["v4"] }
+const_format = "0.2"
+thiserror = "1"
+futures-util = "0.3"
+log = "0.4"
+urlencoding = "2"
+base64 = "0.21"
+digest = "0.10"
+hmac = "0.12"
+sha2 = "0.10"
+mockall = "0.11"
+async-trait = "0.1"
+rand = "0.8"
+pin-project-lite = "0.2.9"
+serde = "1"
+
+# Azure dependencies
+azure_core = { path = "../core", version = "0.15" }
+
+# AMQP dependencies
+fe2o3-amqp = { version = "0.8.22" }
+fe2o3-amqp-types = "0.7.1"
+fe2o3-amqp-management = "0.2.2"
+fe2o3-amqp-cbs = "0.2"
+serde_amqp = { version = "0.5.9", features = ["derive", "time"] }
+fe2o3-amqp-ws = "0.4.0"
+
+[target.'cfg(not(target_arch = "wasm32"))'.dependencies]
+tokio = { version = "1", features = ["rt", "macros", "sync", "time", "net"] }
+tokio-util = "0.7"
+timer-kit = { version = "0.1", features = ["tokio"] }
+
+[target.'cfg(target_arch = "wasm32")'.dependencies]
+# Add dependency to getrandom to enable WASM support
+getrandom = { version = "0.2", features = ["js"] }
+tokio = { version = "1", features = ["rt", "macros"]}
+tokio-util = {version = "0.7.3"}
+fluvio-wasm-timer = "0.2"
+timer-kit = { version = "0.1", features = ["fluvio-wasm-timer"] }
+js-sys = "0.3"
+
+[target.'cfg(not(target_arch = "wasm32"))'.dev-dependencies]
+dotenv = "0.15"
+tokio = {version = "1.0", features = ["macros", "rt", "rt-multi-thread"]}
+env_logger = "0.10.0"
+fe2o3-amqp = { version = "0.8.21", features = ["log"] }
+bytes = "1"
+azure_identity = { path = "../identity" }
+criterion = { version = "0.5", features = ["async_tokio"] }
+
+[features]
+default = ["native-tls"]
+
+# CI tests related features
+test_e2e = [] # This enables tests under the `sdk/messaging_eventhubs/tests` directory
+test_integration = [] # TODO: this feature is not used for now
+
+rustls = ["fe2o3-amqp/rustls", "fe2o3-amqp-ws/rustls-tls-webpki-roots"]
+native-tls = ["fe2o3-amqp/native-tls", "fe2o3-amqp-ws/native-tls"]
+
+[[bench]]
+name = "bench_consumer_client_start_up"
+harness = false
+
+[[bench]]
+name = "bench_event_stream_start_up"
+harness = false
+
+[[bench]]
+name = "bench_event_stream"
+harness = false

sdk/messaging_eventhubs/README.md

@@ -0,0 +1,128 @@
+# azeventhubs
+
+An unofficial and experimental Azure Event Hubs client library for Rust.
+
+This crate follows a similar structure to that of the [Azure SDK for
+.Net](https://github.com/Azure/azure-sdk-for-net/tree/main/sdk/eventhub/Azure.Messaging.EventHubs),
+and thus it should be familiar to anyone who has used the dotnet SDK. This crate is still in
+development, and not all features are implemented yet.
+
+## Examples
+
+### Event Hub Producer Example
+
+```rust
+use azeventhubs::producer::{
+EventHubProducerClient, EventHubProducerClientOptions, SendEventOptions,
+};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let mut producer_client =
+        EventHubProducerClient::new_from_connection_string(
+            "<CONNECTION_STRING>", // Replace with your connection string
+            "<EVENT_HUB_NAME>".to_string(), // Replace with your hub name
+            EventHubProducerClientOptions::default()
+       ).await?;
+
+    let partition_ids = producer_client.get_partition_ids().await?;
+
+    let event = "Hello, world to first partition!";
+    let options = SendEventOptions::new().with_partition_id(&partition_ids[0]);
+    producer_client.send_event(event, options).await?;
+
+    producer_client.close().await?;
+
+    Ok(())
+}
+```
+
+### Event Hub Consumer Example
+
+```rust
+use futures_util::StreamExt;
+use azeventhubs::consumer::{EventHubConsumerClient, EventHubConsumerClientOptions, EventPosition, ReadEventOptions};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    // Create a consumer client
+    let mut consumer_client =
+        EventHubConsumerClient::new_from_connection_string(
+            EventHubConsumerClient::DEFAULT_CONSUMER_GROUP_NAME,
+            "<CONNECTION_STRING>", // Replace with your connection string
+            "<EVENT_HUB_NAME>".to_string(), // Replace with your hub name
+            EventHubConsumerClientOptions::default(),
+        ).await?;
+
+    let partition_ids = consumer_client.get_partition_ids().await?;
+    let starting_position = EventPosition::earliest();
+    let options = ReadEventOptions::default();
+
+    // Get a stream of events from the first partition
+    let mut stream = consumer_client
+        .read_events_from_partition(&partition_ids[0], starting_position, options)
+        .await?;
+
+    // Receive 30 events
+    let mut counter = 0;
+    while let Some(event) = stream.next().await {
+        let event = event?;
+        let body = event.body()?;
+        let value = std::str::from_utf8(body)?;
+        log::info!("{:?}", value);
+
+        log::info!("counter: {}", counter);
+        counter += 1;
+        if counter > 30 {
+            break;
+        }
+    }
+    // Close the stream
+    stream.close().await?;
+
+    // Close the consumer client
+    consumer_client.close().await?;
+
+    Ok(())
+}
+```
+
+## What is implemented and what is not?
+
+| Feature | Supported |
+| ------- | --------- |
+| Event Hub Connection | Yes |
+| Event Hub Producer | Yes |
+| Event Hub Consumer | Yes |
+| Partition Receiver | Yes |
+| Event Hub Buffered Producer | Not yet |
+| Event Hub Processor | Not yet |
+| Checkpoint Store | Not yet |
+
+## TLS Support
+
+Communication between a client application and an Azure Service Event Hub namespace is encrypted
+using Transport Layer Security (TLS). The TLS implementation is exposed to the user through the
+corresponding feature flags (please see the feature flag section below). The user should ensure
+either the `rustls` or `native-tls` feature is enabled, and one and only one TLS implementation
+must be enabled. Enabling both features is **not** supported and will result in an error.
+
+The `native-tls` feature is enabled by default, and it will use the `native-tls` crate to
+provide TLS support. The `rustls` feature will use the `rustls` crate and `webpki-roots` crate
+to provide TLS support.
+
+## Feature Flags
+
+This crate supports the following feature flags:
+
+| Feature | Description |
+| ------- | ----------- |
+| `default` | Enables "native-tls" feature |
+| `rustls` | Enables the use of the `rustls` crate for TLS support |
+| `native-tls` | Enables the use of the `native-tls` crate for TLS support |
+
+## WebAssembly Support
+
+WebAssembly is **NOT** supported yet.
+
+License: MIT