feat: added event stream support

Cargo.toml

@@ -185,6 +185,9 @@ dyn-clone = { version = "1.0", default-features = false }
 either = { version = "1.9.0", default-features = false }
 env_logger = { version = "0.10", default-features = false }
 escargot = { version = "0.5", default-features = false }
+eventsource-stream = { version = "0.2", default-features = false, features = [
+  "std",
+] }
 futures = { version = "0.3", default-features = false, features = ["std"] }
 getrandom = { version = "0.2", default-features = false }
 getset = { version = "0.1", default-features = false }

crates/components/wick-http-client/Cargo.toml

@@ -10,6 +10,7 @@ description = "SQL Database component for the wick project."
 [features]
 
 [dependencies]
+bytes = { workspace = true }
 wick-packet = { workspace = true, features = ["rt-tokio", "invocation"] }
 wick-interface-http = { workspace = true }
 flow-component = { workspace = true, features = ["invocation"] }
@@ -28,6 +29,7 @@ serde = { workspace = true, features = ["derive"] }
 futures = { workspace = true }
 thiserror = { workspace = true }
 serde_json = { workspace = true }
+eventsource-stream = { workspace = true }
 
 #
 [dev-dependencies]

crates/components/wick-http-client/src/component.rs

@@ -2,6 +2,7 @@ use std::collections::HashMap;
 use std::sync::Arc;
 
 use anyhow::anyhow;
+use eventsource_stream::Eventsource;
 use flow_component::{BoxFuture, Component, ComponentError, RuntimeCallback};
 use futures::{Stream, StreamExt, TryStreamExt};
 use reqwest::header::CONTENT_TYPE;
@@ -15,7 +16,7 @@ use wick_config::config::components::{
   HttpClientOperationDefinition,
   OperationConfig,
 };
-use wick_config::config::{Codec, HttpMethod, LiquidJsonConfig, Metadata, UrlResource};
+use wick_config::config::{Codec, HttpEvent, HttpMethod, LiquidJsonConfig, Metadata, UrlResource};
 use wick_config::{ConfigValidation, Resolver};
 use wick_interface_types::{ComponentSignature, OperationSignatures};
 use wick_packet::{Base64Bytes, FluxChannel, Invocation, Observer, Packet, PacketSender, PacketStream, RuntimeConfig};
@@ -281,6 +282,9 @@ async fn handle(
         }
         Codec::FormData => request_builder.form(&body),
         Codec::Text => request_builder.body(body.to_string()),
+        Codec::EventStream => {
+          unimplemented!("Event stream is not a valid client content-type")
+        }
       }
     } else {
       request_builder
@@ -322,9 +326,11 @@ async fn handle(
       match value {
         "application/json" => Codec::Json,
         "application/x-www-form-urlencoded" => Codec::FormData,
+        "text/event-stream" => Codec::EventStream,
         _ => Codec::Raw,
       }
     });
