add request/reply support in http-server (#68)

resolves #67

Signed-off-by: markfisher <mark@modulewise.com>

Author: markfisher

crates/http-server/src/config.rs

@@ -26,7 +26,11 @@ pub enum RouteTarget {
         body: Option<String>,
     },
     /// Publish the request body to a channel.
-    Channel { channel: String },
+    Channel {
+        channel: String,
+        /// If set, use request-reply, else fire-and-forget.
+        reply_timeout_ms: Option<u64>,
+    },
 }
 
 /// Parsed HTTP server definition.
@@ -74,7 +78,15 @@ impl ConfigHandler for HttpServerConfigHandler {
     fn claimed_properties(&self) -> HashMap<&str, &[&str]> {
         HashMap::from([(
             "server",
-            ["type", "port", "route", "otlp-endpoint", "otlp-protocol"].as_slice(),
+            [
+                "type",
+                "port",
+                "route",
+                "otlp-endpoint",
+                "otlp-protocol",
+                "reply-timeout-ms",
+            ]
+            .as_slice(),
         )])
     }
 
@@ -203,6 +215,19 @@ fn parse_routes(server_name: &str, properties: &mut PropertyMap) -> Result<Vec<R
         let function = get_optional_string(&route_props, "function", &ctx)?;
         let channel = get_optional_string(&route_props, "channel", &ctx)?;
         let body = get_optional_string(&route_props, "body", &ctx)?;
+        let reply_timeout_ms = match route_props.get("reply-timeout-ms") {
+            Some(serde_json::Value::Number(n)) => Some(
+                n.as_u64()
+                    .ok_or_else(|| ctx("'reply-timeout-ms'", "must be a non-negative integer"))?,
+            ),
+            Some(got) => {
+                return Err(ctx(
+                    "'reply-timeout-ms'",
+                    &format!("must be a number, got {got}"),
+                ));
+            }
+            None => None,
+        };
 
         let has_component = component.is_some() || function.is_some();
         let has_channel = channel.is_some();
@@ -218,7 +243,10 @@ fn parse_routes(server_name: &str, properties: &mut PropertyMap) -> Result<Vec<R
             if body.is_some() {
                 return Err(ctx("", "'body' is not valid on channel routes"));
             }
-            RouteTarget::Channel { channel }
+            RouteTarget::Channel {
+                channel,
+                reply_timeout_ms,
+            }
         } else {
             let component = component
                 .ok_or_else(|| ctx("", "missing 'component' (or 'channel' for channel routes)"))?;
@@ -534,8 +562,12 @@ mod tests {
         let servers = config.lock().unwrap();
         assert_eq!(servers[0].routes[0].method, "POST");
         match &servers[0].routes[0].target {
-            RouteTarget::Channel { channel } => {
+            RouteTarget::Channel {
+                channel,
+                reply_timeout_ms,
+            } => {
                 assert_eq!(channel, "incoming-events");
+                assert!(reply_timeout_ms.is_none());
             }
             other => panic!("expected Channel target, got {other:?}"),
         }

crates/http-server/src/server.rs

@@ -37,6 +37,7 @@ enum Target {
     },
     Channel {
         channel: String,
+        reply_timeout_ms: Option<u64>,
     },
 }
 
@@ -89,8 +90,12 @@ impl Route {
                     body: body.clone(),
                 }
             }
-            RouteTarget::Channel { channel } => Target::Channel {
+            RouteTarget::Channel {
+                channel,
+                reply_timeout_ms,
+            } => Target::Channel {
                 channel: channel.clone(),
+                reply_timeout_ms: *reply_timeout_ms,
             },
         };
 
@@ -167,7 +172,13 @@ impl Router {
                 self.invoke_component(component, function, body.as_deref(), params, req)
                     .await
             }
-            Target::Channel { channel } => self.publish_to_channel(channel, req).await,
+            Target::Channel {
+                channel,
+                reply_timeout_ms,
+            } => {
+                self.publish_to_channel(channel, req, *reply_timeout_ms)
+                    .await
+            }
         }
     }
 
