Refine event mesh application workflows

Author: daflyinbed

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

@@ -41,7 +41,10 @@ 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("orders-producer"))?;
     let receipt = producer
-        .publish(Message::from(EventMeshMessage::new("orders.created", r#"{"id": 42}"#)))
+        .publish(Message::from(EventMeshMessage::new(
+            "orders.created",
+            r#"{"id": 42}"#,
+        )?))
         .await?;
     println!("accepted with code {}", receipt.code);
     Ok(())
@@ -73,11 +76,17 @@ See the runnable transport-specific consumer programs in [examples/README.md](ex
 
 `HttpClient::consumer` binds its callback socket before registering subscriptions, then owns the axum server, heartbeat, and registration lifecycle. For an application-owned endpoint, use `HttpClient::webhook_registration` with `eventmesh::http::codec::{parse_push_body, WebhookReply}`. TCP unsubscribe is session-wide, so its API is `unsubscribe_all()`.
 
+All consumers use the same local lifecycle contract: `shutdown()` only signals
+background work to stop, while `join().await` waits for it and reports task or
+transport failures. HTTP consumers and webhook registrations additionally
+provide `close().await`, which unregisters remote subscriptions before
+signalling shutdown and joining.
+
 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.
 
-`EventMeshMessage` is likewise a business model rather than a stable serde JSON contract. gRPC, HTTP, and TCP convert it into private transport-specific wire DTOs. Native messages require non-blank topics and content; an explicit `ttl` must be a positive millisecond value no greater than `i32::MAX`. EventMesh does not define a never-expire TTL sentinel.
+`EventMeshMessage` is likewise a business model rather than a stable serde JSON contract. gRPC, HTTP, and TCP convert it into private transport-specific wire DTOs. A topic must be non-blank and content must be present, but empty content is accepted for Java SDK interoperability. Inbound TTL metadata is preserved as received; each transport applies its own outbound content and TTL limits when publishing.
 
 ## Configuration and errors
 

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

@@ -29,7 +29,7 @@ async fn main() -> eventmesh::Result<()> {
         .publish(Message::from(EventMeshMessage::new(
             "test-topic-rust-sdk",
             "hello from rust",
-        )))
+        )?))
         .await?;
     println!("published: {receipt:?}");
     Ok(())

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

@@ -30,7 +30,7 @@ async fn main() -> eventmesh::Result<()> {
         .publish(Message::from(EventMeshMessage::new(
             "test-topic-rust-sdk",
             "hello from rust",
-        )))
+        )?))
         .await?;
     println!("published: {receipt:?}");
     Ok(())

eventmesh-sdks/eventmesh-sdk-rust/examples/tcp/producer.rs

@@ -31,7 +31,7 @@ async fn main() -> eventmesh::Result<()> {
         .publish(Message::from(EventMeshMessage::new(
             "test-topic-rust-sdk",
             "hello from rust",
-        )))
+        )?))
         .await?;
     println!("published: {receipt:?}");
     Ok(())

eventmesh-sdks/eventmesh-sdk-rust/src/config/client.rs

