reworked http request in sink for cloud-event

Author: rjtch

cdviz-collector/src/sinks/http.rs

@@ -1,5 +1,6 @@
 use cdevents_sdk::cloudevents::BuilderExt;
 use cloudevents::{EventBuilder, EventBuilderV10};
+use cloudevents::binding::reqwest::RequestBuilderExt;
 use reqwest::Url;
 use reqwest_middleware::{ClientBuilder, ClientWithMiddleware};
 use serde::{Deserialize, Serialize};
@@ -48,25 +49,34 @@ impl Sink for HttpSink {
     async fn send(&self, msg: &Message) -> Result<()> {
         let cd_event = msg.cdevent.clone();
         // convert  CdEvent to cloudevents
-        let event_result = EventBuilderV10::new().with_cdevent(cd_event).unwrap().build();
-
+        let event_result = EventBuilderV10::new().with_cdevent(cd_event.clone());
         match event_result {
-            Ok(value) => {
-                let resp = self
-                    .client
+            Ok(event_builder) => {
+                let event_result = event_builder.build();
+                let value = event_result.map_err(|e| {
+                    tracing::warn!(error = ?e, "Failed to build event")
+                })?;
+                reqwest::Client::new()
                     .post(self.dest.clone())
-                    .json(&value)
+                    .event(value)
+                    .map_err(|e| {
+                        tracing::warn!(error = ?e, "Failed to build request-builder")
+                    })?
+                    .header("Access-Control-Allow-Origin", "*")
                     .send()
-                    .await?;
-                if !resp.status().is_success() {
-                    tracing::warn!(
-                    cdevent = ?serde_json::to_value(&value)?,
-                    http_status = ?resp.status(),
-                    "failed to send event")
-                }
+                    .await
+                    .map_err(|e| {
+                        tracing::warn!(error = ?e, "Failed to get response")
+                    })?;
             }
             Err(err) => {
                 tracing::warn!(error = ?err, "Failed to convert to cloudevents");
+                // In error case, send the original event
+                self.client
+                    .post(self.dest.clone())
+                    .json(&cd_event)
+                    .send()
+                    .await?;
             }
         };
         Ok(())