move hyper client to its own crate along with tokio/hyper

Author: rochdev

collector/Cargo.lock

@@ -1,6 +1,7 @@
 [workspace]
 members = [
   "common",
+  "hyper-client",
   "server"
 ]
 

collector/Cargo.toml

@@ -6,8 +6,5 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-async-trait = "0.1.64"
 hashbrown = "0.13.2"
-hyper = { version = "0.14.24", features = ["full"] }
 rmp = "0.8.11"
-tokio = { version = "1.25.0", features = ["full"] }

collector/common/Cargo.toml

@@ -0,0 +1,4 @@
+// TODO: Support streaming with a writer instead of slice.
+pub trait Client {
+    fn request(&self, data: Vec<u8>);
+}

collector/common/src/client.rs

@@ -1,9 +1,7 @@
-use crate::tracing::{Traces};
-use async_trait::async_trait;
+use crate::tracing::Traces;
 
 pub mod agent;
 
-#[async_trait]
 pub trait Exporter {
-    async fn export(&self, traces: Traces);
+    fn export(&self, traces: Traces);
 }

collector/common/src/exporting.rs

@@ -1,19 +1,16 @@
+use crate::client::Client;
 use crate::tracing::{Trace, Traces};
 use super::Exporter;
-use async_trait::async_trait;
-use hyper::{Body, Client, Request};
-use hyper::client::HttpConnector;
 use rmp::encode;
 use rmp::encode::ByteBuf;
 use hashbrown::HashMap;
 
 pub struct AgentExporter {
-    client: Client<HttpConnector>
+    client: Box<dyn Client + Send + Sync>
 }
 
-#[async_trait]
 impl Exporter for AgentExporter {
-    async fn export(&self, traces: Traces) {
+    fn export(&self, traces: Traces) {
         let mut wr = ByteBuf::new();
         let trace_count = traces.len();
 
@@ -23,33 +20,18 @@ impl Exporter for AgentExporter {
             self.encode_traces(&mut wr, traces);
 
             let data: Vec<u8> = wr.as_vec().to_vec();
-            let req = Request::builder()
-                .method(hyper::Method::PUT)
-                .uri("http://localhost:8126/v0.4/traces")
-                .header("Content-Type", "application/msgpack")
-                .header("X-Datadog-Trace-Count", trace_count.to_string())
-                // .header("Datadog-Meta-Tracer-Version", "")
-                // .header("Datadog-Meta-Lang", "")
-                // .header("Datadog-Meta-Lang-Version", "")
-                // .header("Datadog-Meta-Lang-Interpreter", "")
-                .body(Body::from(data))
-                .unwrap();
-
-            self.client.request(req).await.unwrap();
-        }
-    }
-}
 
-impl Default for AgentExporter {
-    fn default() -> Self {
-        Self::new()
+            // TODO: Get the response somehow (with a channel?)
+            // TODO: Make client reusable between requests (with a channel?)
+            self.client.request(data);
+        }
     }
 }
 
 impl AgentExporter {
-    pub fn new() -> Self {
+    pub fn new(client: Box<dyn Client + Send + Sync>) -> Self {
         Self {
-            client: Client::new()
+            client
         }
     }
 

collector/common/src/exporting/agent.rs

@@ -1,3 +1,4 @@
+pub mod client;
 pub mod exporting;
 pub mod msgpack;
 pub mod processing;

collector/common/src/lib.rs

@@ -43,12 +43,12 @@ impl Processor {
         }
     }
 
-    pub async fn flush(&mut self) {
+    pub fn flush(&mut self) {
         let finished_traces: HashMap<u64, Trace> = self.traces
             .drain_filter(|_, v| v.started == v.finished)
             .collect();
 
-        self.exporter.export(finished_traces).await;
+        self.exporter.export(finished_traces);
     }
 
     fn process_event<R: Read>(&mut self, strings: &mut Vec<String>, mut rd: R) {

collector/common/src/processing.rs

@@ -0,0 +1,11 @@
+[package]
+name = "hyper-client"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+common = { path = "../common" }
+hyper = { version = "0.14.24", features = ["full"] }
+tokio = { version = "1.25.0", features = ["full"] }

collector/hyper-client/Cargo.toml

@@ -0,0 +1,36 @@
+use common::client::Client;
+
+pub struct HyperClient {}
+
+impl HyperClient {
+    pub fn new() -> Self {
+        Self {}
+    }
+}
+
+impl Default for HyperClient {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Client for HyperClient {
+    fn request(&self, data: Vec<u8>) {
+        // TODO: configuration options
+        let req = hyper::Request::builder()
+            .method(hyper::Method::PUT)
+            .uri("http://localhost:8126/v0.5/traces")
+            .header("Content-Type", "application/msgpack")
+            // .header("X-Datadog-Trace-Count", trace_count.to_string())
+            // .header("Datadog-Meta-Tracer-Version", "")
+            // .header("Datadog-Meta-Lang", "")
+            // .header("Datadog-Meta-Lang-Version", "")
+            // .header("Datadog-Meta-Lang-Interpreter", "")
+            .body(hyper::Body::from(data))
+            .unwrap();
+
+        tokio::spawn(async move {
+            hyper::Client::new().request(req).await.unwrap();
+        });
+    }
+}