@@ -326,7 +326,11 @@ pub struct ConsumerOptions {
 }
 
 impl ConsumerOptions {
-    /// Create options for `group` with serial delivery by default.
+    /// Create options for `group`.
+    ///
+    /// Delivery concurrency is transport-specific. HTTP webhook requests may
+    /// run concurrently, TCP delivery is serial, and gRPC stream consumers use
+    /// [`GrpcConsumerOptions`] to configure handler concurrency.
     pub fn new(group: impl Into<String>) -> Self {
         Self {
             group: group.into(),

eventmesh-sdks/eventmesh-sdk-rust/src/error.rs

@@ -49,19 +49,24 @@ pub enum Error {
     #[error("grpc transport error: {0}")]
     GrpcTransport(String),
 
-    /// An HTTP transport error (Phase 2).
+    /// An HTTP transport error.
     #[error("http error: status {status}: {message}")]
-    Http { status: u16, message: String },
+    Http {
+        /// HTTP response status code.
+        status: u16,
+        /// Error description returned by the HTTP transport or peer.
+        message: String,
+    },
 
-    /// A TCP transport error (Phase 3).
+    /// A TCP transport error.
     #[error("tcp error: {0}")]
     Tcp(String),
 
     /// Serialization / deserialization failure.
     #[error("codec error: {0}")]
     Codec(#[from] serde_json::Error),
 
-    /// A message failed validation before it was sent on the wire.
+    /// A message failed validation during construction, decoding, or sending.
     #[error("invalid message: {0}")]
     InvalidMessage(String),
 
@@ -80,7 +85,12 @@ pub enum Error {
 
     /// The EventMesh server returned a non-success response code.
     #[error("server error: code={code} message={message}")]
-    Server { code: i32, message: String },
+    Server {
+        /// EventMesh server response code.
+        code: i32,
+        /// Error description returned by the EventMesh server.
+        message: String,
+    },
 
     /// Low-level I/O error.
     #[error("io error: {0}")]

eventmesh-sdks/eventmesh-sdk-rust/src/grpc.rs

@@ -154,14 +154,14 @@ impl GrpcWebhookConsumer {
             .map(|_| ())
     }
 
-    /// Stop the heartbeat task.
-    pub async fn shutdown(&self) {
-        self.inner.shutdown().await;
+    /// Signal the heartbeat task to stop.
+    pub fn shutdown(&self) {
+        self.inner.request_shutdown();
     }
 
-    /// Wait for the heartbeat task to stop.
-    pub async fn join(&self) {
-        self.inner.wait_for_shutdown().await;
+    /// Wait for the heartbeat task to stop and report task failure.
+    pub async fn join(&self) -> Result<()> {
+        self.inner.wait_for_shutdown().await
     }
 }
 
@@ -179,9 +179,9 @@ impl<H: MessageHandler> GrpcConsumer<H> {
             .map(|_| ())
     }
 
-    /// Request graceful stream shutdown.
-    pub async fn shutdown(&self) {
-        self.inner.shutdown().await;
+    /// Signal graceful stream shutdown.
+    pub fn shutdown(&self) {
+        self.inner.request_shutdown();
     }
 
     /// Wait for stream shutdown.

eventmesh-sdks/eventmesh-sdk-rust/src/http.rs

@@ -113,15 +113,15 @@ impl HttpClient {
             .await
         {
             lifecycle.cancel();
-            inner.shutdown().await;
+            let _ = inner.shutdown().await;
             let _ = server_handle.await;
             return Err(error);
         }
 
         if server_handle.is_finished() {
             lifecycle.cancel();
             let _ = inner.unsubscribe_all().await;
-            inner.shutdown().await;
+            let _ = inner.shutdown().await;
             return match server_handle.await {
                 Ok(Err(error)) => Err(error),
                 Ok(Ok(())) => Err(EventMeshError::ChannelClosed(
@@ -206,20 +206,29 @@ impl HttpConsumer {
         &self.webhook_url
     }
 
-    /// Unregister all subscriptions and stop heartbeat and callback serving.
-    pub async fn shutdown(&self) -> Result<()> {
-        let unregister_result = self.inner.unsubscribe_all().await;
-        self.inner.shutdown().await;
+    /// Signal heartbeat and callback serving to stop.
+    pub fn shutdown(&self) {
+        self.inner.request_shutdown();
         self.lifecycle.cancel();
-        let server_result = self.wait_for_server().await;
-        unregister_result.and(server_result)
     }
 
-    /// Wait until the callback server exits and all background work stops.
+    /// Wait until the callback server exits and report background task failure.
     pub async fn join(&self) -> Result<()> {
-        self.lifecycle.cancelled().await;
-        self.inner.wait_for_shutdown().await;
-        self.wait_for_server().await
+        let server_result = self.wait_for_server().await;
+        // A panicked server task cannot cancel the lifecycle token itself.
+        // Cancel it here so heartbeat shutdown is guaranteed before returning.
+        self.lifecycle.cancel();
+        self.inner.request_shutdown();
+        let heartbeat_result = self.inner.wait_for_shutdown().await;
+        server_result.and(heartbeat_result)
+    }
+
+    /// Unregister all subscriptions, signal shutdown, and join background work.
+    pub async fn close(&self) -> Result<()> {
+        let unregister_result = self.inner.unsubscribe_all().await;
+        self.shutdown();
+        let join_result = self.join().await;
+        unregister_result.and(join_result)
     }
 
     async fn wait_for_server(&self) -> Result<()> {
@@ -272,16 +281,22 @@ impl WebhookRegistration {
             .map(|_| ())
     }
 
-    /// Unregister every tracked subscription and stop the heartbeat task.
-    pub async fn shutdown(&self) -> Result<()> {
-        let unregister_result = self.inner.unsubscribe_all().await;
-        self.inner.shutdown().await;
-        unregister_result
+    /// Signal the heartbeat task to stop.
+    pub fn shutdown(&self) {
+        self.inner.request_shutdown();
     }
 
-    /// Wait for the heartbeat task to stop.
-    pub async fn join(&self) {
-        self.inner.wait_for_shutdown().await;
+    /// Wait for the heartbeat task to stop and report task failure.
+    pub async fn join(&self) -> Result<()> {
+        self.inner.wait_for_shutdown().await
+    }
+
+    /// Unregister every tracked subscription, shut down, and join.
+    pub async fn close(&self) -> Result<()> {
+        let unregister_result = self.inner.unsubscribe_all().await;
+        self.shutdown();
+        let join_result = self.join().await;
+        unregister_result.and(join_result)
     }
 }
 
@@ -399,7 +414,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn managed_consumer_serves_and_unregisters_on_shutdown() {
+    async fn managed_consumer_serves_and_unregisters_on_close() {
         let (client, codes, runtime) = mock_runtime(false).await;
         let (tx, mut rx) = mpsc::unbounded_channel();
         let consumer = client
@@ -430,18 +445,40 @@ mod tests {
             .await
             .unwrap()
             .unwrap();
-        assert_eq!(
-            received.as_event_mesh().unwrap().content.as_deref(),
-            Some("created")
-        );
+        assert_eq!(received.as_event_mesh().unwrap().content(), "created");
 
-        consumer.shutdown().await.unwrap();
+        consumer.close().await.unwrap();
         let codes = codes.lock().await.clone();
         assert!(codes.contains(&crate::common::status_code::RequestCode::SUBSCRIBE));
         assert!(codes.contains(&crate::common::status_code::RequestCode::UNSUBSCRIBE));
         runtime.abort();
     }
 
+    #[tokio::test]
+    async fn managed_consumer_shutdown_only_signals_local_tasks() {
+        let (client, codes, runtime) = mock_runtime(false).await;
+        let consumer = client
+            .consumer(
+                ConsumerOptions::new("group"),
+                WebhookOptions::new("127.0.0.1:0".parse().unwrap()),
+                [Subscription::new("orders")],
+                |_message| async { Ok(None) },
+            )
+            .await
+            .unwrap();
+
+        consumer.shutdown();
+        consumer.join().await.unwrap();
+
+        let codes = codes.lock().await.clone();
+        assert!(codes.contains(&crate::common::status_code::RequestCode::SUBSCRIBE));
+        assert!(
+            !codes.contains(&crate::common::status_code::RequestCode::UNSUBSCRIBE),
+            "shutdown must not perform remote cleanup; close owns that operation"
+        );
+        runtime.abort();
+    }
+
     #[tokio::test]
     async fn registration_failure_stops_managed_server() {
         let (client, _codes, runtime) = mock_runtime(true).await;

eventmesh-sdks/eventmesh-sdk-rust/src/lib.rs

@@ -41,11 +41,13 @@
 //!
 //! A [`MessageHandler`] returns `Ok(None)` to acknowledge an asynchronous
 //! delivery, `Ok(Some(reply))` to reply to a synchronous delivery, or `Err(_)`
-//! to report application failure. Long-lived consumers expose `shutdown` and
-//! `join`; a managed HTTP consumer also owns its callback server, registration,
-//! and heartbeat lifecycle.
+//! to report application failure. Long-lived consumers use a two-step
+//! lifecycle: `shutdown` only signals cancellation, while `join` waits for
+//! background work and reports failures. Managed HTTP consumers and webhook
+//! registrations additionally expose `close` to unregister remotely before
+//! shutting down and joining.
 
-#![deny(unsafe_code)]
+#![deny(missing_docs, unsafe_code)]
 
 // These modules are wire adapters retained behind the v2 public façade. Some
 // protocol-specific compatibility paths are intentionally feature-dependent,
@@ -82,7 +84,9 @@ pub mod tcp;
 
 pub use error::{Error, Result};
 pub use handler::MessageHandler;
-pub use message::{EventMeshMessage, Message, MessageKind, PublishReceipt};
+pub use message::{
+    EventMeshMessage, EventMeshMessageBuilder, Message, MessageKind, PublishReceipt,
+};
 pub use subscription::{DeliveryMode, DeliveryType, Subscription};
 
 #[cfg(feature = "grpc")]

eventmesh-sdks/eventmesh-sdk-rust/src/message.rs

@@ -27,7 +27,7 @@
 use crate::error::EventMeshError;
 use crate::error::Result;
 
-pub use crate::model::EventMeshMessage;
+pub use crate::model::{EventMeshMessage, EventMeshMessageBuilder};
 
 /// Which public event dialect a [`Message`] contains.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]