Replace reqwest with ureq (#1407)

* Replace reqwest with ureq for less dependencies

* Simplify code

* Cleanup

Cargo.toml

@@ -68,7 +68,6 @@ polars-core = "0.27.1"
 polars-lazy = "0.27.1"
 polars-ops = "0.27.1"
 puffin = "0.14"
-reqwest = { version = "0.11", default-features = false }
 thiserror = "1.0"
 time = "0.3"
 tokio = "1.24"

crates/re_analytics/Cargo.toml

@@ -34,7 +34,9 @@ uuid = { version = "1.1", features = ["serde", "v4", "js"] }
 
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]
 directories-next = "2"
-reqwest = { workspace = true, features = ["blocking", "rustls-tls"] }
+ureq = { version = "2.6", features = [
+  "json",
+] } # TODO(emilk): use ehttp for web supprt
 
 [target.'cfg(target_arch = "wasm32")'.dependencies]
 web-sys = { version = "0.3.58", features = ["Storage"] }

crates/re_analytics/src/pipeline_native.rs

@@ -25,9 +25,6 @@ pub enum PipelineError {
 
     #[error(transparent)]
     Serde(#[from] serde_json::Error),
-
-    #[error(transparent)]
-    Http(#[from] reqwest::Error),
 }
 
 /// An eventual, at-least-once(-ish) event pipeline, backed by a write-ahead log on the local disk.

crates/re_analytics/src/sink_native.rs

@@ -1,11 +1,7 @@
-use std::{collections::HashMap, time::Duration};
+use std::collections::HashMap;
 
-use once_cell::sync::OnceCell;
-use reqwest::{blocking::Client as HttpClient, Url};
 use time::OffsetDateTime;
 
-use re_log::{debug, error};
-
 use crate::{Event, Property};
 
 // TODO(cmc): abstract away the concept of a `Sink` behind an actual trait when comes the time to
@@ -15,6 +11,9 @@ const PUBLIC_POSTHOG_PROJECT_KEY: &str = "phc_sgKidIE4WYYFSJHd8LEYY1UZqASpnfQKeM
 
 // ---
 
+#[derive(Debug, Clone)]
+struct Url(String);
+
 #[derive(thiserror::Error, Debug)]
 pub enum SinkError {
     #[error(transparent)]
@@ -24,19 +23,46 @@ pub enum SinkError {
     Serde(#[from] serde_json::Error),
 
     #[error(transparent)]
-    Http(#[from] reqwest::Error),
+    HttpTransport(Box<ureq::Transport>),
+
+    #[error("HTTP status {status_code} {status_text}: {body}")]
+    HttpStatus {
+        status_code: u16,
+        status_text: String,
+        body: String,
+    },
+}
+
+impl From<ureq::Error> for SinkError {
+    fn from(err: ureq::Error) -> Self {
+        match err {
+            ureq::Error::Status(status_code, response) => Self::HttpStatus {
+                status_code,
+                status_text: response.status_text().to_owned(),
+                body: response.into_string().unwrap_or_default(),
+            },
+
+            ureq::Error::Transport(transport) => Self::HttpTransport(Box::new(transport)),
+        }
+    }
 }
 
-#[derive(Default, Debug, Clone)]
+#[derive(Debug, Clone)]
 pub struct PostHogSink {
-    // NOTE: We need to lazily build the underlying HTTP client, so that we can guarantee that it
-    // is initialized from a thread that is free of a Tokio runtime.
-    // This is necessary because `reqwest` will crash if we try and initialize a blocking HTTP
-    // client from within a thread that has a Tokio runtime instantiated.
-    //
-    // We also use this opportunity to upgrade our public HTTP endpoint into the final HTTP2/TLS
-    // URL by following all 301 redirects.
-    client: OnceCell<(Url, HttpClient)>,
+    agent: ureq::Agent,
+    // Lazily resolve the url so that we don't do blocking requests in `PostHogSink::default`
+    resolved_url: once_cell::sync::OnceCell<String>,
+}
+
+impl Default for PostHogSink {
+    fn default() -> Self {
+        Self {
+            agent: ureq::AgentBuilder::new()
+                .timeout(std::time::Duration::from_secs(5))
+                .build(),
+            resolved_url: Default::default(),
+        }
+    }
 }
 
 impl PostHogSink {
@@ -50,42 +76,39 @@ impl PostHogSink {
         session_id: &str,
         events: &[Event],
     ) -> Result<(), SinkError> {
-        let (resolved_url, client) = self.init()?;
+        let resolved_url = self.init()?;
 
         let events = events
             .iter()
             .map(|event| PostHogEvent::from_event(analytics_id, session_id, event))
             .collect::<Vec<_>>();
         let batch = PostHogBatch::from_events(&events);
 
-        debug!("{}", serde_json::to_string_pretty(&batch)?);
-        let resp = client
-            .post(resolved_url.clone())
-            .body(serde_json::to_vec(&batch)?)
-            .send()?;
-
-        resp.error_for_status().map(|_| ()).map_err(Into::into)
+        re_log::debug!("{}", serde_json::to_string_pretty(&batch)?);
+        self.agent.post(resolved_url).send_json(&batch)?;
+        Ok(())
     }
 
-    fn init(&self) -> Result<&(Url, HttpClient), SinkError> {
-        self.client.get_or_try_init(|| {
-            use reqwest::header;
-            let mut headers = header::HeaderMap::new();
-            headers.insert(
-                header::CONTENT_TYPE,
-                header::HeaderValue::from_static("application/json"),
-            );
-
-            let client = HttpClient::builder()
-                .timeout(Duration::from_secs(5))
-                .connect_timeout(Duration::from_secs(5))
-                .pool_idle_timeout(Duration::from_secs(120))
-                .default_headers(headers)
-                .build()?;
-
-            let resolved_url = client.get(Self::URL).send()?.url().clone();
-
-            Ok((resolved_url, client))
+    fn init(&self) -> Result<&String, SinkError> {
+        self.resolved_url.get_or_try_init(|| {
+            // Make a dummy-request to resolve our final URL.
+            let resolved_url = match self.agent.get(Self::URL).call() {
+                Ok(response) => response.get_url().to_owned(),
+                Err(ureq::Error::Status(status, response)) => {
+                    // We actually expect to get here, because we make a bad request (GET to and end-point that expects a POST).
+                    // We only do this requests to get redirected to the final URL.
+                    let resolved_url = response.get_url().to_owned();
+                    re_log::trace!("status: {status} {}", response.status_text().to_owned());
+                    resolved_url
+                }
+                Err(ureq::Error::Transport(transport)) => {
+                    return Err(SinkError::HttpTransport(Box::new(transport)))
+                }
+            };
+
+            // 2023-02-26 the resolved URL was https://eu.posthog.com/capture (in Europe)
+
+            Ok(resolved_url)
         })
     }
 }