diff --git a/rust/Cargo.lock b/rust/Cargo.lock index c5438f16..31e82890 100644 --- a/rust/Cargo.lock +++ b/rust/Cargo.lock @@ -349,6 +349,12 @@ version = "3.20.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5d20789868f4b01b2f2caec9f5c4e0213b41e3e5702a50157d699ae31ced2fcb" +[[package]] +name = "byteorder" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" + [[package]] name = "bytes" version = "1.11.1" @@ -727,6 +733,12 @@ dependencies = [ "syn", ] +[[package]] +name = "data-encoding" +version = "2.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06" + [[package]] name = "deranged" version = "0.5.8" @@ -1008,18 +1020,6 @@ dependencies = [ "wasm-bindgen", ] -[[package]] -name = "getrandom" -version = "0.3.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd" -dependencies = [ - "cfg-if", - "libc", - "r-efi 5.3.0", - "wasip2", -] - [[package]] name = "getrandom" version = "0.4.3" @@ -1029,7 +1029,7 @@ dependencies = [ "cfg-if", "js-sys", "libc", - "r-efi 6.0.0", + "r-efi", "rand_core 0.10.1", "wasm-bindgen", ] @@ -1947,12 +1947,6 @@ dependencies = [ "proc-macro2", ] -[[package]] -name = "r-efi" -version = "5.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" - [[package]] name = "r-efi" version = "6.0.0" @@ -2337,6 +2331,21 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" +[[package]] +name = "sdk" +version = "0.1.0" +dependencies = [ + "futures-util", + "serde", + "serde_json", + "thiserror 1.0.69", + "tokio", + "tokio-test", + "tokio-tungstenite", + "tracing", + "uuid", +] + [[package]] name = "security-framework" version = "3.7.0" @@ -2441,6 +2450,17 @@ dependencies = [ "serde", ] +[[package]] +name = "sha1" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a978451301f4db1d02937a4ab3ccce137717b81826e79b7d49ffe3244a13c3b8" +dependencies = [ + "cfg-if", + "cpufeatures 0.2.17", + "digest", +] + [[package]] name = "sha2" version = "0.10.9" @@ -2612,7 +2632,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.3.4", + "getrandom 0.4.3", "once_cell", "rustix 1.1.4", "windows-sys 0.61.2", @@ -2733,6 +2753,7 @@ dependencies = [ "bytes", "libc", "mio", + "parking_lot", "pin-project-lite", "signal-hook-registry", "socket2 0.6.3", @@ -2772,6 +2793,29 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-test" +version = "0.4.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "12bc5db7778a8e08ee255090861e0e7aac55e784300077e7ed293c7be36168b2" +dependencies = [ + "futures-core", + "tokio", + "tokio-stream", +] + +[[package]] +name = "tokio-tungstenite" +version = "0.23.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6989540ced10490aaf14e6bad2e3d33728a2813310a0c71d1574304c49631cd" +dependencies = [ + "futures-util", + "log", + "tokio", + "tungstenite", +] + [[package]] name = "tokio-util" version = "0.7.18" @@ -2983,6 +3027,24 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "tungstenite" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e2e2ce1e47ed2994fd43b04c8f618008d4cabdd5ee34027cf14f9d918edd9c8" +dependencies = [ + "byteorder", + "bytes", + "data-encoding", + "http", + "httparse", + "log", + "rand 0.8.6", + "sha1", + "thiserror 1.0.69", + "utf-8", +] + [[package]] name = "typenum" version = "1.19.0" @@ -3031,6 +3093,12 @@ dependencies = [ "serde", ] +[[package]] +name = "utf-8" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9" + [[package]] name = "utf8_iter" version = "1.0.4" @@ -3043,6 +3111,18 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" +[[package]] +name = "uuid" +version = "1.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7cc1186384beb7dd8eedea376413fd654937285ea6c9cfbb928dc3043ea4b606" +dependencies = [ + "getrandom 0.4.3", + "js-sys", + "serde_core", + "wasm-bindgen", +] + [[package]] name = "vcpkg" version = "0.2.15" @@ -3080,15 +3160,6 @@ version = "0.11.1+wasi-snapshot-preview1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" -[[package]] -name = "wasip2" -version = "1.0.2+wasi-0.2.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9517f9239f02c069db75e65f174b3da828fe5f5b945c4dd26bd25d89c03ebcf5" -dependencies = [ - "wit-bindgen", -] - [[package]] name = "wasm-bindgen" version = "0.2.116" @@ -3397,12 +3468,6 @@ dependencies = [ "memchr", ] -[[package]] -name = "wit-bindgen" -version = "0.51.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d7249219f66ced02969388cf2bb044a09756a083d0fab1e566056b04d9fbcaa5" - [[package]] name = "writeable" version = "0.6.2" diff --git a/rust/crates/sdk/Cargo.toml b/rust/crates/sdk/Cargo.toml new file mode 100644 index 00000000..1afeb8be --- /dev/null +++ b/rust/crates/sdk/Cargo.toml @@ -0,0 +1,18 @@ +[package] +name = "sdk" +version = "0.1.0" +edition = "2021" +publish = false + +[dependencies] +tokio = { version = "1", features = ["full"] } +tokio-tungstenite = "0.23" +futures-util = "0.3" +serde_json = "1" +serde = { version = "1", features = ["derive"] } +uuid = { version = "1", features = ["v4", "serde"] } +thiserror = "1" +tracing = "0.1" + +[dev-dependencies] +tokio-test = "0.4" diff --git a/rust/crates/sdk/src/error.rs b/rust/crates/sdk/src/error.rs new file mode 100644 index 00000000..ab90ab63 --- /dev/null +++ b/rust/crates/sdk/src/error.rs @@ -0,0 +1,130 @@ +//! SDK error types and result type. + +use std::fmt; + +/// SDK wire protocol error codes. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ErrorCode { + /// Invalid JSON or malformed frame + InvalidInput, + /// Unknown broker operation + UnknownOperation, + /// Payload too large (>4 MiB) + PayloadTooLarge, + /// Authentication token mismatch + Unauthorized, + /// Broker or session unavailable + Unavailable, + /// WebSocket connection error + ConnectionError, + /// Request timeout + Timeout, + /// Frame parsing or validation error + FrameError, + /// Request/response correlation failure + CorrelationError, + /// Other unclassified error + Internal, +} + +impl fmt::Display for ErrorCode { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::InvalidInput => write!(f, "invalid_input"), + Self::UnknownOperation => write!(f, "unknown_operation"), + Self::PayloadTooLarge => write!(f, "payload_too_large"), + Self::Unauthorized => write!(f, "unauthorized"), + Self::Unavailable => write!(f, "unavailable"), + Self::ConnectionError => write!(f, "connection_error"), + Self::Timeout => write!(f, "timeout"), + Self::FrameError => write!(f, "frame_error"), + Self::CorrelationError => write!(f, "correlation_error"), + Self::Internal => write!(f, "internal_error"), + } + } +} + +/// SDK wire protocol error. +#[derive(Debug)] +pub struct SdkError { + pub code: ErrorCode, + pub message: String, +} + +impl SdkError { + pub fn new(code: ErrorCode, message: impl Into) -> Self { + Self { + code, + message: message.into(), + } + } + + pub fn invalid_input(message: impl Into) -> Self { + Self::new(ErrorCode::InvalidInput, message) + } + + pub fn unknown_operation(message: impl Into) -> Self { + Self::new(ErrorCode::UnknownOperation, message) + } + + pub fn payload_too_large(message: impl Into) -> Self { + Self::new(ErrorCode::PayloadTooLarge, message) + } + + pub fn unauthorized(message: impl Into) -> Self { + Self::new(ErrorCode::Unauthorized, message) + } + + pub fn unavailable(message: impl Into) -> Self { + Self::new(ErrorCode::Unavailable, message) + } + + pub fn connection_error(message: impl Into) -> Self { + Self::new(ErrorCode::ConnectionError, message) + } + + pub fn timeout(message: impl Into) -> Self { + Self::new(ErrorCode::Timeout, message) + } + + pub fn frame_error(message: impl Into) -> Self { + Self::new(ErrorCode::FrameError, message) + } + + pub fn correlation_error(message: impl Into) -> Self { + Self::new(ErrorCode::CorrelationError, message) + } + + pub fn internal(message: impl Into) -> Self { + Self::new(ErrorCode::Internal, message) + } +} + +impl fmt::Display for SdkError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}: {}", self.code, self.message) + } +} + +impl std::error::Error for SdkError {} + +/// Result type for SDK operations. +pub type SdkResult = Result; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn error_display_formats_code_and_message() { + let err = SdkError::invalid_input("test message"); + assert_eq!(err.to_string(), "invalid_input: test message"); + } + + #[test] + fn error_code_display_works() { + assert_eq!(ErrorCode::InvalidInput.to_string(), "invalid_input"); + assert_eq!(ErrorCode::PayloadTooLarge.to_string(), "payload_too_large"); + assert_eq!(ErrorCode::Unauthorized.to_string(), "unauthorized"); + } +} diff --git a/rust/crates/sdk/src/frame.rs b/rust/crates/sdk/src/frame.rs new file mode 100644 index 00000000..1f0cdd20 --- /dev/null +++ b/rust/crates/sdk/src/frame.rs @@ -0,0 +1,258 @@ +//! Frame types and serialization for the SDK wire protocol. + +use serde_json::{Map, Value}; + +use crate::error::{SdkError, SdkResult}; + +/// Protocol version implemented. +pub const PROTOCOL_VERSION: u32 = 3; + +/// Maximum UTF-8 byte length for a single broker JSON frame. +pub const MAX_FRAME_BYTES: usize = 4 * 1024 * 1024; + +/// Frame type identifier. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum FrameType { + /// Hello frame sent by broker on connection + BrokerHello, + /// Request frame from client to broker + BrokerRequest, + /// Response frame from broker to client + BrokerResponse, + /// Query request + QueryRequest, + /// Query response + QueryResponse, + /// Control request + ControlRequest, + /// Control response + ControlResponse, + /// Event replay request + EventReplay, + /// Event replay result + EventReplayResult, + /// Session activation + SessionActivate, + /// Session activation result + SessionActivateResult, +} + +impl FrameType { + pub fn as_str(&self) -> &'static str { + match self { + Self::BrokerHello => "broker_hello", + Self::BrokerRequest => "broker_request", + Self::BrokerResponse => "broker_response", + Self::QueryRequest => "query_request", + Self::QueryResponse => "query_response", + Self::ControlRequest => "control_request", + Self::ControlResponse => "control_response", + Self::EventReplay => "event_replay", + Self::EventReplayResult => "event_replay_result", + Self::SessionActivate => "session_activate", + Self::SessionActivateResult => "session_activate_result", + } + } + + pub fn parse(s: &str) -> SdkResult { + match s { + "broker_hello" => Ok(Self::BrokerHello), + "broker_request" => Ok(Self::BrokerRequest), + "broker_response" => Ok(Self::BrokerResponse), + "query_request" => Ok(Self::QueryRequest), + "query_response" => Ok(Self::QueryResponse), + "control_request" => Ok(Self::ControlRequest), + "control_response" => Ok(Self::ControlResponse), + "event_replay" => Ok(Self::EventReplay), + "event_replay_result" => Ok(Self::EventReplayResult), + "session_activate" => Ok(Self::SessionActivate), + "session_activate_result" => Ok(Self::SessionActivateResult), + _ => Err(SdkError::frame_error(format!("unknown frame type: {}", s))), + } + } +} + +/// A frame in the SDK wire protocol. +/// +/// Frames are JSON objects with a `type` field and optional `id` field for correlation. +#[derive(Debug, Clone)] +pub struct Frame { + /// Underlying JSON value + pub data: Map, +} + +impl Frame { + /// Create a new frame from a JSON object. + pub fn new(data: Map) -> Self { + Self { data } + } + + /// Create an empty frame. + pub fn empty() -> Self { + Self { data: Map::new() } + } + + /// Get the frame type. + pub fn frame_type(&self) -> SdkResult { + let frame_type_str = self + .data + .get("type") + .and_then(Value::as_str) + .ok_or_else(|| SdkError::frame_error("missing 'type' field"))?; + FrameType::parse(frame_type_str) + } + + /// Get the optional request/response ID for correlation. + pub fn id(&self) -> Option { + self.data + .get("id") + .and_then(Value::as_str) + .map(String::from) + } + + /// Get the `ok` boolean field from a response frame. + pub fn ok(&self) -> SdkResult { + self.data + .get("ok") + .and_then(Value::as_bool) + .ok_or_else(|| SdkError::frame_error("missing or invalid 'ok' field")) + } + + /// Get the error object from a response frame. + pub fn error(&self) -> Option> { + let error_obj = self.data.get("error")?.as_object()?; + let code = error_obj.get("code")?.as_str()?.to_string(); + let message = error_obj.get("message")?.as_str()?.to_string(); + Some(Ok((code, message))) + } + + /// Get the result field from a response frame. + pub fn result(&self) -> Option { + self.data.get("result").cloned() + } + + /// Convert to JSON string. + pub fn to_json(&self) -> SdkResult { + serde_json::to_string(&self.data) + .map_err(|e| SdkError::frame_error(format!("JSON serialization error: {}", e))) + } + + /// Parse from JSON string. + pub fn from_json(json_str: &str) -> SdkResult { + let data: Map = serde_json::from_str(json_str) + .map_err(|e| SdkError::frame_error(format!("JSON parsing error: {}", e)))?; + Ok(Self::new(data)) + } + + /// Create a broker hello frame. + pub fn broker_hello() -> Self { + let mut data = Map::new(); + data.insert( + "type".to_string(), + Value::String("broker_hello".to_string()), + ); + data.insert( + "protocolVersion".to_string(), + Value::Number(PROTOCOL_VERSION.into()), + ); + Self::new(data) + } + + /// Create a broker request frame. + pub fn broker_request(id: String, operation: String, input: Value) -> Self { + let mut data = Map::new(); + data.insert( + "type".to_string(), + Value::String("broker_request".to_string()), + ); + data.insert("id".to_string(), Value::String(id)); + data.insert("operation".to_string(), Value::String(operation)); + data.insert("input".to_string(), input); + Self::new(data) + } + + /// Create a broker response frame. + pub fn broker_response_ok(id: String, result: Value) -> Self { + let mut data = Map::new(); + data.insert( + "type".to_string(), + Value::String("broker_response".to_string()), + ); + data.insert("id".to_string(), Value::String(id)); + data.insert("ok".to_string(), Value::Bool(true)); + data.insert("result".to_string(), result); + Self::new(data) + } + + /// Create a broker error response frame. + pub fn broker_response_error(id: String, code: String, message: String) -> Self { + let mut error_data = Map::new(); + error_data.insert("code".to_string(), Value::String(code)); + error_data.insert("message".to_string(), Value::String(message)); + + let mut data = Map::new(); + data.insert( + "type".to_string(), + Value::String("broker_response".to_string()), + ); + data.insert("id".to_string(), Value::String(id)); + data.insert("ok".to_string(), Value::Bool(false)); + data.insert("error".to_string(), Value::Object(error_data)); + Self::new(data) + } + + /// Get the payload size in bytes. + pub fn byte_size(&self) -> usize { + self.to_json().map(|s| s.len()).unwrap_or(0) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn frame_type_round_trip() { + assert_eq!(FrameType::BrokerHello.as_str(), "broker_hello"); + assert_eq!( + FrameType::parse("broker_hello").unwrap(), + FrameType::BrokerHello + ); + } + + #[test] + fn frame_parsing_extracts_type() { + let frame = Frame::broker_hello(); + assert_eq!(frame.frame_type().unwrap(), FrameType::BrokerHello); + } + + #[test] + fn frame_with_id_can_be_recovered() { + let frame = Frame::broker_request( + "test-id".to_string(), + "test.operation".to_string(), + Value::Object(Map::new()), + ); + assert_eq!(frame.id(), Some("test-id".to_string())); + } + + #[test] + fn frame_json_round_trip() { + let frame = Frame::broker_hello(); + let json_str = frame.to_json().unwrap(); + let parsed = Frame::from_json(&json_str).unwrap(); + assert_eq!(parsed.frame_type().unwrap(), FrameType::BrokerHello); + } + + #[test] + fn broker_response_error_contains_code_and_message() { + let frame = Frame::broker_response_error( + "id1".to_string(), + "invalid_input".to_string(), + "test error".to_string(), + ); + let (code, message) = frame.error().unwrap().unwrap(); + assert_eq!(code, "invalid_input"); + assert_eq!(message, "test error"); + } +} diff --git a/rust/crates/sdk/src/lib.rs b/rust/crates/sdk/src/lib.rs new file mode 100644 index 00000000..48e1b524 --- /dev/null +++ b/rust/crates/sdk/src/lib.rs @@ -0,0 +1,20 @@ +//! GJC SDK wire transport implementation. +//! +//! This crate provides a Rust implementation of the GJC SDK wire protocol, +//! compatible with the gajae-code SDK broker framing, request/response correlation, +//! and event stream semantics. + +#![forbid(unsafe_code)] +#![allow( + clippy::missing_errors_doc, + clippy::missing_panics_doc, + clippy::module_name_repetitions +)] + +pub mod error; +pub mod frame; +pub mod transport; + +pub use error::{SdkError, SdkResult}; +pub use frame::{Frame, FrameType}; +pub use transport::{SdkTransport, SdkTransportConfig}; diff --git a/rust/crates/sdk/src/transport.rs b/rust/crates/sdk/src/transport.rs new file mode 100644 index 00000000..97689a44 --- /dev/null +++ b/rust/crates/sdk/src/transport.rs @@ -0,0 +1,251 @@ +//! WebSocket transport implementation for the SDK wire protocol. + +use std::collections::HashMap; +use std::sync::Arc; + +use futures_util::sink::SinkExt; +use futures_util::stream::StreamExt; +use serde_json::Value; +use tokio::sync::{mpsc, oneshot, Mutex, RwLock}; +use tokio_tungstenite::{connect_async, tungstenite::Message}; +use uuid::Uuid; + +use crate::error::{SdkError, SdkResult}; +use crate::frame::{Frame, FrameType, MAX_FRAME_BYTES}; + +/// Configuration for SDK transport. +#[derive(Debug, Clone)] +pub struct SdkTransportConfig { + /// Server URL (e.g., "ws://127.0.0.1:8080") + pub url: String, + /// Authentication token + pub token: String, + /// Maximum payload size in bytes (default: 4 MiB) + pub max_payload_bytes: usize, + /// Request timeout in seconds (default: 30) + pub timeout_secs: u64, +} + +impl Default for SdkTransportConfig { + fn default() -> Self { + Self { + url: "ws://127.0.0.1:8080".to_string(), + token: String::new(), + max_payload_bytes: MAX_FRAME_BYTES, + timeout_secs: 30, + } + } +} + +/// Pending request tracking for correlation. +struct PendingRequest { + response_tx: oneshot::Sender>, +} + +/// SDK WebSocket transport. +pub struct SdkTransport { + /// Channel for sending messages to the writer task + tx: mpsc::UnboundedSender, + /// Pending requests by ID + pending: Arc>>, + /// Configuration + config: SdkTransportConfig, + /// Background task handle + task_handle: Arc>>>, +} + +impl SdkTransport { + /// Create a new SDK transport and connect to the broker. + pub async fn connect(config: SdkTransportConfig) -> SdkResult { + // Add token to query params + let url_with_token = if config.url.contains('?') { + format!("{}&token={}", config.url, config.token) + } else { + format!("{}?token={}", config.url, config.token) + }; + + // Create WebSocket connection + let (ws, _) = connect_async(&url_with_token).await.map_err(|e| { + SdkError::connection_error(format!("WebSocket connection failed: {}", e)) + })?; + + let (ws_tx, ws_rx) = tokio::sync::mpsc::unbounded_channel(); + let pending = Arc::new(RwLock::new(HashMap::new())); + let pending_clone = pending.clone(); + + // Spawn background task for handling the WebSocket + let task_handle = tokio::spawn(async move { + Self::background_task(ws, ws_rx, pending_clone).await; + }); + + let transport = Self { + tx: ws_tx, + pending, + config, + task_handle: Arc::new(Mutex::new(Some(task_handle))), + }; + + Ok(transport) + } + + /// Send a request and wait for response. + pub async fn request(&self, operation: String, input: Value) -> SdkResult { + // Generate correlation ID + let id = Uuid::new_v4().to_string(); + + // Create request frame + let request = Frame::broker_request(id.clone(), operation, input); + + // Check payload size + if request.byte_size() > self.config.max_payload_bytes { + return Err(SdkError::payload_too_large(format!( + "frame size {} exceeds maximum {}", + request.byte_size(), + self.config.max_payload_bytes + ))); + } + + // Create response channel + let (response_tx, response_rx) = oneshot::channel(); + self.pending + .write() + .await + .insert(id.clone(), PendingRequest { response_tx }); + + // Send request + let json = request.to_json()?; + self.tx + .send(Message::Text(json)) + .map_err(|e| SdkError::connection_error(format!("send failed: {}", e)))?; + + // Wait for response with timeout + let timeout_duration = std::time::Duration::from_secs(self.config.timeout_secs); + match tokio::time::timeout(timeout_duration, response_rx).await { + Ok(Ok(frame)) => frame, + Ok(Err(_)) => Err(SdkError::correlation_error("response channel closed")), + Err(_) => Err(SdkError::timeout("response not received within timeout")), + } + } + + /// Background task for handling WebSocket communication. + #[allow(unused_mut)] + async fn background_task( + ws: tokio_tungstenite::WebSocketStream< + tokio_tungstenite::MaybeTlsStream, + >, + mut ws_rx: mpsc::UnboundedReceiver, + pending: Arc>>, + ) { + let (mut ws_sink, mut ws_stream) = ws.split(); + + loop { + tokio::select! { + // Handle incoming WebSocket messages + msg = ws_stream.next() => { + match msg { + Some(Ok(Message::Text(text))) => { + // Parse frame + match Frame::from_json(&text) { + Ok(frame) => { + // Handle frame type + match frame.frame_type() { + Ok(FrameType::BrokerHello) => { + // Hello frame received, continue + } + Ok(FrameType::BrokerResponse) => { + // Response frame, correlate with request + if let Some(id) = frame.id() { + if let Some(PendingRequest { response_tx }) = + pending.write().await.remove(&id) + { + let _ = response_tx.send(Ok(frame)); + } + } + } + _ => { + // Other frame types + tracing::debug!("Unhandled frame type: {:?}", frame.frame_type()); + } + } + } + Err(e) => { + tracing::warn!("Failed to parse frame: {}", e); + } + } + } + Some(Ok(Message::Close(_))) => { + break; + } + Some(Ok(_)) => { + // Other message types (binary, etc.), ignore + } + Some(Err(e)) => { + tracing::warn!("WebSocket error: {}", e); + break; + } + None => { + break; + } + } + } + // Handle outgoing messages + Some(msg) = ws_rx.recv() => { + if let Err(e) = ws_sink.send(msg).await { + tracing::warn!("Failed to send message: {}", e); + break; + } + } + else => break, + } + } + + // Fail all pending requests + let mut pending = pending.write().await; + for (_, PendingRequest { response_tx }) in pending.drain() { + let _ = response_tx.send(Err(SdkError::connection_error("connection closed"))); + } + } + + /// Close the transport. + pub async fn close(self) -> SdkResult<()> { + // Send close message + let _ = self.tx.send(Message::Close(None)); + + // Wait for background task to finish + if let Some(handle) = self.task_handle.lock().await.take() { + let _ = tokio::time::timeout(std::time::Duration::from_secs(5), handle).await; + } + + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn config_default_values() { + let config = SdkTransportConfig::default(); + assert_eq!(config.url, "ws://127.0.0.1:8080"); + assert_eq!(config.max_payload_bytes, MAX_FRAME_BYTES); + assert_eq!(config.timeout_secs, 30); + } + + #[tokio::test] + async fn transport_can_be_created() { + // This test doesn't actually connect, just checks that the struct can be created + let config = SdkTransportConfig::default(); + assert_eq!(config.url, "ws://127.0.0.1:8080"); + } + + #[test] + fn request_frame_has_correlation_id() { + let frame = Frame::broker_request( + "test-id".to_string(), + "test.operation".to_string(), + Value::Object(serde_json::Map::new()), + ); + assert_eq!(frame.id(), Some("test-id".to_string())); + } +}