feat: http crate (#3)

Cargo.toml

@@ -18,18 +18,21 @@ full = [
   "profiler",
   "collections",
   "future",
+  "http",
   "metrics",
 ]
 alloc = ["dep:alloc"]
 collections = ["dep:collections"]
 future = ["dep:future"]
-metrics = ["dep:metrics", "future/metrics", "alloc/metrics"]
+http = []
+metrics = ["dep:metrics", "future/metrics", "alloc/metrics", "http/metrics"]
 profiler = ["alloc/profiler"]
 
 [dependencies]
 alloc = { path = "./crates/alloc", optional = true }
 collections = { path = "./crates/collections", optional = true }
 future = { path = "./crates/future", optional = true }
+http = { path = "./crates/http", optional = true }
 metrics = { path = "./crates/metrics", optional = true }
 
 [dev-dependencies]

README.md

@@ -14,6 +14,10 @@ Extensions for collections such as `HashMap`.
 
 Convenience `Future` extensions.
 
+## `http`
+
+Metrics and other utils for HTTP servers.
+
 ## `metrics`
 
 Global service metrics. Currently based on `opentelemetry` SDK and exported in `prometheus` format.

crates/http/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "http"
+version = "0.1.0"
+edition = "2021"
+
+[features]
+default = []
+full = ["metrics"]
+metrics = ["dep:metrics", "dep:future"]
+
+[dependencies]
+future = { path = "../future", features = ["metrics"], optional = true }
+metrics = { path = "../metrics", optional = true }
+hyper = "0.14"
+tokio = { version = "1", default-features = false, features = ["rt-multi-thread", "time", "macros"] }

crates/http/src/executor.rs

@@ -0,0 +1,59 @@
+use {
+    future::FutureExt,
+    metrics::TaskMetrics,
+    std::{future::Future, time::Duration},
+};
+
+/// Global `hyper` service task executor that uses the `tokio` runtime and adds
+/// metrics for the executed tasks.
+#[derive(Default, Clone)]
+pub struct ServiceTaskExecutor {
+    timeout: Option<Duration>,
+    metrics_name: &'static str,
+}
+
+impl ServiceTaskExecutor {
+    pub fn new() -> Self {
+        Default::default()
+    }
+
+    /// Optional `task_name` metrics attribute.
+    pub fn name(self, metrics_name: Option<&'static str>) -> Self {
+        Self {
+            timeout: self.timeout,
+            metrics_name: metrics_name.unwrap_or(""),
+        }
+    }
+
+    /// Apply a timeout to all service tasks to prevent them from becoming
+    /// zombies for various reasons.
+    ///
+    /// Default is no timeout.
+    pub fn timeout(self, timeout: Option<Duration>) -> Self {
+        Self {
+            timeout,
+            metrics_name: self.metrics_name,
+        }
+    }
+}
+
+impl<F> hyper::rt::Executor<F> for ServiceTaskExecutor
+where
+    F: Future + Send + 'static,
+    F::Output: Send + 'static,
+{
+    fn execute(&self, fut: F) {
+        static METRICS: TaskMetrics = TaskMetrics::new("hyper_service_task");
+
+        let fut = fut.with_metrics(METRICS.with_name(self.metrics_name));
+        let timeout = self.timeout;
+
+        tokio::spawn(async move {
+            if let Some(timeout) = timeout {
+                let _ = fut.with_timeout(timeout).await;
+            } else {
+                fut.await;
+            }
+        });
+    }
+}

crates/http/src/lib.rs

@@ -0,0 +1,5 @@
+#[cfg(feature = "metrics")]
+mod executor;
+
+#[cfg(feature = "metrics")]
+pub use executor::*;

src/lib.rs

@@ -5,5 +5,7 @@ pub use alloc;
 pub use collections;
 #[cfg(feature = "future")]
 pub use future;
+#[cfg(feature = "http")]
+pub use http;
 #[cfg(feature = "metrics")]
 pub use metrics;