Make Rust SDK transport dependencies feature-specific

Author: daflyinbed

eventmesh-sdks/eventmesh-sdk-rust/Cargo.toml

@@ -42,15 +42,47 @@ rustdoc-args = ["--cfg", "docsrs"]
 # compiling a protocol implementation.
 default = []
 # Transport protocols (implemented incrementally).
-grpc = ["dep:tonic", "dep:prost", "dep:prost-types"]
+grpc = [
+    "dep:tonic",
+    "dep:prost",
+    "dep:prost-types",
+    "dep:tonic-build",
+    "dep:tokio",
+    "dep:tokio-util",
+    "dep:tracing",
+    "dep:rand",
+    "dep:uuid",
+    "dep:http",
+]
 # HTTP transport (producer, consumer, webhook middleware + built-in server).
-http = ["dep:reqwest", "dep:serde_urlencoded", "dep:axum", "dep:tower", "dep:tower-http"]
+http = [
+    "dep:reqwest",
+    "dep:serde_urlencoded",
+    "dep:axum",
+    "dep:tokio",
+    "dep:tokio-util",
+    "dep:tracing",
+    "dep:rand",
+    "dep:uuid",
+    "dep:bytes",
+    "dep:http",
+]
 # TCP transport (producer, consumer, native TCP wire protocol).
 # `tokio-util/codec` is required by the framed codec in
 # `src/transport/tcp/{codec,connection}.rs`; without it the crate fails to
 # compile when `tcp` is enabled without `grpc` (tonic pulls in `codec` only as
 # a side effect of the `grpc` feature).
-tcp = ["tokio-util/codec"]
+tcp = [
+    "dep:tokio",
+    "dep:tokio-stream",
+    "dep:tokio-util",
+    "tokio-util/codec",
+    "dep:futures",
+    "dep:tracing",
+    "dep:rand",
+    "dep:uuid",
+    "dep:bytes",
+]
 # Message models.
 cloud_events = ["dep:cloudevents", "dep:chrono"]
 # Convenience aggregate.
@@ -64,18 +96,18 @@ e2e = ["grpc", "http", "tcp", "cloud_events"]
 interop_e2e = ["e2e"]
 
 [dependencies]
-tokio = { version = "1", features = ["full"] }
-tokio-stream = { version = "0.1", features = ["net"] }
-tokio-util = "0.7"
-futures = "0.3"
+tokio = { version = "1", optional = true, features = ["full"] }
+tokio-stream = { version = "0.1", optional = true, features = ["net"] }
+tokio-util = { version = "0.7", optional = true }
+futures = { version = "0.3", optional = true }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 thiserror = "1"
-bytes = "1"
-rand = "0.8"
-uuid = { version = "1", features = ["v4"] }
-tracing = "0.1"
-http = "1"
+bytes = { version = "1", optional = true }
+rand = { version = "0.8", optional = true }
+uuid = { version = "1", optional = true, features = ["v4"] }
+tracing = { version = "0.1", optional = true }
+http = { version = "1", optional = true }
 # gRPC transport.
 tonic = { version = "0.12", optional = true }
 prost = { version = "0.13", optional = true }
@@ -86,12 +118,10 @@ chrono = { version = "0.4", optional = true, default-features = false, features
 # HTTP transport.
 reqwest = { version = "0.12", optional = true, features = ["json"] }
 serde_urlencoded = { version = "0.7", optional = true }
-tower = { version = "0.5", optional = true }
-tower-http = { version = "0.6", optional = true }
 axum = { version = "0.7", optional = true }
 
 [build-dependencies]
-tonic-build = "0.12"
+tonic-build = { version = "0.12", optional = true }
 
 [dev-dependencies]
 tokio = { version = "1", features = ["full", "test-util"] }
@@ -110,16 +140,36 @@ name = "grpc_consumer"
 path = "examples/grpc/consumer.rs"
 required-features = ["grpc"]
 
+[[example]]
+name = "grpc_webhook_consumer"
+path = "examples/grpc/webhook_consumer.rs"
+required-features = ["grpc"]
+
 [[example]]
 name = "grpc_producer_cloud_events"
 path = "examples/grpc/producer_cloud_events.rs"
 required-features = ["grpc", "cloud_events"]
 
+[[example]]
+name = "grpc_batch"
+path = "examples/grpc/batch.rs"
+required-features = ["grpc"]
+
+[[example]]
+name = "grpc_request_reply"
+path = "examples/grpc/request_reply.rs"
+required-features = ["grpc"]
+
 [[example]]
 name = "http_producer"
 path = "examples/http/producer.rs"
 required-features = ["http"]
 
+[[example]]
+name = "http_producer_cloud_events"
+path = "examples/http/producer_cloud_events.rs"
+required-features = ["http", "cloud_events"]
+
 [[example]]
 name = "http_consumer_server"
 path = "examples/http/consumer_server.rs"
@@ -144,3 +194,13 @@ required-features = ["tcp"]
 name = "tcp_producer_cloud_events"
 path = "examples/tcp/producer_cloud_events.rs"
 required-features = ["tcp", "cloud_events"]
+
+[[example]]
+name = "tcp_broadcast"
+path = "examples/tcp/broadcast.rs"
+required-features = ["tcp"]
+
+[[example]]
+name = "tcp_request_reply"
+path = "examples/tcp/request_reply.rs"
+required-features = ["tcp"]

eventmesh-sdks/eventmesh-sdk-rust/README.md

@@ -82,6 +82,11 @@ transport failures. HTTP consumers and webhook registrations additionally
 provide `close().await`, which unregisters remote subscriptions before
 signalling shutdown and joining.
 
+`GrpcWebhookConsumer` does not automatically unregister remote webhook
+subscriptions when `shutdown()` or `join()` is called. Retain the subscriptions
+and webhook URL, call `unsubscribe(...).await` explicitly, and only then call
+`shutdown()` and `join().await`. See the `grpc_webhook_consumer` example.
+
 HTTP request/reply is not exposed because the current SDK and stock Runtime do not provide a complete HTTP responder path. Use gRPC or TCP for request/reply.
 
 `Message` is a public dialect envelope, not a wire format. The selected transport owns protobuf, HTTP form, or TCP frame serialization. With `cloud_events`, CloudEvents remain CloudEvents; `Message::into_event_mesh()` does not silently flatten them into the native EventMesh model.

eventmesh-sdks/eventmesh-sdk-rust/build.rs

@@ -25,6 +25,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
         return Ok(());
     }
 
