Compare commits
5 Commits
0a790ccaaa
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
386ed710d9 | ||
|
|
6732647655 | ||
|
|
7ad031c6a9 | ||
|
|
519f75803d | ||
|
|
13ccb14826 |
219
Cargo.lock
generated
219
Cargo.lock
generated
@@ -2,6 +2,34 @@
|
||||
# It is not intended for manual editing.
|
||||
version = 4
|
||||
|
||||
[[package]]
|
||||
name = "ahash"
|
||||
version = "0.8.12"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"getrandom 0.3.4",
|
||||
"once_cell",
|
||||
"version_check",
|
||||
"zerocopy",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "aho-corasick"
|
||||
version = "1.1.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ddd31a130427c27518df266943a5308ed92d4b226cc639f5a8f1002816174301"
|
||||
dependencies = [
|
||||
"memchr",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "allocator-api2"
|
||||
version = "0.2.21"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923"
|
||||
|
||||
[[package]]
|
||||
name = "android_system_properties"
|
||||
version = "0.1.5"
|
||||
@@ -67,6 +95,12 @@ version = "1.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8"
|
||||
|
||||
[[package]]
|
||||
name = "base64"
|
||||
version = "0.22.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6"
|
||||
|
||||
[[package]]
|
||||
name = "bitflags"
|
||||
version = "2.10.0"
|
||||
@@ -129,6 +163,15 @@ dependencies = [
|
||||
"windows-link",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "chumsky"
|
||||
version = "0.9.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8eebd66744a15ded14960ab4ccdbfb51ad3b81f51f3f04a80adac98c985396c9"
|
||||
dependencies = [
|
||||
"hashbrown 0.14.5",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "clap"
|
||||
version = "4.5.55"
|
||||
@@ -237,6 +280,18 @@ dependencies = [
|
||||
"syn",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "dyn-clone"
|
||||
version = "1.0.20"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555"
|
||||
|
||||
[[package]]
|
||||
name = "equivalent"
|
||||
version = "1.0.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f"
|
||||
|
||||
[[package]]
|
||||
name = "errno"
|
||||
version = "0.3.14"
|
||||
@@ -360,12 +415,34 @@ dependencies = [
|
||||
"wasip2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "hashbrown"
|
||||
version = "0.14.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1"
|
||||
dependencies = [
|
||||
"ahash",
|
||||
"allocator-api2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "hashbrown"
|
||||
version = "0.16.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100"
|
||||
|
||||
[[package]]
|
||||
name = "heck"
|
||||
version = "0.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea"
|
||||
|
||||
[[package]]
|
||||
name = "hifijson"
|
||||
version = "0.2.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0a7763b98ba8a24f59e698bf9ab197e7676c640d6455d1580b4ce7dc560f0f0d"
|
||||
|
||||
[[package]]
|
||||
name = "http"
|
||||
version = "1.4.0"
|
||||
@@ -508,6 +585,16 @@ dependencies = [
|
||||
"icu_properties",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "indexmap"
|
||||
version = "2.13.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7714e70437a7dc3ac8eb7e6f8df75fd8eb422675fc7678aff7364301092b1017"
|
||||
dependencies = [
|
||||
"equivalent",
|
||||
"hashbrown 0.16.1",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "is_terminal_polyfill"
|
||||
version = "1.70.2"
|
||||
@@ -520,6 +607,66 @@ version = "1.0.17"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "92ecc6618181def0457392ccd0ee51198e065e016d1d527a7ac1b6dc7c1f09d2"
|
||||
|
||||
[[package]]
|
||||
name = "jaq-core"
|
||||
version = "1.5.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d6fda09ee08c84c81293fdf811d9ebaa87b327557b5391f290c926d728c2ddd4"
|
||||
dependencies = [
|
||||
"aho-corasick",
|
||||
"base64",
|
||||
"chrono",
|
||||
"hifijson",
|
||||
"jaq-interpret",
|
||||
"libm",
|
||||
"log",
|
||||
"regex",
|
||||
"urlencoding",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "jaq-interpret"
|
||||
version = "1.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2fe95ec3c24af3fd9f3dd1091593f5e49b003a66c496a8aa39d764d0a06ae17b"
|
||||
dependencies = [
|
||||
"ahash",
|
||||
"dyn-clone",
|
||||
"hifijson",
|
||||
"indexmap",
|
||||
"jaq-syn",
|
||||
"once_cell",
|
||||
"serde_json",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "jaq-parse"
|
||||
version = "1.0.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0346d7d3146cdda8acd929581f3d6626a332356c74d5c95aeaffaac2eb6dee82"
|
||||
dependencies = [
|
||||
"chumsky",
|
||||
"jaq-syn",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "jaq-std"
|
||||
version = "1.6.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "bfbaa55578fd3b70433b594a370741e0c364e4afff92cc0099623fce87311bc1"
|
||||
dependencies = [
|
||||
"jaq-syn",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "jaq-syn"
|
||||
version = "1.6.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1ba44fe4428c71304604261ecbae047ee9cfb60c4f1a6bd222ebbb31726d3948"
|
||||
dependencies = [
|
||||
"serde",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "js-sys"
|
||||
version = "0.3.85"
|
||||
@@ -542,6 +689,12 @@ version = "0.2.180"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "bcc35a38544a891a5f7c865aca548a982ccb3b8650a5b06d0fd33a10283c56fc"
|
||||
|
||||
[[package]]
|
||||
name = "libm"
|
||||
version = "0.2.16"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981"
|
||||
|
||||
[[package]]
|
||||
name = "linux-raw-sys"
|
||||
version = "0.11.0"
|
||||
@@ -569,6 +722,12 @@ version = "0.4.29"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897"
|
||||
|
||||
[[package]]
|
||||
name = "memchr"
|
||||
version = "2.7.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f52b00d39961fc5b2736ea853c9cc86238e165017a493d1d5c8eac6bdc4cc273"
|
||||
|
||||
[[package]]
|
||||
name = "mio"
|
||||
version = "1.1.1"
|
||||
@@ -799,6 +958,35 @@ dependencies = [
|
||||
"bitflags",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "regex"
|
||||
version = "1.12.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e10754a14b9137dd7b1e3e5b0493cc9171fdd105e0ab477f51b72e7f3ac0e276"
|
||||
dependencies = [
|
||||
"aho-corasick",
|
||||
"memchr",
|
||||
"regex-automata",
|
||||
"regex-syntax",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "regex-automata"
|
||||
version = "0.4.14"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6e1dd4122fc1595e8162618945476892eefca7b88c52820e74af6262213cae8f"
|
||||
dependencies = [
|
||||
"aho-corasick",
|
||||
"memchr",
|
||||
"regex-syntax",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "regex-syntax"
|
||||
version = "0.8.9"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a96887878f22d7bad8a3b6dc5b7440e0ada9a245242924394987b21cf2210a4c"
|
||||
|
||||
[[package]]
|
||||
name = "rustix"
|
||||
version = "1.1.3"
|
||||
@@ -863,6 +1051,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e"
|
||||
dependencies = [
|
||||
"serde_core",
|
||||
"serde_derive",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -885,6 +1074,19 @@ dependencies = [
|
||||
"syn",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "serde_json"
|
||||
version = "1.0.149"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "83fc039473c5595ace860d8c4fafa220ff474b3fc6bfdb4293327f1a37e94d86"
|
||||
dependencies = [
|
||||
"itoa",
|
||||
"memchr",
|
||||
"serde",
|
||||
"serde_core",
|
||||
"zmij",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "sha1"
|
||||
version = "0.10.6"
|
||||
@@ -1182,6 +1384,12 @@ dependencies = [
|
||||
"serde",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "urlencoding"
|
||||
version = "2.1.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "daf8dba3b7eb870caf1ddeed7bc9d2a049f3cfdfae7cb521b087cc33ae4c49da"
|
||||
|
||||
[[package]]
|
||||
name = "utf-8"
|
||||
version = "0.7.6"
|
||||
@@ -1285,6 +1493,11 @@ dependencies = [
|
||||
"chrono",
|
||||
"clap",
|
||||
"futures-util",
|
||||
"jaq-core",
|
||||
"jaq-interpret",
|
||||
"jaq-parse",
|
||||
"jaq-std",
|
||||
"serde_json",
|
||||
"tokio",
|
||||
"tokio-tungstenite",
|
||||
"tracing",
|
||||
@@ -1542,3 +1755,9 @@ dependencies = [
|
||||
"quote",
|
||||
"syn",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "zmij"
|
||||
version = "1.0.19"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3ff05f8caa9038894637571ae6b9e29466c1f4f829d26c9b28f869a29cbe3445"
|
||||
|
||||
@@ -12,3 +12,8 @@ clap = { version = "4", features = ["derive"] }
|
||||
futures-util = "0.3"
|
||||
chrono = "0.4"
|
||||
url = "2"
|
||||
jaq-core = "1"
|
||||
jaq-interpret = "1"
|
||||
jaq-parse = "1"
|
||||
jaq-std = "1"
|
||||
serde_json = "1"
|
||||
|
||||
30
src/cli.rs
Normal file
30
src/cli.rs
Normal file
@@ -0,0 +1,30 @@
|
||||
use clap::Parser;
|
||||
|
||||
#[derive(Parser)]
|
||||
#[command(name = "websocket-debug")]
|
||||
#[command(about = "A WebSocket debugging tool that logs and saves messages")]
|
||||
pub struct Args {
|
||||
/// WebSocket URLs to connect to (e.g., ws://localhost:8080 or wss://example.com/ws)
|
||||
#[arg(required = true)]
|
||||
pub urls: Vec<String>,
|
||||
|
||||
/// Bearer token for Authorization header
|
||||
#[arg(long)]
|
||||
pub bearer_token: Option<String>,
|
||||
|
||||
/// Enable debug logging (shows request/response headers)
|
||||
#[arg(long)]
|
||||
pub debug: bool,
|
||||
|
||||
/// Query string parameters to add to all URLs (pre-encoded, e.g., "name=First%20Last&key=value")
|
||||
#[arg(short = 'q', long = "query-string-all")]
|
||||
pub query_string_all: Option<String>,
|
||||
|
||||
/// jq expression(s) to evaluate on JSON text messages for logging (can be specified multiple times)
|
||||
#[arg(short = 'j', long = "jaq")]
|
||||
pub jaq: Vec<String>,
|
||||
|
||||
/// Normalize JSON messages: save as .json with pretty-printing and sorted keys (for easier diffing)
|
||||
#[arg(short = 'n', long = "json-normalize")]
|
||||
pub json_normalize: bool,
|
||||
}
|
||||
165
src/connection.rs
Normal file
165
src/connection.rs
Normal file
@@ -0,0 +1,165 @@
|
||||
use tokio_tungstenite::{
|
||||
connect_async_with_config, MaybeTlsStream, WebSocketStream,
|
||||
tungstenite::{
|
||||
client::IntoClientRequest,
|
||||
http::header::{HeaderValue, AUTHORIZATION, USER_AGENT},
|
||||
},
|
||||
};
|
||||
use tracing::{debug, info};
|
||||
use url::{form_urlencoded, Url};
|
||||
|
||||
/// Process URLs by adding extra query parameters if specified.
|
||||
pub fn process_urls(
|
||||
urls: &[String],
|
||||
query_string_all: Option<&String>,
|
||||
) -> Result<Vec<String>, url::ParseError> {
|
||||
let extra_params: Vec<(String, String)> = query_string_all
|
||||
.map(|qs| {
|
||||
form_urlencoded::parse(qs.as_bytes())
|
||||
.map(|(k, v)| (k.into_owned(), v.into_owned()))
|
||||
.collect()
|
||||
})
|
||||
.unwrap_or_default();
|
||||
|
||||
let mut processed_urls = Vec::new();
|
||||
for url_str in urls {
|
||||
let mut url = Url::parse(url_str)?;
|
||||
|
||||
if !extra_params.is_empty() {
|
||||
let existing: Vec<(String, String)> = url
|
||||
.query_pairs()
|
||||
.map(|(k, v)| (k.into_owned(), v.into_owned()))
|
||||
.collect();
|
||||
|
||||
let mut query_pairs = url.query_pairs_mut();
|
||||
query_pairs.clear();
|
||||
for (k, v) in &existing {
|
||||
query_pairs.append_pair(k, v);
|
||||
}
|
||||
for (k, v) in &extra_params {
|
||||
query_pairs.append_pair(k, v);
|
||||
}
|
||||
}
|
||||
|
||||
processed_urls.push(url.to_string());
|
||||
}
|
||||
|
||||
Ok(processed_urls)
|
||||
}
|
||||
|
||||
/// Connect to a WebSocket URL with optional bearer token and debug logging.
|
||||
pub async fn connect(
|
||||
url: &str,
|
||||
idx: usize,
|
||||
bearer_token: Option<&str>,
|
||||
debug_enabled: bool,
|
||||
) -> Result<(char, String, WebSocketStream<MaybeTlsStream<tokio::net::TcpStream>>), String> {
|
||||
let letter = (b'A' + idx as u8) as char;
|
||||
|
||||
let mut request = url
|
||||
.into_client_request()
|
||||
.map_err(|e| format!("[{}] Invalid URL {}: {}", letter, url, e))?;
|
||||
|
||||
if let Some(token) = bearer_token {
|
||||
let auth_value = HeaderValue::from_str(&format!("Bearer {}", token))
|
||||
.map_err(|e| format!("Invalid bearer token: {}", e))?;
|
||||
request.headers_mut().insert(AUTHORIZATION, auth_value);
|
||||
}
|
||||
|
||||
let user_agent = format!("{}/{}", env!("CARGO_PKG_NAME"), env!("CARGO_PKG_VERSION"));
|
||||
request
|
||||
.headers_mut()
|
||||
.insert(USER_AGENT, HeaderValue::from_str(&user_agent).unwrap());
|
||||
|
||||
info!("[{}] Connecting to {}", letter, url);
|
||||
|
||||
if debug_enabled {
|
||||
log_request(&request, letter);
|
||||
}
|
||||
|
||||
let (ws_stream, response) = match connect_async_with_config(request, None, false).await {
|
||||
Ok(result) => result,
|
||||
Err(e) => {
|
||||
if debug_enabled {
|
||||
if let tokio_tungstenite::tungstenite::Error::Http(ref response) = e {
|
||||
log_error_response(response, letter);
|
||||
}
|
||||
}
|
||||
return Err(format!("[{}] Failed to connect to {}: {}", letter, url, e));
|
||||
}
|
||||
};
|
||||
|
||||
info!(
|
||||
"[{}] Connected successfully (status: {})",
|
||||
letter,
|
||||
response.status()
|
||||
);
|
||||
|
||||
if debug_enabled {
|
||||
log_response(&response, letter);
|
||||
}
|
||||
|
||||
Ok((letter, url.to_string(), ws_stream))
|
||||
}
|
||||
|
||||
fn log_request(request: &tokio_tungstenite::tungstenite::http::Request<()>, letter: char) {
|
||||
debug!(
|
||||
"[{}] Request: {} {}",
|
||||
letter,
|
||||
request.method(),
|
||||
request.uri()
|
||||
);
|
||||
for (name, value) in request.headers() {
|
||||
debug!(
|
||||
"[{}] {}: {}",
|
||||
letter,
|
||||
name,
|
||||
value.to_str().unwrap_or("<binary>")
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
fn log_response(
|
||||
response: &tokio_tungstenite::tungstenite::http::Response<Option<Vec<u8>>>,
|
||||
letter: char,
|
||||
) {
|
||||
debug!(
|
||||
"[{}] Response: {} {}",
|
||||
letter,
|
||||
response.status().as_u16(),
|
||||
response.status().canonical_reason().unwrap_or("")
|
||||
);
|
||||
for (name, value) in response.headers() {
|
||||
debug!(
|
||||
"[{}] {}: {}",
|
||||
letter,
|
||||
name,
|
||||
value.to_str().unwrap_or("<binary>")
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
fn log_error_response(
|
||||
response: &tokio_tungstenite::tungstenite::http::Response<Option<Vec<u8>>>,
|
||||
letter: char,
|
||||
) {
|
||||
debug!(
|
||||
"[{}] Error response: {} {}",
|
||||
letter,
|
||||
response.status().as_u16(),
|
||||
response.status().canonical_reason().unwrap_or("")
|
||||
);
|
||||
for (name, value) in response.headers() {
|
||||
debug!(
|
||||
"[{}] {}: {}",
|
||||
letter,
|
||||
name,
|
||||
value.to_str().unwrap_or("<binary>")
|
||||
);
|
||||
}
|
||||
if let Some(body) = response.body() {
|
||||
if let Ok(body_str) = std::str::from_utf8(body) {
|
||||
debug!("[{}] Response body: {}", letter, body_str);
|
||||
}
|
||||
}
|
||||
}
|
||||
79
src/filter.rs
Normal file
79
src/filter.rs
Normal file
@@ -0,0 +1,79 @@
|
||||
use std::sync::Arc;
|
||||
|
||||
use jaq_interpret::{Ctx, Filter, FilterT, ParseCtx, RcIter, Val};
|
||||
use tracing::{info, warn};
|
||||
|
||||
pub type JaqFilters = Arc<Vec<Filter>>;
|
||||
|
||||
/// Compile a list of jaq expressions into filters.
|
||||
pub fn compile_filters(exprs: &[String]) -> Result<JaqFilters, String> {
|
||||
let mut filters = Vec::new();
|
||||
|
||||
for expr in exprs {
|
||||
let mut defs = ParseCtx::new(Vec::new());
|
||||
defs.insert_natives(jaq_core::core());
|
||||
defs.insert_defs(jaq_std::std());
|
||||
|
||||
let (parsed, errs) = jaq_parse::parse(expr, jaq_parse::main());
|
||||
if !errs.is_empty() {
|
||||
let err_msgs: Vec<String> = errs.iter().map(|e| format!("{:?}", e)).collect();
|
||||
return Err(format!(
|
||||
"Failed to parse jaq expression '{}': {}",
|
||||
expr,
|
||||
err_msgs.join(", ")
|
||||
));
|
||||
}
|
||||
|
||||
let parsed = parsed.ok_or_else(|| format!("Failed to parse jaq expression '{}'", expr))?;
|
||||
let filter = defs.compile(parsed);
|
||||
|
||||
if !defs.errs.is_empty() {
|
||||
return Err(format!(
|
||||
"Failed to compile jaq expression '{}' ({} error(s))",
|
||||
expr,
|
||||
defs.errs.len()
|
||||
));
|
||||
}
|
||||
|
||||
info!("Using jaq filter: {}", expr);
|
||||
filters.push(filter);
|
||||
}
|
||||
|
||||
Ok(Arc::new(filters))
|
||||
}
|
||||
|
||||
/// Evaluate all filters against a JSON value, returning the outputs for each filter.
|
||||
pub fn evaluate_filters(
|
||||
filters: &JaqFilters,
|
||||
json_val: &serde_json::Value,
|
||||
letter: char,
|
||||
seq_num: u64,
|
||||
) -> Vec<String> {
|
||||
let mut all_outputs = Vec::new();
|
||||
|
||||
for filter in filters.iter() {
|
||||
let inputs = RcIter::new(core::iter::empty());
|
||||
let ctx = Ctx::new([], &inputs);
|
||||
let out = filter.run((ctx, Val::from(json_val.clone())));
|
||||
|
||||
let mut filter_outputs = Vec::new();
|
||||
for result in out {
|
||||
match result {
|
||||
Ok(val) => {
|
||||
filter_outputs.push(val.to_string());
|
||||
}
|
||||
Err(e) => {
|
||||
warn!("[{}:{}] jaq error: {}", letter, seq_num, e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if filter_outputs.is_empty() {
|
||||
all_outputs.push("(no output)".to_string());
|
||||
} else {
|
||||
all_outputs.push(filter_outputs.join(", "));
|
||||
}
|
||||
}
|
||||
|
||||
all_outputs
|
||||
}
|
||||
125
src/handler.rs
Normal file
125
src/handler.rs
Normal file
@@ -0,0 +1,125 @@
|
||||
use std::fs;
|
||||
use std::path::PathBuf;
|
||||
|
||||
use futures_util::StreamExt;
|
||||
use tokio::net::TcpStream;
|
||||
use tokio_tungstenite::{tungstenite::Message, MaybeTlsStream, WebSocketStream};
|
||||
use tracing::{error, info, warn};
|
||||
|
||||
use crate::filter::{evaluate_filters, JaqFilters};
|
||||
|
||||
/// Configuration for the connection handler.
|
||||
#[derive(Clone)]
|
||||
pub struct HandlerConfig {
|
||||
pub session_dir: PathBuf,
|
||||
pub jaq_filters: JaqFilters,
|
||||
pub json_normalize: bool,
|
||||
}
|
||||
|
||||
/// Handle a WebSocket connection, processing messages and writing them to files.
|
||||
pub async fn handle_connection(
|
||||
letter: char,
|
||||
url: String,
|
||||
ws_stream: WebSocketStream<MaybeTlsStream<TcpStream>>,
|
||||
config: HandlerConfig,
|
||||
) -> (char, String) {
|
||||
let (_, mut read) = ws_stream.split();
|
||||
let mut seq_num: u64 = 0;
|
||||
|
||||
while let Some(message_result) = read.next().await {
|
||||
match message_result {
|
||||
Ok(message) => {
|
||||
match message {
|
||||
Message::Text(text) => {
|
||||
handle_text_message(&text, letter, seq_num, &config);
|
||||
}
|
||||
Message::Binary(data) => {
|
||||
handle_binary_message(&data, letter, seq_num, &config);
|
||||
}
|
||||
Message::Ping(data) => {
|
||||
info!("[{}] Ping: {} bytes", letter, data.len());
|
||||
continue;
|
||||
}
|
||||
Message::Pong(data) => {
|
||||
info!("[{}] Pong: {} bytes", letter, data.len());
|
||||
continue;
|
||||
}
|
||||
Message::Close(frame) => {
|
||||
if let Some(cf) = frame {
|
||||
info!("[{}] Connection closed: {} - {}", letter, cf.code, cf.reason);
|
||||
} else {
|
||||
info!("[{}] Connection closed", letter);
|
||||
}
|
||||
break;
|
||||
}
|
||||
Message::Frame(_) => {
|
||||
continue;
|
||||
}
|
||||
}
|
||||
seq_num += 1;
|
||||
}
|
||||
Err(e) => {
|
||||
warn!("[{}] Error receiving message: {}", letter, e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
info!("[{}] Session ended. Received {} messages.", letter, seq_num);
|
||||
(letter, url)
|
||||
}
|
||||
|
||||
fn handle_text_message(text: &str, letter: char, seq_num: u64, config: &HandlerConfig) {
|
||||
// Log message based on jaq filters
|
||||
if config.jaq_filters.is_empty() {
|
||||
let preview: String = text.chars().take(50).collect();
|
||||
let truncated = if text.len() > 50 { "..." } else { "" };
|
||||
info!("[{}:{}] Text: {}{}", letter, seq_num, preview, truncated);
|
||||
} else {
|
||||
match serde_json::from_str::<serde_json::Value>(text) {
|
||||
Ok(json_val) => {
|
||||
let outputs = evaluate_filters(&config.jaq_filters, &json_val, letter, seq_num);
|
||||
info!("[{}:{}] {}", letter, seq_num, outputs.join(" | "));
|
||||
}
|
||||
Err(e) => {
|
||||
warn!("[{}:{}] JSON parse error: {}", letter, seq_num, e);
|
||||
let preview: String = text.chars().take(50).collect();
|
||||
let truncated = if text.len() > 50 { "..." } else { "" };
|
||||
info!("[{}:{}] Text: {}{}", letter, seq_num, preview, truncated);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Write message to file
|
||||
let (filename, content) = if config.json_normalize {
|
||||
if let Ok(json_val) = serde_json::from_str::<serde_json::Value>(text) {
|
||||
let pretty = serde_json::to_string_pretty(&json_val).unwrap_or_else(|_| text.to_string());
|
||||
(
|
||||
config.session_dir.join(format!("{}{}.json", letter, seq_num)),
|
||||
pretty,
|
||||
)
|
||||
} else {
|
||||
(
|
||||
config.session_dir.join(format!("{}{}.txt", letter, seq_num)),
|
||||
text.to_string(),
|
||||
)
|
||||
}
|
||||
} else {
|
||||
(
|
||||
config.session_dir.join(format!("{}{}.txt", letter, seq_num)),
|
||||
text.to_string(),
|
||||
)
|
||||
};
|
||||
|
||||
if let Err(e) = fs::write(&filename, content) {
|
||||
error!("[{}] Failed to write {:?}: {}", letter, filename, e);
|
||||
}
|
||||
}
|
||||
|
||||
fn handle_binary_message(data: &[u8], letter: char, seq_num: u64, config: &HandlerConfig) {
|
||||
info!("[{}:{}] Binary: {} bytes", letter, seq_num, data.len());
|
||||
|
||||
let filename = config.session_dir.join(format!("{}{}.bin", letter, seq_num));
|
||||
if let Err(e) = fs::write(&filename, data) {
|
||||
error!("[{}] Failed to write {:?}: {}", letter, filename, e);
|
||||
}
|
||||
}
|
||||
250
src/main.rs
250
src/main.rs
@@ -1,51 +1,27 @@
|
||||
mod cli;
|
||||
mod connection;
|
||||
mod filter;
|
||||
mod handler;
|
||||
|
||||
use std::fs;
|
||||
use std::path::PathBuf;
|
||||
|
||||
use chrono::Local;
|
||||
use clap::Parser;
|
||||
use url::{form_urlencoded, Url};
|
||||
use futures_util::StreamExt;
|
||||
use tokio::task::JoinSet;
|
||||
use tokio_tungstenite::{
|
||||
connect_async_with_config,
|
||||
tungstenite::{
|
||||
client::IntoClientRequest,
|
||||
http::header::{HeaderValue, AUTHORIZATION, USER_AGENT},
|
||||
Message,
|
||||
},
|
||||
};
|
||||
use tracing::{debug, error, info, warn, Level};
|
||||
use tracing::{error, info, Level};
|
||||
|
||||
#[derive(Parser)]
|
||||
#[command(name = "websocket-debug")]
|
||||
#[command(about = "A WebSocket debugging tool that logs and saves messages")]
|
||||
struct Args {
|
||||
/// WebSocket URLs to connect to (e.g., ws://localhost:8080 or wss://example.com/ws)
|
||||
#[arg(required = true)]
|
||||
urls: Vec<String>,
|
||||
|
||||
/// Bearer token for Authorization header
|
||||
#[arg(long)]
|
||||
bearer_token: Option<String>,
|
||||
|
||||
/// Enable debug logging (shows request/response headers)
|
||||
#[arg(long)]
|
||||
debug: bool,
|
||||
|
||||
/// Query string parameters to add to all URLs (pre-encoded, e.g., "name=First%20Last&key=value")
|
||||
#[arg(short = 'q', long = "query-string-all")]
|
||||
query_string_all: Option<String>,
|
||||
}
|
||||
use cli::Args;
|
||||
use connection::{connect, process_urls};
|
||||
use filter::compile_filters;
|
||||
use handler::{handle_connection, HandlerConfig};
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
let args = Args::parse();
|
||||
|
||||
let log_level = if args.debug {
|
||||
Level::DEBUG
|
||||
} else {
|
||||
Level::INFO
|
||||
};
|
||||
// Initialize logging
|
||||
let log_level = if args.debug { Level::DEBUG } else { Level::INFO };
|
||||
tracing_subscriber::fmt()
|
||||
.with_target(false)
|
||||
.with_thread_ids(false)
|
||||
@@ -58,140 +34,21 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
fs::create_dir_all(&session_dir)?;
|
||||
info!("Created session directory: {}", session_dir.display());
|
||||
|
||||
// Parse extra query params once if specified
|
||||
let extra_params: Vec<(String, String)> = args
|
||||
.query_string_all
|
||||
.as_ref()
|
||||
.map(|qs| {
|
||||
form_urlencoded::parse(qs.as_bytes())
|
||||
.map(|(k, v)| (k.into_owned(), v.into_owned()))
|
||||
.collect()
|
||||
})
|
||||
.unwrap_or_default();
|
||||
// Compile jaq filters
|
||||
let jaq_filters = compile_filters(&args.jaq)?;
|
||||
|
||||
// Process URLs and add query string parameters if specified
|
||||
let mut processed_urls = Vec::new();
|
||||
for url_str in &args.urls {
|
||||
let mut url = Url::parse(url_str)?;
|
||||
|
||||
if !extra_params.is_empty() {
|
||||
let existing: Vec<(String, String)> = url
|
||||
.query_pairs()
|
||||
.map(|(k, v)| (k.into_owned(), v.into_owned()))
|
||||
.collect();
|
||||
|
||||
let mut query_pairs = url.query_pairs_mut();
|
||||
query_pairs.clear();
|
||||
for (k, v) in &existing {
|
||||
query_pairs.append_pair(k, v);
|
||||
}
|
||||
for (k, v) in &extra_params {
|
||||
query_pairs.append_pair(k, v);
|
||||
}
|
||||
}
|
||||
|
||||
processed_urls.push(url.to_string());
|
||||
}
|
||||
// Process URLs with extra query parameters
|
||||
let processed_urls = process_urls(&args.urls, args.query_string_all.as_ref())?;
|
||||
|
||||
// Connect to all URLs simultaneously
|
||||
let mut connect_futures = Vec::new();
|
||||
for (idx, url) in processed_urls.iter().enumerate() {
|
||||
let letter = (b'A' + idx as u8) as char;
|
||||
let bearer_token = args.bearer_token.clone();
|
||||
let url = url.clone();
|
||||
let bearer_token = args.bearer_token.clone();
|
||||
let debug_enabled = args.debug;
|
||||
|
||||
connect_futures.push(async move {
|
||||
let mut request = url
|
||||
.as_str()
|
||||
.into_client_request()
|
||||
.map_err(|e| format!("[{}] Invalid URL {}: {}", letter, url, e))?;
|
||||
|
||||
if let Some(ref token) = bearer_token {
|
||||
let auth_value = HeaderValue::from_str(&format!("Bearer {}", token))
|
||||
.map_err(|e| format!("Invalid bearer token: {}", e))?;
|
||||
request.headers_mut().insert(AUTHORIZATION, auth_value);
|
||||
}
|
||||
|
||||
let user_agent = format!("{}/{}", env!("CARGO_PKG_NAME"), env!("CARGO_PKG_VERSION"));
|
||||
request
|
||||
.headers_mut()
|
||||
.insert(USER_AGENT, HeaderValue::from_str(&user_agent).unwrap());
|
||||
|
||||
info!("[{}] Connecting to {}", letter, url);
|
||||
|
||||
if debug_enabled {
|
||||
debug!(
|
||||
"[{}] Request: {} {}",
|
||||
letter,
|
||||
request.method(),
|
||||
request.uri()
|
||||
);
|
||||
for (name, value) in request.headers() {
|
||||
debug!(
|
||||
"[{}] {}: {}",
|
||||
letter,
|
||||
name,
|
||||
value.to_str().unwrap_or("<binary>")
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
let (ws_stream, response) = match connect_async_with_config(request, None, false).await
|
||||
{
|
||||
Ok(result) => result,
|
||||
Err(e) => {
|
||||
if debug_enabled {
|
||||
if let tokio_tungstenite::tungstenite::Error::Http(ref response) = e {
|
||||
debug!(
|
||||
"[{}] Error response: {} {}",
|
||||
letter,
|
||||
response.status().as_u16(),
|
||||
response.status().canonical_reason().unwrap_or("")
|
||||
);
|
||||
for (name, value) in response.headers() {
|
||||
debug!(
|
||||
"[{}] {}: {}",
|
||||
letter,
|
||||
name,
|
||||
value.to_str().unwrap_or("<binary>")
|
||||
);
|
||||
}
|
||||
if let Some(body) = response.body() {
|
||||
if let Ok(body_str) = std::str::from_utf8(body) {
|
||||
debug!("[{}] Response body: {}", letter, body_str);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return Err(format!("[{}] Failed to connect to {}: {}", letter, url, e));
|
||||
}
|
||||
};
|
||||
|
||||
info!(
|
||||
"[{}] Connected successfully (status: {})",
|
||||
letter,
|
||||
response.status()
|
||||
);
|
||||
|
||||
if debug_enabled {
|
||||
debug!(
|
||||
"[{}] Response: {} {}",
|
||||
letter,
|
||||
response.status().as_u16(),
|
||||
response.status().canonical_reason().unwrap_or("")
|
||||
);
|
||||
for (name, value) in response.headers() {
|
||||
debug!(
|
||||
"[{}] {}: {}",
|
||||
letter,
|
||||
name,
|
||||
value.to_str().unwrap_or("<binary>")
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
Ok::<_, String>((letter, url, ws_stream))
|
||||
connect(&url, idx, bearer_token.as_deref(), debug_enabled).await
|
||||
});
|
||||
}
|
||||
|
||||
@@ -211,74 +68,17 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
|
||||
info!("All {} connections established", connections.len());
|
||||
|
||||
// Spawn tasks for each connection
|
||||
// Spawn handler tasks for each connection
|
||||
let mut join_set: JoinSet<(char, String)> = JoinSet::new();
|
||||
|
||||
for (letter, url, ws_stream) in connections {
|
||||
let session_dir = session_dir.clone();
|
||||
let config = HandlerConfig {
|
||||
session_dir: session_dir.clone(),
|
||||
jaq_filters: jaq_filters.clone(),
|
||||
json_normalize: args.json_normalize,
|
||||
};
|
||||
|
||||
join_set.spawn(async move {
|
||||
let (_, mut read) = ws_stream.split();
|
||||
let mut seq_num: u64 = 0;
|
||||
|
||||
while let Some(message_result) = read.next().await {
|
||||
match message_result {
|
||||
Ok(message) => {
|
||||
match message {
|
||||
Message::Text(text) => {
|
||||
let preview: String = text.chars().take(50).collect();
|
||||
let truncated = if text.len() > 50 { "..." } else { "" };
|
||||
info!("[{}:{}] Text: {}{}", letter, seq_num, preview, truncated);
|
||||
|
||||
let filename =
|
||||
session_dir.join(format!("{}{}.txt", letter, seq_num));
|
||||
if let Err(e) = fs::write(&filename, &text) {
|
||||
error!("[{}] Failed to write {:?}: {}", letter, filename, e);
|
||||
}
|
||||
}
|
||||
Message::Binary(data) => {
|
||||
info!("[{}:{}] Binary: {} bytes", letter, seq_num, data.len());
|
||||
|
||||
let filename =
|
||||
session_dir.join(format!("{}{}.bin", letter, seq_num));
|
||||
if let Err(e) = fs::write(&filename, &data) {
|
||||
error!("[{}] Failed to write {:?}: {}", letter, filename, e);
|
||||
}
|
||||
}
|
||||
Message::Ping(data) => {
|
||||
info!("[{}] Ping: {} bytes", letter, data.len());
|
||||
continue;
|
||||
}
|
||||
Message::Pong(data) => {
|
||||
info!("[{}] Pong: {} bytes", letter, data.len());
|
||||
continue;
|
||||
}
|
||||
Message::Close(frame) => {
|
||||
if let Some(cf) = frame {
|
||||
info!(
|
||||
"[{}] Connection closed: {} - {}",
|
||||
letter, cf.code, cf.reason
|
||||
);
|
||||
} else {
|
||||
info!("[{}] Connection closed", letter);
|
||||
}
|
||||
break;
|
||||
}
|
||||
Message::Frame(_) => {
|
||||
continue;
|
||||
}
|
||||
}
|
||||
seq_num += 1;
|
||||
}
|
||||
Err(e) => {
|
||||
warn!("[{}] Error receiving message: {}", letter, e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
info!("[{}] Session ended. Received {} messages.", letter, seq_num);
|
||||
(letter, url)
|
||||
});
|
||||
join_set.spawn(handle_connection(letter, url, ws_stream, config));
|
||||
}
|
||||
|
||||
// Wait for all connections to finish
|
||||
|
||||
Reference in New Issue
Block a user