+
     let (our_response, body_stream) = match crate::conversions::to_wick_response(response) {
       Ok(r) => r,
       Err(e) => {
@@ -357,6 +363,22 @@ fn output_task(
 ) -> BoxFuture<'static, ()> {
   let task = async move {
     match codec {
+      Codec::EventStream => {
+        let mut stream = body_stream.map(Into::into).eventsource();
+        while let Some(event) = stream.next().await {
+          match event {
+            Ok(event) => {
+              let wick_event = HttpEvent::new(Some(event.event), event.data, Some(event.id), event.retry);
+              span.in_scope(|| debug!("{} {}", format!("{:?}", wick_event), "http:client:response_body"));
+              let _ = tx.send(Packet::encode("body", wick_event));
+            }
+            Err(e) => {
+              let _ = tx.error(wick_packet::Error::component_error(e.to_string()));
+              return;
+            }
+          }
+        }
+      }
       Codec::Json => {
         let bytes: Vec<Base64Bytes> = match body_stream.try_collect().await {
           Ok(r) => r,
@@ -659,5 +681,43 @@ mod test {
 
       Ok(())
     }
+
+    #[test_logger::test(tokio::test)]
+    async fn test_event_stream() -> Result<()> {
+      let (app_config, component_config) = get_config();
+      let comp = get_component(&app_config, component_config);
+
+      let event_stream = "data: {\"id\":\"1\",\"object\":\"event1\"}\n\n\
+         data: {\"id\":\"2\",\"object\":\"event2\"}\n\n";
+      let packets = packet_stream!(("input", event_stream));
+
+      let invocation = Invocation::test(
+        "test_event_stream",
+        Entity::local("event_stream_op"),
+        packets,
+        Default::default(),
+      )?;
+      let stream = comp
+        .handle(invocation, Default::default(), panic_callback())
+        .await?
+        .filter(|p| futures::future::ready(p.as_ref().map_or(false, |p| p.has_data())))
+        .collect::<Vec<_>>()
+        .await;
+
+      let packets = stream.into_iter().collect::<Result<Vec<_>, _>>()?;
+      for packet in packets {
+        if packet.port() == "body" {
+          let response: HttpEvent = packet.decode().unwrap();
+          let response_id = response.get_id().as_ref().unwrap();
+          let response_event = response.get_event().as_ref().unwrap();
+          assert!(response_id == "1" && response_event == "event1" || response_id == "2" && response_event == "event2");
+        } else {
+          let response: HttpResponse = packet.decode().unwrap();
+          assert_eq!(response.version, HttpVersion::Http11);
+        }
+      }
+
+      Ok(())
+    }
   }
 }

crates/components/wick-http-client/src/error.rs

@@ -20,4 +20,7 @@ pub enum Error {
 
   #[error("Proxy and baseurl must not be the same: {0}")]
   ProxyLoop(Url),
+
+  #[error("Conversion error: {0}")]
+  Conversion(String),
 }

crates/interfaces/wick-interface-http/component.yaml

@@ -3,7 +3,7 @@
 name: http
 kind: wick/types@v1
 metadata:
-  version: 0.4.0
+  version: 0.5.0
 package:
   registry:
     host: registry.candle.dev
@@ -242,3 +242,19 @@ types:
     types:
       - HttpRequest
       - HttpResponse
+  - name: HttpEvent
+    kind: wick/type/struct@v1
+    description: HTTP server side event
+    fields:
+      - name: event
+        type: string?
+        description: The event name if given
+      - name: data
+        type: string
+        description: The event data
+      - name: id
+        type: string?
+        description: The event id if given
+      - name: retry
+        type: u32?
+        description: Retry duration if given

crates/wick/wick-config/definitions/v1/manifest.apex

@@ -1036,6 +1036,9 @@ enum Codec {
 
   "Raw text"
   Text = 3 as "text",
+
+  "Eventstream"
+  EventStream = 4 as "event-stream",
 }
 
 "Supported HTTP methods"

crates/wick/wick-config/docs/v1.md

@@ -1894,6 +1894,7 @@ Any one of the following types:
 | Raw | unknown type | Raw bytes |
 | FormData | unknown type | Form Data |
 | Text | unknown type | Raw text |
+| EventStream | unknown type | Eventstream |
 
 
 --------

crates/wick/wick-config/json-schema/manifest.json

@@ -2980,7 +2980,8 @@
         "Json",
         "Raw",
         "FormData",
-        "Text"
+        "Text",
+        "EventStream"
       ]
     },
     "v1.HttpMethod": {

crates/wick/wick-config/json-schema/v1/manifest.json

@@ -2588,7 +2588,7 @@
 
   "v1.Codec": {
     "$anchor": "v1.Codec",
-    "enum": ["Json", "Raw", "FormData", "Text"]
+    "enum": ["Json", "Raw", "FormData", "Text", "EventStream"]
   },
 
   "v1.HttpMethod": {

crates/wick/wick-config/src/config/common.rs

@@ -29,7 +29,7 @@ pub use self::error_behavior::ErrorBehavior;
 pub use self::exposed_resources::{ExposedVolume, ExposedVolumeBuilder};
 pub use self::glob::Glob;
 pub use self::host_definition::{HostConfig, HostConfigBuilder, HttpConfig, HttpConfigBuilder};
-pub use self::http::{Codec, HttpMethod};
+pub use self::http::{Codec, HttpEvent, HttpMethod};
 pub use self::import_definition::ImportDefinition;
 pub use self::interface::InterfaceDefinition;
 pub use self::liquid_json_config::LiquidJsonConfig;