@@ -346,6 +357,7 @@ impl Router {
         &self,
         channel: &str,
         req: Request<Incoming>,
+        reply_timeout_ms: Option<u64>,
     ) -> Response<Full<Bytes>> {
         let publisher = match &self.publisher {
             Some(p) => p,
@@ -379,17 +391,61 @@ impl Router {
         let mut headers = HashMap::new();
         headers.insert("content-type".to_string(), content_type);
 
-        match publisher.publish(channel, body_bytes, headers).await {
-            Ok(()) => Response::builder()
-                .status(StatusCode::ACCEPTED)
-                .body(Full::new(Bytes::new()))
-                .unwrap(),
-            Err(e) => Response::builder()
-                .status(StatusCode::INTERNAL_SERVER_ERROR)
-                .body(Full::new(Bytes::from(format!(
-                    "failed to publish to channel '{channel}': {e}"
-                ))))
-                .unwrap(),
+        match reply_timeout_ms {
+            None => match publisher.publish(channel, body_bytes, headers).await {
+                Ok(()) => Response::builder()
+                    .status(StatusCode::ACCEPTED)
+                    .body(Full::new(Bytes::new()))
+                    .unwrap(),
+                Err(e) => Response::builder()
+                    .status(StatusCode::INTERNAL_SERVER_ERROR)
+                    .body(Full::new(Bytes::from(format!(
+                        "failed to publish to channel '{channel}': {e}"
+                    ))))
+                    .unwrap(),
+            },
+            Some(timeout_ms) => {
+                let handle = match publisher
+                    .publish_request(channel, body_bytes, headers)
+                    .await
+                {
+                    Ok(h) => h,
+                    Err(e) => {
+                        return Response::builder()
+                            .status(StatusCode::INTERNAL_SERVER_ERROR)
+                            .body(Full::new(Bytes::from(format!(
+                                "failed to publish to channel '{channel}': {e}"
+                            ))))
+                            .unwrap();
+                    }
+                };
+
+                let timeout = std::time::Duration::from_millis(timeout_ms);
+                match tokio::time::timeout(timeout, handle.take()).await {
+                    Ok(Ok(reply)) => {
+                        let reply_content_type = reply
+                            .headers()
+                            .content_type()
+                            .unwrap_or("application/octet-stream")
+                            .to_string();
+                        Response::builder()
+                            .status(StatusCode::OK)
+                            .header("content-type", reply_content_type)
+                            .body(Full::new(Bytes::copy_from_slice(reply.body())))
+                            .unwrap()
+                    }
+                    Ok(Err(e)) => Response::builder()
+                        .status(StatusCode::INTERNAL_SERVER_ERROR)
+                        .body(Full::new(Bytes::from(format!("reply error: {e}"))))
+                        .unwrap(),
+                    Err(_) => Response::builder()
+                        .status(StatusCode::GATEWAY_TIMEOUT)
+                        .body(Full::new(Bytes::from(format!(
+                            "no reply received within {timeout_ms}ms"
+                        ))))
+                        .unwrap(),
+                }
+            }
         }
     }
 }

examples/http-server/config.toml

@@ -10,6 +10,7 @@ function = "greet"
 [server.api.route.bonjour]
 path = "/bonjour"
 channel = "names"
+reply-timeout-ms = 1000
 
 [component.hello]
 uri = "../hello-world/greeter/target/wasm32-unknown-unknown/release/greeter.wasm"

examples/http-server/greet.sh

@@ -6,4 +6,4 @@ echo "--- GET /hello/world (component route) ---"
 curl -s localhost:8888/hello/World; echo
 
 echo "--- POST /bonjour (channel route -> subscription -> bonjour component) ---"
-curl -s -X POST -H "Content-Type: text/plain" -d "le Monde" localhost:8888/bonjour
+curl -s -X POST -H "Content-Type: text/plain" -d "le Monde" localhost:8888/bonjour; echo