add worker

Cargo.lock

@@ -26,6 +26,56 @@ dependencies = [
  "memchr",
 ]
 
+[[package]]
+name = "anstream"
+version = "0.6.21"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "43d5b281e737544384e969a5ccad3f1cdd24b48086a0fc1b2a5262a26b8f4f4a"
+dependencies = [
+ "anstyle",
+ "anstyle-parse",
+ "anstyle-query",
+ "anstyle-wincon",
+ "colorchoice",
+ "is_terminal_polyfill",
+ "utf8parse",
+]
+
+[[package]]
+name = "anstyle"
+version = "1.0.13"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5192cca8006f1fd4f7237516f40fa183bb07f8fbdfedaa0036de5ea9b0b45e78"
+
+[[package]]
+name = "anstyle-parse"
+version = "0.2.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4e7644824f0aa2c7b9384579234ef10eb7efb6a0deb83f9630a49594dd9c15c2"
+dependencies = [
+ "utf8parse",
+]
+
+[[package]]
+name = "anstyle-query"
+version = "1.1.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9e231f6134f61b71076a3eab506c379d4f36122f2af15a9ff04415ea4c3339e2"
+dependencies = [
+ "windows-sys 0.60.2",
+]
+
+[[package]]
+name = "anstyle-wincon"
+version = "3.0.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3e0633414522a32ffaac8ac6cc8f748e090c5717661fddeea04219e2344f5f2a"
+dependencies = [
+ "anstyle",
+ "once_cell_polyfill",
+ "windows-sys 0.60.2",
+]
+
 [[package]]
 name = "async-compression"
 version = "0.4.30"
@@ -63,7 +113,7 @@ dependencies = [
  "miniz_oxide",
  "object",
  "rustc-demangle",
- "windows-targets",
+ "windows-targets 0.52.6",
 ]
 
 [[package]]
@@ -114,6 +164,52 @@ version = "1.0.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2fd1289c04a9ea8cb22300a459a72a385d7c73d3259e2ed7dcb2af674838cfa9"
 