+    #[cfg(feature = "grpc")]
     tonic_build::configure()
         .build_server(false)
         .build_client(true)

eventmesh-sdks/eventmesh-sdk-rust/examples/README.md

@@ -6,14 +6,28 @@ Each example is a small executable with one responsibility. They use the default
 | --- | --- | --- | --- |
 | gRPC | `grpc_producer` | Publish a native EventMesh message | `cargo run --example grpc_producer --features grpc` |
 | gRPC | `grpc_consumer` | Stream consumption with `MessageHandler` | `cargo run --example grpc_consumer --features grpc` |
+| gRPC | `grpc_webhook_consumer` | Register an application-owned webhook and explicitly unregister it on exit | `cargo run --example grpc_webhook_consumer --features grpc` |
 | gRPC | `grpc_producer_cloud_events` | Publish a CloudEvent | `cargo run --example grpc_producer_cloud_events --features grpc,cloud_events` |
+| gRPC | `grpc_batch` | Publish several native messages with one batch RPC | `cargo run --example grpc_batch --features grpc` |
+| gRPC | `grpc_request_reply` | Run a synchronous subscriber and complete request/reply | `cargo run --example grpc_request_reply --features grpc` |
 | HTTP | `http_producer` | Publish over HTTP | `cargo run --example http_producer --features http` |
+| HTTP | `http_producer_cloud_events` | Publish a CloudEvent over HTTP | `cargo run --example http_producer_cloud_events --features http,cloud_events` |
 | HTTP | `http_consumer_server` | SDK-managed axum callback server | `cargo run --example http_consumer_server --features http` |
 | HTTP | `http_consumer_custom` | Application-owned axum webhook endpoint | `cargo run --example http_consumer_custom --features http` |
 | TCP | `tcp_producer` | Connected TCP publish | `cargo run --example tcp_producer --features tcp` |
 | TCP | `tcp_consumer` | Connected TCP subscribe | `cargo run --example tcp_consumer --features tcp` |
 | TCP | `tcp_producer_cloud_events` | Publish a CloudEvent over TCP | `cargo run --example tcp_producer_cloud_events --features tcp,cloud_events` |
+| TCP | `tcp_broadcast` | Send a fire-and-forget broadcast | `cargo run --example tcp_broadcast --features tcp` |
+| TCP | `tcp_request_reply` | Run a synchronous subscriber and complete request/reply | `cargo run --example tcp_request_reply --features tcp` |
 
 Run a consumer first, then run its corresponding producer. The HTTP examples listen on ports 8080 (built-in server) and 8081 (custom endpoint); change the advertised callback URL when EventMesh cannot reach `127.0.0.1`.
 
 The examples intentionally use minimal configuration. For timeouts, identity, credentials, HTTP TLS, endpoint weights, and TCP reconnect tuning, consult the public rustdoc with `cargo doc --features full --no-deps --open`.
+
+The two HTTP consumer examples handle Ctrl-C and call `close().await`, which
+unregisters their remote subscriptions before stopping local background work.
+Use the same shutdown pattern in long-running applications.
+
+The gRPC webhook consumer has no automatic remote cleanup. Its example retains
+the subscription and URL, calls `unsubscribe().await` after Ctrl-C, and then
+stops and joins the local heartbeat task.

eventmesh-sdks/eventmesh-sdk-rust/examples/grpc/batch.rs

