Refine request-reply transport support and webhook protocol detection

Author: daflyinbed

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

@@ -73,7 +73,7 @@ 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()`.
 
-`HttpProducer::request_reply` encodes EventMesh's HTTP synchronous-publish request, but the current stock Runtime cannot route an HTTP-originated synchronous message through a gRPC stream consumer and return its reply to the HTTP request. The corresponding real-Runtime e2e is retained but ignored to make this compatibility gap visible. Use gRPC or TCP for request/reply unless the target deployment provides a compatible HTTP synchronous-reply path.
+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.
 
@@ -83,7 +83,7 @@ See the runnable transport-specific consumer programs in [examples/README.md](ex
 
 All configurations require a validated `Endpoint`; HTTP uses a non-empty `EndpointSet`. Use `with_*` methods to set optional identity, credentials, timeouts, HTTP TLS, proxy, and reconnect settings. EventMesh Runtime's gRPC endpoint is plaintext and the gRPC client intentionally does not expose TLS configuration. `Debug` output redacts secrets.
 
-Default request timeouts are 5 seconds (gRPC), 15 seconds (HTTP), and 20 seconds (TCP). `ClientOptions::with_request_timeout` changes a client's default; every producer also has `request_reply_with_timeout` for one call. TCP separately has a 1-second connect timeout and a 20-second control timeout.
+Default request timeouts are 5 seconds (gRPC), 15 seconds (HTTP), and 20 seconds (TCP). `ClientOptions::with_request_timeout` changes a client's default; gRPC and TCP producers also have `request_reply_with_timeout` for one call. TCP separately has a 1-second connect timeout and a 20-second control timeout.
 
 Operations return the pattern-matchable `eventmesh::Error`; common variants include `Config`, `InvalidArgument`, `InvalidMessage`, `Timeout`, `Server`, `Protocol`, `Unsupported`, and transport-specific errors.
 

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

@@ -1083,6 +1083,7 @@ mod tests {
             .is_err());
     }
 
+    #[cfg(any(feature = "grpc", feature = "http", feature = "tcp"))]
     #[test]
     fn role_groups_must_not_be_blank() {
         assert!(ProducerOptions::new("  ").validate().is_err());

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

@@ -26,7 +26,7 @@ use crate::transport::grpc::{
     GrpcClient as ChannelClient, GrpcProducer as LegacyProducer,
     GrpcStreamConsumer as LegacyConsumer, GrpcWebhookConsumer as LegacyWebhookConsumer,
 };
-use crate::transport::Publisher as LegacyPublisher;
+use crate::transport::{Publisher as LegacyPublisher, RequestReply as LegacyRequestReply};
 use crate::MessageHandler;
 
 /// A configured EventMesh gRPC client.

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

@@ -62,7 +62,6 @@ impl HttpClient {
         options.validate()?;
         Ok(HttpProducer {
             inner: LegacyProducer::new(self.config.legacy(Some(&options), None))?,
-            timeout: self.config.request_timeout(),
         })
     }
 
@@ -155,7 +154,6 @@ impl HttpClient {
 /// HTTP publishing capability.
 pub struct HttpProducer {
     inner: LegacyProducer,
-    timeout: std::time::Duration,
 }
 
 impl HttpProducer {
@@ -175,50 +173,6 @@ impl HttpProducer {
                 .map(PublishReceipt::from_legacy),
         }
     }
-
-    /// Send an event and await its reply.
-    ///
-    /// # Runtime compatibility
-    ///
-    /// The SDK implements EventMesh's HTTP synchronous-publish wire request,
-    /// but the current EventMesh Runtime cannot route an HTTP-originated
-    /// synchronous message through a gRPC stream consumer and return that
-    /// consumer's reply to the HTTP request. Against the stock Runtime this
-    /// operation can therefore fail or time out; use gRPC or TCP when a
-    /// verified request/reply path is required. Keep this method only for
-    /// deployments that provide a compatible HTTP synchronous-reply path.
-    pub async fn request_reply(&self, message: Message) -> Result<Message> {
-        self.request_reply_with_timeout(message, self.timeout).await
-    }
-
-    /// Send an event and await its reply with a per-operation timeout.
-    ///
-    /// This has the same current Runtime limitation documented on
-    /// [`request_reply`](Self::request_reply).
-    pub async fn request_reply_with_timeout(
-        &self,
-        message: Message,
-        timeout: std::time::Duration,
-    ) -> Result<Message> {
-        if timeout.is_zero() {
-            return Err(EventMeshError::InvalidArgument(
-                "request/reply timeout must be greater than zero".into(),
-            ));
-        }
-        match message {
-            Message::EventMesh(message) => self
-                .inner
-                .request_reply(message, timeout)
-                .await
-                .map(Message::EventMesh),
-            #[cfg(feature = "cloud_events")]
-            Message::CloudEvent(event) => self
-                .inner
-                .request_reply_cloud_event(event, timeout)
-                .await
-                .map(Message::CloudEvent),
-        }
-    }
 }
 
 /// An SDK-managed HTTP consumer with an embedded axum callback server.

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

@@ -25,7 +25,7 @@ use crate::subscription::Subscription;
 use crate::transport::tcp::{
     TcpConsumer as LegacyConsumer, TcpMessage, TcpProducer as LegacyProducer,
 };
-use crate::transport::Publisher as LegacyPublisher;
+use crate::transport::{Publisher as LegacyPublisher, RequestReply as LegacyRequestReply};
 use crate::MessageHandler;
 use tracing::warn;
 

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

@@ -25,7 +25,7 @@ use crate::error::{EventMeshError, Result};
 use crate::model::{EventMeshMessage, PublishResponse};
 use crate::transport::grpc::client::GrpcClient;
 use crate::transport::grpc::codec;
-use crate::transport::Publisher;
+use crate::transport::{Publisher, RequestReply};
 
 /// gRPC-based producer.
 pub struct GrpcProducer {
@@ -204,7 +204,9 @@ impl Publisher for GrpcProducer {
         }
         Ok(response)
     }
+}
 
+impl RequestReply for GrpcProducer {
     async fn request_reply(
         &self,
         message: EventMeshMessage,

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

@@ -418,36 +418,6 @@ pub fn parse_response(body: &str) -> Result<PublishResponse> {
     Ok(obj.into())
 }
 
-/// The reply payload returned inside the `retMsg` field of a request-reply
-/// `EventMeshRetObj`.
-///
-/// Mirrors `SendMessageResponseBody.ReplyMessage` on the Java side:
-/// `topic`, `body`, and `properties`.
-#[derive(Debug, Clone, Deserialize)]
-pub struct ReplyMessage {
-    #[serde(default)]
-    pub topic: Option<String>,
-    #[serde(default)]
-    pub body: Option<String>,
-    #[serde(default)]
-    pub properties: HashMap<String, String>,
-}
-
-/// Parse the reply message from the `retMsg` field of a request-reply
-/// response, mapping `body` → `content`, `topic` → `topic`, and
-/// `properties` → `props`.
-///
-/// Mirrors the Java SDK's `EventMeshMessageProducer.transformMessage`, which
-/// deserializes `retMsg` as a `SendMessageResponseBody.ReplyMessage`.
-pub fn parse_reply(ret_msg: &str) -> Result<EventMeshMessage> {
-    let reply: ReplyMessage = serde_json::from_str(ret_msg)?;
-    Ok(EventMeshMessage::builder()
-        .topic(reply.topic.unwrap_or_default())
-        .content(reply.body.unwrap_or_default())
-        .props(reply.properties)
-        .build())
-}
-
 /// Form-encode a list of `(key, value)` pairs into a URL-encoded body string.
 pub fn form_encode(fields: &[(String, String)]) -> String {
     serde_urlencoded::to_string(fields).unwrap_or_default()
@@ -458,11 +428,6 @@ pub fn publish_code() -> i32 {
     RequestCode::MSG_SEND_ASYNC
 }
 
-/// Request code for synchronous request-reply (code-based routing).
-pub fn publish_sync_code() -> i32 {
-    RequestCode::MSG_SEND_SYNC
-}
-
 pub fn subscribe_code() -> i32 {
     RequestCode::SUBSCRIBE
 }
@@ -710,12 +675,6 @@ mod tests {
         assert!(!headers.iter().any(|(k, _)| *k == "token"));
     }
 
-    #[test]
-    fn publish_sync_code_is_101() {
-        assert_eq!(publish_sync_code(), RequestCode::MSG_SEND_SYNC);
-        assert_eq!(publish_sync_code(), 101);
-    }
-
     #[test]
     fn parse_response_success() {
         let body = r#"{"retCode":0,"retMsg":"success","resTime":42}"#;
@@ -741,15 +700,6 @@ mod tests {
         assert!(parse_response(body).is_err());
     }
 
-    #[test]
-    fn parse_reply_from_ret_msg() {
-        let ret_msg = r#"{"topic":"reply-topic","body":"reply-body","properties":{"k":"v"}}"#;
-        let msg = parse_reply(ret_msg).unwrap();
-        assert_eq!(msg.topic.as_deref(), Some("reply-topic"));
-        assert_eq!(msg.content.as_deref(), Some("reply-body"));
-        assert_eq!(msg.get_prop("k"), Some("v"));
-    }
-
     #[test]
     fn parse_push_body_form_urlencoded() {
         let body = "content=hello&topic=test-topic&bizseqno=seq1";

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

@@ -17,8 +17,6 @@
 
 //! HTTP producer.
 
-use std::time::Duration;
-
 use tracing::debug;
 
 use crate::config::HttpClientConfig;
@@ -30,8 +28,7 @@ use crate::transport::Publisher;
 
 /// HTTP-based producer.
 ///
-/// Implements fire-and-forget publish, batch publish, and synchronous
-/// request-reply over the EventMesh HTTP protocol.
+/// Implements publishing over the EventMesh HTTP protocol.
 pub struct HttpProducer {
     client: EventMeshHttpClient,
 }
@@ -75,38 +72,6 @@ impl HttpProducer {
             .await
     }
 
-    /// Send a native CloudEvent and wait for a native CloudEvent reply.
-    ///
-    /// The current Runtime cannot complete the HTTP-originated synchronous
-    /// request through its gRPC stream-consumer reply path. This codec remains
-    /// for deployments that provide a compatible HTTP synchronous-reply path.
-    #[cfg(feature = "cloud_events")]
-    pub async fn request_reply_cloud_event(
-        &self,
-        mut event: cloudevents::Event,
-        timeout: Duration,
-    ) -> Result<cloudevents::Event> {
-        use cloudevents::AttributesReader;
-
-        ensure_ce_extension(&mut event, "bizseqno");
-        ensure_ce_extension(&mut event, "uniqueid");
-        ensure_ce_ttl(&mut event);
-        let topic = event
-            .subject()
-            .ok_or_else(|| {
-                EventMeshError::InvalidMessage("CloudEvent subject (topic) is required".into())
-            })?
-            .to_string();
-        let message = EventMeshMessage::builder()
-            .topic(topic)
-            .content(serde_json::to_string(&event)?)
-            .build();
-        let reply = self
-            .request_reply_with_protocol(message, timeout, EventMeshProtocolType::CloudEvents)
-            .await?;
-        decode_cloud_event_reply(reply)
-    }
-
     /// Internal publish with a specific protocol type.
     async fn publish_with_protocol(
         &self,
@@ -133,33 +98,6 @@ impl HttpProducer {
         debug!("published topic={:?}", message.topic);
         Ok(response)
     }
-
-    async fn request_reply_with_protocol(
-        &self,
-        message: EventMeshMessage,
-        timeout: Duration,
-        protocol_type: EventMeshProtocolType,
-    ) -> Result<EventMeshMessage> {
-        message.validate_for_publish()?;
-        let config = self.client.config();
-        let body = codec::encode_publish(&message, &config.identity);
-        let headers =
-            codec::build_headers(codec::publish_sync_code(), protocol_type, &config.identity);
-        let text = self
-            .client
-            .post_form(uri::ROOT, &body, &headers, timeout)
-            .await?;
-        let response = codec::parse_response(&text)?;
-        if !response.is_success() {
-            return Err(EventMeshError::Server {
-                code: response.code.unwrap_or(-1) as i32,
-                message: response
-                    .message
-                    .unwrap_or_else(|| "request-reply failed".into()),
-            });
-        }
-        codec::parse_reply(response.message.as_deref().unwrap_or(""))
-    }
 }
 
 impl Publisher for HttpProducer {
@@ -180,39 +118,6 @@ impl Publisher for HttpProducer {
                 .into(),
         ))
     }
-
-    async fn request_reply(
-        &self,
-        message: EventMeshMessage,
-        timeout: Duration,
-    ) -> Result<EventMeshMessage> {
-        self.request_reply_with_protocol(message, timeout, EventMeshProtocolType::EventMeshMessage)
-            .await
-    }
-}
-
-/// Decode the HTTP reply payload into a native CloudEvent.  EventMesh
-/// installations return either CloudEvents JSON directly or the legacy
-/// `ReplyMessage` envelope; accepting both keeps interop with Java runtimes.
-#[cfg(feature = "cloud_events")]
-fn decode_cloud_event_reply(reply: EventMeshMessage) -> Result<cloudevents::Event> {
-    if let Some(content) = &reply.content {
-        if let Ok(event) = serde_json::from_str(content) {
-            return Ok(event);
-        }
-    }
-    use cloudevents::{EventBuilder, EventBuilderV10};
-    let topic = reply.topic.unwrap_or_default();
-    EventBuilderV10::new()
-        .id(reply
-            .unique_id
-            .unwrap_or_else(|| uuid::Uuid::new_v4().to_string()))
-        .source("eventmesh://http-reply")
-        .ty("eventmesh.reply")
-        .subject(topic)
-        .data("text/plain", reply.content.unwrap_or_default())
-        .build()
-        .map_err(|e| EventMeshError::InvalidMessage(format!("invalid CloudEvent reply: {e}")))
 }
 
 /// Ensure a CloudEvent extension attribute is present and non-blank,