+[[package]]
+name = "clap"
+version = "4.5.48"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e2134bb3ea021b78629caa971416385309e0131b351b25e01dc16fb54e1b5fae"
+dependencies = [
+ "clap_builder",
+ "clap_derive",
+]
+
+[[package]]
+name = "clap_builder"
+version = "4.5.48"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c2ba64afa3c0a6df7fa517765e31314e983f51dda798ffba27b988194fb65dc9"
+dependencies = [
+ "anstream",
+ "anstyle",
+ "clap_lex",
+ "strsim",
+]
+
+[[package]]
+name = "clap_derive"
+version = "4.5.47"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bbfd7eae0b0f1a6e63d4b13c9c478de77c2eb546fba158ad50b4203dc24b9f9c"
+dependencies = [
+ "heck",
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "clap_lex"
+version = "0.7.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b94f61472cee1439c0b966b47e3aca9ae07e45d070759512cd390ea2bebc6675"
+
+[[package]]
+name = "colorchoice"
+version = "1.0.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b05b61dc5112cbb17e4b6cd61790d9845d13888356391624cbe7e41efeac1e75"
+
 [[package]]
 name = "compact_str"
 version = "0.9.0"
@@ -416,6 +512,12 @@ version = "0.15.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1"
 
+[[package]]
+name = "heck"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea"
+
 [[package]]
 name = "http"
 version = "1.3.1"
@@ -680,6 +782,12 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "is_terminal_polyfill"
+version = "1.70.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7943c866cc5cd64cbc25b2e01621d07fa8eb2a1a23160ee81ce38704e97b8ecf"
+
 [[package]]
 name = "itoa"
 version = "1.0.15"
@@ -806,6 +914,12 @@ version = "1.21.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "42f5e15c9953c5e4ccceeb2e7382a716482c34515315f7b03532b8b4e8393d2d"
 
+[[package]]
+name = "once_cell_polyfill"
+version = "1.70.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a4895175b425cb1f87721b59f0f286c2092bd4af812243672510e1ac53e2e0ad"
+
 [[package]]
 name = "openssl"
 version = "0.10.73"
@@ -880,7 +994,7 @@ dependencies = [
  "libc",
  "redox_syscall",
  "smallvec",
- "windows-targets",
+ "windows-targets 0.52.6",
 ]
 
 [[package]]
@@ -1294,6 +1408,12 @@ version = "1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f"
 
+[[package]]
+name = "strsim"
+version = "0.11.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f"
+
 [[package]]
 name = "subtle"
 version = "2.6.1"
@@ -1568,9 +1688,11 @@ version = "0.1.0"
 dependencies = [
  "bytes",
  "compact_str",
+ "libc",
  "rand",
  "regex",
  "reqwest",
+ "socket2",
  "thiserror",
  "tokio",
  "tokio-util",
@@ -1579,6 +1701,21 @@ dependencies = [
  "wyrand",
 ]
 
+[[package]]
+name = "ubw-worker"
+version = "0.1.0"
+dependencies = [
+ "bytes",
+ "clap",
+ "compact_str",
+ "rand",
+ "reqwest",
+ "thiserror",
+ "tokio",
+ "tower",
+ "url",
+]
+
 [[package]]
 name = "unicode-ident"
 version = "1.0.18"
@@ -1609,6 +1746,12 @@ version = "1.0.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be"
 
+[[package]]
+name = "utf8parse"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"
+
 [[package]]
 name = "vcpkg"
 version = "0.2.15"
@@ -1745,13 +1888,19 @@ version = "0.1.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5e6ad25900d524eaabdbbb96d20b4311e1e7ae1699af4fb28c17ae66c80d798a"
 
+[[package]]
+name = "windows-link"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "45e46c0661abb7180e7b9c281db115305d49ca1709ab8242adf09666d2173c65"
+
 [[package]]
 name = "windows-registry"
 version = "0.5.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5b8a9ed28765efc97bbc954883f4e6796c33a06546ebafacbabee9696967499e"
 dependencies = [
- "windows-link",
+ "windows-link 0.1.3",
  "windows-result",
  "windows-strings",
 ]
@@ -1762,7 +1911,7 @@ version = "0.3.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "56f42bd332cc6c8eac5af113fc0c1fd6a8fd2aa08a0119358686e5160d0586c6"
 dependencies = [
- "windows-link",
+ "windows-link 0.1.3",
 ]
 
 [[package]]
@@ -1771,7 +1920,7 @@ version = "0.4.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "56e6c93f3a0c3b36176cb1327a4958a0353d5d166c2a35cb268ace15e91d3b57"
 dependencies = [
- "windows-link",
+ "windows-link 0.1.3",
 ]
 
 [[package]]
@@ -1780,7 +1929,7 @@ version = "0.52.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d"
 dependencies = [
- "windows-targets",
+ "windows-targets 0.52.6",
 ]
 
 [[package]]
@@ -1789,7 +1938,16 @@ version = "0.59.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b"
 dependencies = [
- "windows-targets",
+ "windows-targets 0.52.6",
+]
+
+[[package]]
+name = "windows-sys"
+version = "0.60.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f2f500e4d28234f72040990ec9d39e3a6b950f9f22d3dba18416c35882612bcb"
+dependencies = [
+ "windows-targets 0.53.4",
 ]
 
 [[package]]
@@ -1798,14 +1956,31 @@ version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973"
 dependencies = [
- "windows_aarch64_gnullvm",
+ "windows_aarch64_gnullvm 0.52.6",
- "windows_aarch64_msvc",
+ "windows_aarch64_msvc 0.52.6",
- "windows_i686_gnu",
+ "windows_i686_gnu 0.52.6",
- "windows_i686_gnullvm",
+ "windows_i686_gnullvm 0.52.6",
- "windows_i686_msvc",
+ "windows_i686_msvc 0.52.6",
- "windows_x86_64_gnu",
+ "windows_x86_64_gnu 0.52.6",
- "windows_x86_64_gnullvm",
+ "windows_x86_64_gnullvm 0.52.6",
- "windows_x86_64_msvc",
+ "windows_x86_64_msvc 0.52.6",
+]
+
+[[package]]
+name = "windows-targets"
+version = "0.53.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2d42b7b7f66d2a06854650af09cfdf8713e427a439c97ad65a6375318033ac4b"
+dependencies = [
+ "windows-link 0.2.0",
+ "windows_aarch64_gnullvm 0.53.0",
+ "windows_aarch64_msvc 0.53.0",
+ "windows_i686_gnu 0.53.0",
+ "windows_i686_gnullvm 0.53.0",
+ "windows_i686_msvc 0.53.0",
+ "windows_x86_64_gnu 0.53.0",
+ "windows_x86_64_gnullvm 0.53.0",
+ "windows_x86_64_msvc 0.53.0",
 ]
 
 [[package]]
@@ -1814,48 +1989,96 @@ version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3"
 
+[[package]]
+name = "windows_aarch64_gnullvm"
+version = "0.53.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "86b8d5f90ddd19cb4a147a5fa63ca848db3df085e25fee3cc10b39b6eebae764"
+
 [[package]]
 name = "windows_aarch64_msvc"
 version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469"
 
+[[package]]
+name = "windows_aarch64_msvc"
+version = "0.53.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c7651a1f62a11b8cbd5e0d42526e55f2c99886c77e007179efff86c2b137e66c"
+
 [[package]]
 name = "windows_i686_gnu"
 version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b"
 
+[[package]]
+name = "windows_i686_gnu"
+version = "0.53.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c1dc67659d35f387f5f6c479dc4e28f1d4bb90ddd1a5d3da2e5d97b42d6272c3"
+
 [[package]]
 name = "windows_i686_gnullvm"
 version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66"
 
+[[package]]
+name = "windows_i686_gnullvm"
+version = "0.53.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9ce6ccbdedbf6d6354471319e781c0dfef054c81fbc7cf83f338a4296c0cae11"
+
 [[package]]
 name = "windows_i686_msvc"
 version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66"
 
+[[package]]
+name = "windows_i686_msvc"
+version = "0.53.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "581fee95406bb13382d2f65cd4a908ca7b1e4c2f1917f143ba16efe98a589b5d"
+
 [[package]]
 name = "windows_x86_64_gnu"
 version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78"
 
+[[package]]
+name = "windows_x86_64_gnu"
+version = "0.53.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2e55b5ac9ea33f2fc1716d1742db15574fd6fc8dadc51caab1c16a3d3b4190ba"
+
 [[package]]
 name = "windows_x86_64_gnullvm"
 version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d"
 
+[[package]]
+name = "windows_x86_64_gnullvm"
+version = "0.53.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0a6e035dd0599267ce1ee132e51c27dd29437f63325753051e71dd9e42406c57"
+
 [[package]]
 name = "windows_x86_64_msvc"
 version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec"
 
+[[package]]
+name = "windows_x86_64_msvc"
+version = "0.53.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "271414315aff87387382ec3d271b52d7ae78726f5d44ac98b4f4030c91880486"
+
 [[package]]
 name = "wit-bindgen"
 version = "0.45.0"

Cargo.toml

@@ -1,6 +1,6 @@
 [workspace]
 resolver = "3"
-members = ["ubw-sward"]
+members = ["ubw-sward", "ubw-worker"]
 
 [workspace.dependencies]
 tokio = { version = "1", features = ["full"] }

ubw-sward/Cargo.toml

@@ -12,6 +12,8 @@ rand = {workspace = true}
 thiserror = {workspace = true}
 compact_str = {workspace = true}
 bytes = {workspace = true}
+libc = "0.2"
+socket2 = "0.6"
 wyrand = "0.3"
 regex = "1.11"
 tokio-util = "0.7"

ubw-sward/src/http/counter.rs

@@ -0,0 +1,34 @@
+use std::sync::atomic::AtomicUsize;
+
+#[derive(Default)]
+pub struct AtomicHttpCounter {
+    pub sent: AtomicUsize,
+    pub http_2xx: AtomicUsize,
+    pub http_3xx: AtomicUsize,
+    pub http_4xx: AtomicUsize,
+    pub http_5xx: AtomicUsize,
+    pub error: AtomicUsize,
+}
+
+impl AtomicHttpCounter {
+    pub fn read(&self) -> HttpCounter {
+        HttpCounter {
+            sent: self.sent.load(std::sync::atomic::Ordering::Relaxed),
+            http_2xx: self.http_2xx.load(std::sync::atomic::Ordering::Relaxed),
+            http_3xx: self.http_3xx.load(std::sync::atomic::Ordering::Relaxed),
+            http_4xx: self.http_4xx.load(std::sync::atomic::Ordering::Relaxed),
+            http_5xx: self.http_5xx.load(std::sync::atomic::Ordering::Relaxed),
+            error: self.error.load(std::sync::atomic::Ordering::Relaxed),
+        }
+    }
+}
+
+#[derive(Default, Clone, Copy)]
+pub struct HttpCounter {
+    pub sent: usize,
+    pub http_2xx: usize,
+    pub http_3xx: usize,
+    pub http_4xx: usize,
+    pub http_5xx: usize,
+    pub error: usize,
+}

ubw-sward/src/http/integrated.rs

@@ -1,12 +1,15 @@
+use crate::http::counter::AtomicHttpCounter;
 use crate::http::header_config::HeadersConfig;
 use crate::http::simple::HttpSwardArray;
-use crate::http::{RandomUrlGenerator, SimpleHttpSward};
+use crate::http::{RandomUrlGenerator, SimpleHttpRequest, SimpleHttpSward};
-use crate::utils::multiplexed::MultiplexedSward;
+use crate::utils::multiplexed::{MultiplexedSward, MultiplexedSwardError};
 use rand::Rng;
 use reqwest::Method;
 use std::net::{IpAddr, SocketAddr};
-use std::sync::Arc;
+use std::sync::{Arc, atomic};
 use thiserror::Error;
+use tower::ServiceExt;
+use tower::{BoxError, Service};
 use url::Url;
 use wyrand::WyRand;
 
@@ -19,6 +22,77 @@ enum AttackTarget {
 pub struct IntegratedHttpSward {
     attack_target: AttackTarget,
     request_sender: MultiplexedSward<HttpSwardArray<Box<[SimpleHttpSward]>>>,
+    result_counter: Arc<AtomicHttpCounter>,
+}
+
+impl IntegratedHttpSward {
+    pub async fn oneshot(&mut self) -> Result<reqwest::Response, HttpSwardError> {
+        self.result_counter
+            .sent
+            .fetch_add(1, atomic::Ordering::Relaxed);
+        let url = match &mut self.attack_target {
+            AttackTarget::Random(random) => random.generate_url()?,
+            AttackTarget::Fixed(url) => url.clone(),
+        };
+        let request = SimpleHttpRequest { body: None, url };
+        let res = self
+            .request_sender
+            .ready()
+            .await
+            .map_err(HttpSwardError::MultiplexError)?
+            .call(request)
+            .await
+            .map_err(HttpSwardError::MultiplexError)?;
+
+        match res.status().as_u16() {
+            200..=299 => {
+                self.result_counter
+                    .http_2xx
+                    .fetch_add(1, atomic::Ordering::Relaxed);
+            }
+            300..=399 => {
+                self.result_counter
+                    .http_3xx
+                    .fetch_add(1, atomic::Ordering::Relaxed);
+            }
+            400..=499 => {
+                self.result_counter
+                    .http_4xx
+                    .fetch_add(1, atomic::Ordering::Relaxed);
+            }
+            500..=599 => {
+                self.result_counter
+                    .http_5xx
+                    .fetch_add(1, atomic::Ordering::Relaxed);
+            }
+            _ => {}
+        }
+
+        Ok(res)
+    }
+
+    pub async fn run_with_signal(
+        mut self,
+        mut signal: tokio::sync::oneshot::Receiver<()>,
+    ) -> Result<(), tokio::sync::oneshot::error::RecvError> {
+        loop {
+            tokio::select! {
+                receive_result = &mut signal => {
+                    receive_result?;
+                    return Ok(());
+                }
+                oneshot_result = self.oneshot() => {
+                    if oneshot_result.is_err() {
+                        self.result_counter.error.fetch_and(1, atomic::Ordering::Relaxed);
+                    }
+                }
+            }
+        }
+    }
+
+    pub fn get_counter(&self) -> Arc<AtomicHttpCounter> {
+        self.result_counter.clone()
+    }
 }
 
 impl IntegratedHttpSward {
@@ -92,7 +166,7 @@ impl IntegratedHttpSwardBuilder {
         // get the cookie
         let maybe_cookie = self
             .headers_config
-            .get_cookie_jar(&url_example)
+            .get_cookie_jar(url_example)
             .map(Arc::new);
 
         // build the clients
@@ -132,6 +206,7 @@ impl IntegratedHttpSwardBuilder {
         Ok(IntegratedHttpSward {
             attack_target: AttackTarget::Fixed(url),
             request_sender: MultiplexedSward::new(SimpleHttpSward::array(swards), capacity),
+            result_counter: Arc::new(AtomicHttpCounter::default()),
         })
     }
 
@@ -162,6 +237,7 @@ impl IntegratedHttpSwardBuilder {
         Ok(IntegratedHttpSward {
             attack_target: AttackTarget::Random(random_url_generator),
             request_sender: MultiplexedSward::new(SimpleHttpSward::array(swards), capacity),
+            result_counter: Arc::new(AtomicHttpCounter::default()),
         })
     }
 }
@@ -180,3 +256,12 @@ pub enum HttpSwardBuildError {
     #[error("Failed to build reqwest client {0}")]
     ReqwestBuildError(#[from] reqwest::Error),
 }
+
+#[derive(Debug, Error)]
+pub enum HttpSwardError {
+    #[error("Error when pooling")]
+    MultiplexError(MultiplexedSwardError<BoxError>),
+
+    #[error(transparent)]
+    BadUrl(#[from] url::ParseError),
+}

ubw-sward/src/http/mod.rs

@@ -1,8 +1,9 @@
+pub mod counter;
+pub mod header_config;
+pub mod integrated;
 pub mod random;
 pub mod simple;
-pub mod integrated;
-pub mod header_config;
 
+pub use integrated::IntegratedHttpSward;
 pub use random::RandomUrlGenerator;
 pub use simple::{SimpleHttpRequest, SimpleHttpSward};
-pub use integrated::IntegratedHttpSward;

ubw-sward/src/http/simple.rs

@@ -1,6 +1,7 @@
 use bytes::Bytes;
 use reqwest::header::HeaderMap;
 use reqwest::{Client, Method};
+use std::future::Future;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 use tower::Service;
@@ -52,7 +53,7 @@ pub struct SimpleHttpRequest {
 impl Service<SimpleHttpRequest> for SimpleHttpSward {
     type Response = reqwest::Response;
     type Error = reqwest::Error;
-    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>>>>;
+    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
     fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         Poll::Ready(Ok(()))
     }

ubw-sward/src/lib.rs

@@ -3,4 +3,5 @@
 #![deny(clippy::expect_used)]
 
 pub mod http;
+pub mod socket_stream;
 pub mod utils;

ubw-sward/src/socket_stream/counter.rs

@@ -0,0 +1,25 @@
+use std::sync::atomic::AtomicU64;
+
+#[derive(Default, Clone, Copy)]
+pub struct SocketStreamCounter {
+    pub sent_bytes: u64,
+    pub sent_packets: u64,
+    pub error: u64,
+}
+
+#[derive(Default)]
+pub struct AtomicSocketStreamCounter {
+    pub sent_bytes: AtomicU64,
+    pub sent_packets: AtomicU64,
+    pub error: AtomicU64,
+}
+
+impl AtomicSocketStreamCounter {
+    pub fn read(&self) -> SocketStreamCounter {
+        SocketStreamCounter {
+            sent_bytes: self.sent_bytes.load(std::sync::atomic::Ordering::Relaxed),
+            sent_packets: self.sent_packets.load(std::sync::atomic::Ordering::Relaxed),
+            error: self.error.load(std::sync::atomic::Ordering::Relaxed),
+        }
+    }
+}

ubw-sward/src/socket_stream/generator.rs

@@ -0,0 +1,23 @@
+use rand::RngCore;
+
+use crate::socket_stream::{BoxedStreamWorkload, SizedStreamWorkload};
+
+pub struct RandomWorkloadGenerator<Rng: RngCore> {
+    rng: Rng,
+}
+
+impl<Rng: RngCore> RandomWorkloadGenerator<Rng> {
+    pub fn new(rng: Rng) -> Self {
+        Self { rng }
+    }
+    pub fn generate_sized<const SIZE: usize>(&mut self) -> SizedStreamWorkload<SIZE> {
+        let mut buf = [0u8; SIZE];
+        self.rng.fill_bytes(&mut buf);
+        SizedStreamWorkload { bytes: buf }
+    }
+    pub fn generate_boxed(&mut self, size: usize) -> BoxedStreamWorkload {
+        let mut buf = vec![0u8; size].into_boxed_slice();
+        self.rng.fill_bytes(&mut buf);
+        BoxedStreamWorkload(buf)
+    }
+}

ubw-sward/src/socket_stream/integrated.rs

@@ -0,0 +1,69 @@
+use rand::Rng;
+use std::net::SocketAddr;
+use std::sync::{Arc, atomic};
+use wyrand::WyRand;
+
+use crate::socket_stream::StreamSward;
+use crate::socket_stream::counter::AtomicSocketStreamCounter;
+use crate::socket_stream::generator::RandomWorkloadGenerator;
+
+pub struct IntegratedSocketStreamSward<Swd: StreamSward> {
+    sward: Swd,
+    random_workload_generator: RandomWorkloadGenerator<WyRand>,
+    counter: Arc<AtomicSocketStreamCounter>,
+}
+
+impl<Swd: StreamSward> IntegratedSocketStreamSward<Swd> {
+    pub async fn new(target: SocketAddr) -> Result<Self, std::io::Error> {
+        let sward = Swd::connect(target).await?;
+        let random_workload_generator =
+            RandomWorkloadGenerator::new(WyRand::new(rand::rng().random()));
+        let counter = Arc::new(AtomicSocketStreamCounter::default());
+        Ok(Self {
+            sward,
+            random_workload_generator,
+            counter,
+        })
+    }
+    pub async fn oneshot_array<const N: usize>(&mut self) -> Result<usize, std::io::Error> {
+        self.counter
+            .sent_bytes
+            .fetch_add(N as u64, atomic::Ordering::Relaxed);
+        self.counter
+            .sent_packets
+            .fetch_add(1, atomic::Ordering::Relaxed);
+        let content = self.random_workload_generator.generate_sized::<N>();
+        self.sward.send_sized(content).await
+    }
+    pub async fn oneshot_dynamic(&mut self, size: usize) -> Result<usize, std::io::Error> {
+        self.counter
+            .sent_bytes
+            .fetch_add(size as u64, atomic::Ordering::Relaxed);
+        self.counter
+            .sent_packets
+            .fetch_add(1, atomic::Ordering::Relaxed);
+        let content = self.random_workload_generator.generate_boxed(size);
+        self.sward.send_boxed(content).await
+    }
+    pub async fn run_with_signal<const SIZE: usize>(
+        mut self,
+        mut signal: tokio::sync::oneshot::Receiver<()>,
+    ) -> Result<(), tokio::sync::oneshot::error::RecvError> {
+        loop {
+            tokio::select! {
+                receive_result = &mut signal => {
+                    receive_result?;
+                    return Ok(())
+                },
+                oneshot_result = self.oneshot_array::<SIZE>() => {
+                    if oneshot_result.is_err() {
+                        self.counter.error.fetch_add(1, atomic::Ordering::Relaxed);
+                    }
+                }
+            }
+        }
+    }
+    pub fn get_counter(&self) -> Arc<AtomicSocketStreamCounter> {
+        self.counter.clone()
+    }
+}

ubw-sward/src/socket_stream/mod.rs

@@ -0,0 +1,37 @@
+use std::pin::Pin;
+
+pub mod counter;
+pub mod generator;
+pub mod integrated;
+pub mod tcp;
+pub mod udp;
+
+#[derive(Clone, Copy)]
+pub struct SizedStreamWorkload<const N: usize> {
+    pub bytes: [u8; N],
+}
+
+#[derive(Clone)]
+pub struct BoxedStreamWorkload(pub Box<[u8]>);
+
+pub type StreamSendFuture = Pin<Box<dyn Future<Output = Result<usize, std::io::Error>> + Send>>;
+
+pub trait StreamSward {
+    fn connect(
+        addr: std::net::SocketAddr,
+    ) -> impl Future<Output = Result<Self, std::io::Error>> + Send
+    where
+        Self: Sized;
+    fn add_request_count(&mut self);
+    fn send_sized<const N: usize>(
+        &self,
+        workload: SizedStreamWorkload<N>,
+    ) -> impl Future<Output = Result<usize, std::io::Error>> + Send + 'static;
+    fn send_boxed(
+        &self,
+        workload: BoxedStreamWorkload,
+    ) -> impl Future<Output = Result<usize, std::io::Error>> + Send + 'static;
+}
+
+pub type IntegratedUdpSward = integrated::IntegratedSocketStreamSward<udp::UdpSward>;
+pub type IntegratedTcpSward = integrated::IntegratedSocketStreamSward<tcp::TcpSward>;

ubw-sward/src/socket_stream/tcp.rs

@@ -0,0 +1,95 @@
+use std::{
+    sync::Arc,
+    task::{Context, Poll},
+};
+use tokio::{io::AsyncWriteExt, net::TcpStream, sync::Mutex};
+use tower::Service;
+
+use crate::socket_stream::{
+    BoxedStreamWorkload, SizedStreamWorkload, StreamSendFuture, StreamSward,
+};
+
+pub struct TcpSward {
+    stream: Arc<Mutex<TcpStream>>,
+    sent_count: usize,
+}
+
+impl TcpSward {
+    pub fn new(stream: Arc<Mutex<TcpStream>>) -> Self {
+        Self {
+            stream,
+            sent_count: 0,
+        }
+    }
+}
+
+impl StreamSward for TcpSward {
+    async fn connect(addr: std::net::SocketAddr) -> Result<Self, std::io::Error> {
+        let stream = TcpStream::connect(addr).await?;
+        let stream = Arc::new(Mutex::new(stream));
+        Ok(Self {
+            stream,
+            sent_count: 0,
+        })
+    }
+    fn add_request_count(&mut self) {
+        self.sent_count += 1;
+    }
+    fn send_sized<const N: usize>(
+        &self,
+        workload: SizedStreamWorkload<N>,
+    ) -> impl Future<Output = Result<usize, std::io::Error>> + Send + 'static {
+        let stream = self.stream.clone();
+        async move {
+            let mut stream = stream.lock().await;
+            stream.write_all(&workload.bytes).await?;
+            Ok(N)
+        }
+    }
+    fn send_boxed(
+        &self,
+        workload: BoxedStreamWorkload,
+    ) -> impl Future<Output = Result<usize, std::io::Error>> + Send + 'static {
+        let stream = self.stream.clone();
+        async move {
+            let len = workload.0.len();
+            let mut stream = stream.lock().await;
+            stream.write_all(&workload.0).await?;
+            Ok(len)
+        }
+    }
+}
+
+impl<const N: usize> Service<SizedStreamWorkload<N>> for TcpSward {
+    type Response = usize;
+    type Error = std::io::Error;
+    type Future = StreamSendFuture;
+    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+    fn call(&mut self, req: SizedStreamWorkload<N>) -> Self::Future {
+        self.add_request_count();
+        Box::pin(self.send_sized(req))
+    }
+}
+
+impl Service<BoxedStreamWorkload> for TcpSward {
+    type Response = usize;
+    type Error = std::io::Error;
+    type Future = StreamSendFuture;
+    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+    fn call(&mut self, req: BoxedStreamWorkload) -> Self::Future {
+        self.add_request_count();
+        Box::pin(self.send_boxed(req))
+    }
+}
+
+impl tower::load::Load for TcpSward {
+    type Metric = usize;
+    fn load(&self) -> Self::Metric {
+        self.sent_count
+    }
+}
+

ubw-sward/src/socket_stream/udp.rs

@@ -0,0 +1,86 @@
+use std::{
+    sync::Arc,
+    task::{Context, Poll},
+};
+use tokio::net::UdpSocket;
+use tower::Service;
+
+use crate::socket_stream::{
+    BoxedStreamWorkload, SizedStreamWorkload, StreamSendFuture, StreamSward,
+};
+
+pub struct UdpSward {
+    socket: Arc<UdpSocket>,
+    sent_count: usize,
+}
+
+impl UdpSward {
+    pub fn new(socket: Arc<UdpSocket>) -> Self {
+        Self {
+            socket,
+            sent_count: 0,
+        }
+    }
+}
+
+impl StreamSward for UdpSward {
+    async fn connect(addr: std::net::SocketAddr) -> Result<Self, std::io::Error> {
+        let socket = UdpSocket::bind("0.0.0.0:0").await?;
+        socket.connect(addr).await?;
+        let socket = Arc::new(socket);
+        Ok(Self {
+            socket,
+            sent_count: 0,
+        })
+    }
+    fn add_request_count(&mut self) {
+        self.sent_count += 1;
+    }
+    fn send_sized<const N: usize>(
+        &self,
+        workload: SizedStreamWorkload<N>,
+    ) -> impl Future<Output = Result<usize, std::io::Error>> + Send + 'static {
+        let socket = self.socket.clone();
+        async move { socket.send(&workload.bytes).await }
+    }
+    fn send_boxed(
+        &self,
+        workload: BoxedStreamWorkload,
+    ) -> impl Future<Output = Result<usize, std::io::Error>> + Send + 'static {
+        let socket = self.socket.clone();
+        async move { socket.send(&workload.0).await }
+    }
+}
+
+impl<const N: usize> Service<SizedStreamWorkload<N>> for UdpSward {
+    type Response = usize;
+    type Error = std::io::Error;
+    type Future = StreamSendFuture;
+    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+    fn call(&mut self, req: SizedStreamWorkload<N>) -> Self::Future {
+        self.add_request_count();
+        Box::pin(self.send_sized(req))
+    }
+}
+
+impl Service<BoxedStreamWorkload> for UdpSward {
+    type Response = usize;
+    type Error = std::io::Error;
+    type Future = StreamSendFuture;
+    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+    fn call(&mut self, req: BoxedStreamWorkload) -> Self::Future {
+        self.add_request_count();
+        Box::pin(self.send_boxed(req))
+    }
+}
+
+impl tower::load::Load for UdpSward {
+    type Metric = usize;
+    fn load(&self) -> Self::Metric {
+        self.sent_count
+    }
+}

ubw-sward/src/utils/multiplexed.rs

@@ -64,7 +64,7 @@ where
 /// Error that can occur when multiplexing requests.
 #[derive(Debug, thiserror::Error)]
 pub enum MultiplexedSwardError<E> {
-    #[error("{0}")]
+    #[error(transparent)]
     Inner(E),
 
     #[error("Semaphore is closed.")]

ubw-worker/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "ubw-worker"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+tokio = { workspace = true }
+tower = { workspace = true }
+reqwest = {workspace = true}
+url = {workspace = true}
+rand = {workspace = true}
+thiserror = {workspace = true}
+compact_str = {workspace = true}
+bytes = {workspace = true}
+clap = {version = "4.5", features = ["derive"]}

ubw-worker/src/main.rs

@@ -0,0 +1,3 @@
+fn main() {
+    println!("Hello, world!");
+}