@@ -0,0 +1,39 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use eventmesh::{
+    config::{Endpoint, GrpcConfig, ProducerOptions},
+    EventMeshMessage, GrpcClient, Message,
+};
+
+#[tokio::main]
+async fn main() -> eventmesh::Result<()> {
+    let client = GrpcClient::new(GrpcConfig::new(Endpoint::new("127.0.0.1", 10_205)?))?;
+    let producer = client.producer(ProducerOptions::new("test-producerGroup"))?;
+    let messages = (1..=3)
+        .map(|index| {
+            EventMeshMessage::new(
+                "test-topic-rust-sdk",
+                format!("hello from rust batch #{index}"),
+            )
+            .map(Message::from)
+        })
+        .collect::<eventmesh::Result<Vec<_>>>()?;
+
+    println!("published: {:?}", producer.publish_batch(messages).await?);
+    Ok(())
+}

eventmesh-sdks/eventmesh-sdk-rust/examples/grpc/request_reply.rs

@@ -0,0 +1,54 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use eventmesh::{
+    config::{Endpoint, GrpcConfig, GrpcConsumerOptions, ProducerOptions},
+    EventMeshMessage, GrpcClient, Message, Subscription,
+};
+use std::time::Duration;
+
+const TOPIC: &str = "test-topic-rust-sdk";
+
+#[tokio::main]
+async fn main() -> eventmesh::Result<()> {
+    let client = GrpcClient::new(GrpcConfig::new(Endpoint::new("127.0.0.1", 10_205)?))?;
+    let consumer = client
+        .stream_consumer(
+            GrpcConsumerOptions::new("test-consumerGroup"),
+            [Subscription::new(TOPIC).with_delivery_type(eventmesh::DeliveryType::Sync)],
+            |request: Message| async move {
+                let request = request.into_event_mesh()?;
+                Ok(Some(Message::from(EventMeshMessage::new(
+                    request.topic(),
+                    "pong",
+                )?)))
+            },
+        )
+        .await?;
+    let producer = client.producer(ProducerOptions::new("test-producerGroup"))?;
+
+    // Give EventMesh time to make the new subscription routable before the
+    // first synchronous request.
+    tokio::time::sleep(Duration::from_secs(1)).await;
+    let reply = producer
+        .request_reply(Message::from(EventMeshMessage::new(TOPIC, "ping")?))
+        .await?;
+    println!("reply: {reply:?}");
+
+    consumer.shutdown();
+    consumer.join().await
+}

eventmesh-sdks/eventmesh-sdk-rust/examples/grpc/webhook_consumer.rs

@@ -0,0 +1,48 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use eventmesh::{
+    config::{ConsumerOptions, Endpoint, GrpcConfig},
+    GrpcClient, Subscription,
+};
+
+const TOPIC: &str = "test-topic-rust-sdk";
+const WEBHOOK_URL: &str = "http://127.0.0.1:8080/eventmesh/callback";
+
+#[tokio::main]
+async fn main() -> eventmesh::Result<()> {
+    // Start an HTTP endpoint at WEBHOOK_URL before running this example.
+    let client = GrpcClient::new(GrpcConfig::new(Endpoint::new("127.0.0.1", 10_205)?))?;
+    let consumer = client
+        .webhook_consumer(ConsumerOptions::new("test-consumerGroup"))
+        .await?;
+    let subscription = Subscription::new(TOPIC);
+
+    consumer
+        .subscribe([subscription.clone()], WEBHOOK_URL)
+        .await?;
+    println!("registered {TOPIC}; press Ctrl-C to unregister and exit");
+    tokio::signal::ctrl_c().await?;
+
+    // shutdown() only stops local heartbeat work. Explicitly remove the
+    // remote registration first so it does not linger until server expiry.
+    let unsubscribe_result = consumer.unsubscribe([subscription], WEBHOOK_URL).await;
+    consumer.shutdown();
+    let join_result = consumer.join().await;
+    unsubscribe_result?;
+    join_result
+}

eventmesh-sdks/eventmesh-sdk-rust/examples/http/consumer_custom.rs

@@ -57,6 +57,10 @@ async fn main() -> eventmesh::Result<()> {
     consumer
         .subscribe(Subscription::new("test-topic-rust-sdk"), webhook_url)
         .await?;
-    axum::serve(listener, app).await?;
-    Ok(())
+    axum::serve(listener, app)
+        .with_graceful_shutdown(async {
+            let _ = tokio::signal::ctrl_c().await;
+        })
+        .await?;
+    consumer.close().await
 }

eventmesh-sdks/eventmesh-sdk-rust/examples/http/consumer_server.rs

@@ -45,5 +45,11 @@ async fn main() -> eventmesh::Result<()> {
             PrintHandler,
         )
         .await?;
-    consumer.join().await
+    tokio::select! {
+        result = consumer.join() => result,
+        signal = tokio::signal::ctrl_c() => {
+            signal?;
+            consumer.close().await
+        }
+    }
 }