feat(sdk): complete WebSocket client transport

This commit is contained in:
clawdbot committed 2026-10-13 02:36:07 +00:00
1 parent a35caea528
commit 7c35206add
7 files changed
+1541 -388

No files matched your search

+18 -15
View File
@@ -1215,7 +1215,7 @@ dependencies = [
"tokio", "tokio",
"tokio-rustls", "tokio-rustls",
"tower-service", "tower-service",
"webpki-roots", "webpki-roots 1.0.6",
] ]
[[package]] [[package]]
@@ -2117,7 +2117,7 @@ dependencies = [
"wasm-bindgen-futures", "wasm-bindgen-futures",
"wasm-streams", "wasm-streams",
"web-sys", "web-sys",
"webpki-roots", "webpki-roots 1.0.6",
] ]
[[package]] [[package]]
@@ -2339,11 +2339,10 @@ dependencies = [
"futures-util", "futures-util",
"serde", "serde",
"serde_json", "serde_json",
"thiserror 1.0.69",
"tokio", "tokio",
"tokio-test",
"tokio-tungstenite", "tokio-tungstenite",
"tracing", "tracing",
"url",
"uuid", "uuid",
] ]
@@ -2794,17 +2793,6 @@ dependencies = [
"tokio", "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]] [[package]]
name = "tokio-tungstenite" name = "tokio-tungstenite"
version = "0.23.1" version = "0.23.1"
@@ -2813,8 +2801,12 @@ checksum = "c6989540ced10490aaf14e6bad2e3d33728a2813310a0c71d1574304c49631cd"
dependencies = [ dependencies = [
"futures-util", "futures-util",
"log", "log",
"rustls",
"rustls-pki-types",
"tokio", "tokio",
"tokio-rustls",
"tungstenite", "tungstenite",
"webpki-roots 0.26.11",
] ]
[[package]] [[package]]
@@ -3041,6 +3033,8 @@ dependencies = [
"httparse", "httparse",
"log", "log",
"rand 0.8.6", "rand 0.8.6",
"rustls",
"rustls-pki-types",
"sha1", "sha1",
"thiserror 1.0.69", "thiserror 1.0.69",
"utf-8", "utf-8",
@@ -3249,6 +3243,15 @@ dependencies = [
"wasm-bindgen", "wasm-bindgen",
] ]
[[package]]
name = "webpki-roots"
version = "0.26.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9"
dependencies = [
"webpki-roots 1.0.6",
]
[[package]] [[package]]
name = "webpki-roots" name = "webpki-roots"
version = "1.0.6" version = "1.0.6"
+2 -5
View File
@@ -6,13 +6,10 @@ publish = false
[dependencies] [dependencies]
tokio = { version = "1", features = ["full"] } tokio = { version = "1", features = ["full"] }
tokio-tungstenite = "0.23" tokio-tungstenite = { version = "0.23", features = ["rustls-tls-webpki-roots"] }
futures-util = "0.3" futures-util = "0.3"
serde_json = "1" serde_json = "1"
serde = { version = "1", features = ["derive"] } serde = { version = "1", features = ["derive"] }
uuid = { version = "1", features = ["v4", "serde"] } uuid = { version = "1", features = ["v4", "serde"] }
thiserror = "1"
tracing = "0.1" tracing = "0.1"
url = "2"
[dev-dependencies]
tokio-test = "0.4"
+128 -40
View File
@@ -1,102 +1,172 @@
//! SDK error types and result type. //! Typed SDK wire protocol errors and request recovery records.
use std::fmt; use std::fmt;
use std::time::SystemTime;
use uuid::Uuid;
/// SDK wire protocol error codes. /// SDK wire protocol error codes.
#[derive(Debug, Clone, Copy, PartialEq, Eq)] #[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ErrorCode { pub enum SdkErrorCode {
/// Invalid JSON or malformed frame
InvalidInput, InvalidInput,
/// Unknown broker operation
UnknownOperation, UnknownOperation,
/// Payload too large (>4 MiB) NotFound,
PayloadTooLarge, PayloadTooLarge,
/// Authentication token mismatch
Unauthorized, Unauthorized,
/// Broker or session unavailable EndpointCredentialForbidden,
Unavailable, Unavailable,
/// WebSocket connection error
ConnectionError, ConnectionError,
/// Request timeout ConnectionClosed,
Timeout, Timeout,
/// Frame parsing or validation error
FrameError, FrameError,
/// Request/response correlation failure
CorrelationError, CorrelationError,
/// Other unclassified error ProtocolError,
ReconnectExhausted,
UncertainAfterSend,
Internal, Internal,
} }
impl fmt::Display for ErrorCode { impl SdkErrorCode {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { /// Return the stable wire representation of this error code.
pub const fn as_str(self) -> &'static str {
match self { match self {
Self::InvalidInput => write!(f, "invalid_input"), Self::InvalidInput => "invalid_input",
Self::UnknownOperation => write!(f, "unknown_operation"), Self::UnknownOperation => "unknown_operation",
Self::PayloadTooLarge => write!(f, "payload_too_large"), Self::NotFound => "not_found",
Self::Unauthorized => write!(f, "unauthorized"), Self::PayloadTooLarge => "payload_too_large",
Self::Unavailable => write!(f, "unavailable"), Self::Unauthorized => "unauthorized",
Self::ConnectionError => write!(f, "connection_error"), Self::EndpointCredentialForbidden => "endpoint_credential_forbidden",
Self::Timeout => write!(f, "timeout"), Self::Unavailable => "unavailable",
Self::FrameError => write!(f, "frame_error"), Self::ConnectionError => "connection_error",
Self::CorrelationError => write!(f, "correlation_error"), Self::ConnectionClosed => "connection_closed",
Self::Internal => write!(f, "internal_error"), Self::Timeout => "timeout",
Self::FrameError => "frame_error",
Self::CorrelationError => "correlation_error",
Self::ProtocolError => "protocol_error",
Self::ReconnectExhausted => "reconnect_exhausted",
Self::UncertainAfterSend => "uncertain_after_send",
Self::Internal => "internal_error",
}
}
/// Parse a server error code, mapping unrecognized values to `Internal`.
pub fn from_wire(value: &str) -> Self {
match value {
"invalid_input" => Self::InvalidInput,
"unknown_operation" => Self::UnknownOperation,
"not_found" => Self::NotFound,
"payload_too_large" => Self::PayloadTooLarge,
"unauthorized" => Self::Unauthorized,
"endpoint_credential_forbidden" => Self::EndpointCredentialForbidden,
"unavailable" => Self::Unavailable,
"connection_error" => Self::ConnectionError,
"connection_closed" => Self::ConnectionClosed,
"timeout" => Self::Timeout,
"frame_error" => Self::FrameError,
"correlation_error" => Self::CorrelationError,
"protocol_error" => Self::ProtocolError,
"reconnect_exhausted" => Self::ReconnectExhausted,
"uncertain_after_send" => Self::UncertainAfterSend,
"internal" | "internal_error" => Self::Internal,
_ => Self::Internal,
} }
} }
} }
impl fmt::Display for SdkErrorCode {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(self.as_str())
}
}
/// Identity retained when a request may have reached the broker but its reply was lost.
///
/// Records deliberately contain no input payload so credentials and other request data
/// are not retained solely for diagnostics. Callers can use the UUID to reconcile the
/// request with an operation-specific lookup before deciding whether to retry.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SdkSentRecord {
pub id: Uuid,
pub operation: Option<String>,
pub sent_at: SystemTime,
}
/// SDK wire protocol error. /// SDK wire protocol error.
#[derive(Debug)] #[derive(Debug, Clone, PartialEq, Eq)]
pub struct SdkError { pub struct SdkError {
pub code: ErrorCode, pub code: SdkErrorCode,
pub message: String, pub message: String,
/// Original server code when it is not represented by `SdkErrorCode`.
pub remote_code: Option<String>,
/// Present only when delivery may have happened before the error was observed.
pub sent_record: Option<SdkSentRecord>,
} }
impl SdkError { impl SdkError {
pub fn new(code: ErrorCode, message: impl Into<String>) -> Self { pub fn new(code: SdkErrorCode, message: impl Into<String>) -> Self {
Self { Self {
code, code,
message: message.into(), message: message.into(),
remote_code: None,
sent_record: None,
} }
} }
pub fn invalid_input(message: impl Into<String>) -> Self { pub fn invalid_input(message: impl Into<String>) -> Self {
Self::new(ErrorCode::InvalidInput, message) Self::new(SdkErrorCode::InvalidInput, message)
} }
pub fn unknown_operation(message: impl Into<String>) -> Self { pub fn unknown_operation(message: impl Into<String>) -> Self {
Self::new(ErrorCode::UnknownOperation, message) Self::new(SdkErrorCode::UnknownOperation, message)
} }
pub fn payload_too_large(message: impl Into<String>) -> Self { pub fn payload_too_large(message: impl Into<String>) -> Self {
Self::new(ErrorCode::PayloadTooLarge, message) Self::new(SdkErrorCode::PayloadTooLarge, message)
} }
pub fn unauthorized(message: impl Into<String>) -> Self { pub fn unauthorized(message: impl Into<String>) -> Self {
Self::new(ErrorCode::Unauthorized, message) Self::new(SdkErrorCode::Unauthorized, message)
} }
pub fn unavailable(message: impl Into<String>) -> Self { pub fn unavailable(message: impl Into<String>) -> Self {
Self::new(ErrorCode::Unavailable, message) Self::new(SdkErrorCode::Unavailable, message)
} }
pub fn connection_error(message: impl Into<String>) -> Self { pub fn connection_error(message: impl Into<String>) -> Self {
Self::new(ErrorCode::ConnectionError, message) Self::new(SdkErrorCode::ConnectionError, message)
} }
pub fn timeout(message: impl Into<String>) -> Self { pub fn timeout(message: impl Into<String>) -> Self {
Self::new(ErrorCode::Timeout, message) Self::new(SdkErrorCode::Timeout, message)
} }
pub fn frame_error(message: impl Into<String>) -> Self { pub fn frame_error(message: impl Into<String>) -> Self {
Self::new(ErrorCode::FrameError, message) Self::new(SdkErrorCode::FrameError, message)
} }
pub fn correlation_error(message: impl Into<String>) -> Self { pub fn correlation_error(message: impl Into<String>) -> Self {
Self::new(ErrorCode::CorrelationError, message) Self::new(SdkErrorCode::CorrelationError, message)
} }
pub fn internal(message: impl Into<String>) -> Self { pub fn internal(message: impl Into<String>) -> Self {
Self::new(ErrorCode::Internal, message) Self::new(SdkErrorCode::Internal, message)
}
pub fn uncertain_after_send(record: SdkSentRecord) -> Self {
let mut error = Self::new(
SdkErrorCode::UncertainAfterSend,
"SDK request outcome is uncertain after the frame was sent.",
);
error.sent_record = Some(record);
error
}
pub fn from_remote(code: &str, message: impl Into<String>) -> Self {
let mut error = Self::new(SdkErrorCode::from_wire(code), message);
if error.code == SdkErrorCode::Internal && code != "internal" && code != "internal_error" {
error.remote_code = Some(code.to_string());
}
error
} }
} }
@@ -123,8 +193,26 @@ mod tests {
#[test] #[test]
fn error_code_display_works() { fn error_code_display_works() {
assert_eq!(ErrorCode::InvalidInput.to_string(), "invalid_input"); assert_eq!(SdkErrorCode::InvalidInput.to_string(), "invalid_input");
assert_eq!(ErrorCode::PayloadTooLarge.to_string(), "payload_too_large"); assert_eq!(
assert_eq!(ErrorCode::Unauthorized.to_string(), "unauthorized"); SdkErrorCode::PayloadTooLarge.to_string(),
"payload_too_large"
);
assert_eq!(SdkErrorCode::Unauthorized.to_string(), "unauthorized");
assert_eq!(
SdkErrorCode::from_wire("endpoint_credential_forbidden"),
SdkErrorCode::EndpointCredentialForbidden
);
assert_eq!(
SdkErrorCode::UncertainAfterSend.to_string(),
"uncertain_after_send"
);
}
#[test]
fn unknown_remote_code_is_preserved_with_typed_internal_code() {
let error = SdkError::from_remote("future_error", "failed");
assert_eq!(error.code, SdkErrorCode::Internal);
assert_eq!(error.remote_code.as_deref(), Some("future_error"));
} }
} }
+179 -112
View File
@@ -1,45 +1,40 @@
//! Frame types and serialization for the SDK wire protocol. //! Typed frame structures and JSON serialization for the SDK wire protocol.
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value}; use serde_json::{Map, Value};
use uuid::Uuid;
use crate::error::{SdkError, SdkResult}; use crate::error::{SdkError, SdkResult};
/// Protocol version implemented. /// Protocol version implemented by this SDK client.
pub const PROTOCOL_VERSION: u32 = 3; pub const PROTOCOL_VERSION: u32 = 3;
/// Maximum UTF-8 byte length for a single broker JSON frame. /// Maximum UTF-8 byte length for a single broker JSON frame.
pub const MAX_FRAME_BYTES: usize = 4 * 1024 * 1024; pub const MAX_FRAME_BYTES: usize = 4 * 1024 * 1024;
/// Frame type identifier. /// Known frame type identifiers in the SDK wire protocol.
#[derive(Debug, Clone, Copy, PartialEq, Eq)] #[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FrameType { pub enum FrameType {
/// Hello frame sent by broker on connection Hello,
ServerHello,
BrokerHello, BrokerHello,
/// Request frame from client to broker
BrokerRequest, BrokerRequest,
/// Response frame from broker to client
BrokerResponse, BrokerResponse,
/// Query request
QueryRequest, QueryRequest,
/// Query response
QueryResponse, QueryResponse,
/// Control request
ControlRequest, ControlRequest,
/// Control response
ControlResponse, ControlResponse,
/// Event replay request
EventReplay, EventReplay,
/// Event replay result
EventReplayResult, EventReplayResult,
/// Session activation
SessionActivate, SessionActivate,
/// Session activation result
SessionActivateResult, SessionActivateResult,
} }
impl FrameType { impl FrameType {
pub fn as_str(&self) -> &'static str { pub const fn as_str(self) -> &'static str {
match self { match self {
Self::Hello => "hello",
Self::ServerHello => "server_hello",
Self::BrokerHello => "broker_hello", Self::BrokerHello => "broker_hello",
Self::BrokerRequest => "broker_request", Self::BrokerRequest => "broker_request",
Self::BrokerResponse => "broker_response", Self::BrokerResponse => "broker_response",
@@ -54,8 +49,10 @@ impl FrameType {
} }
} }
pub fn parse(s: &str) -> SdkResult<Self> { pub fn parse(value: &str) -> SdkResult<Self> {
match s { match value {
"hello" => Ok(Self::Hello),
"server_hello" => Ok(Self::ServerHello),
"broker_hello" => Ok(Self::BrokerHello), "broker_hello" => Ok(Self::BrokerHello),
"broker_request" => Ok(Self::BrokerRequest), "broker_request" => Ok(Self::BrokerRequest),
"broker_response" => Ok(Self::BrokerResponse), "broker_response" => Ok(Self::BrokerResponse),
@@ -67,50 +64,81 @@ impl FrameType {
"event_replay_result" => Ok(Self::EventReplayResult), "event_replay_result" => Ok(Self::EventReplayResult),
"session_activate" => Ok(Self::SessionActivate), "session_activate" => Ok(Self::SessionActivate),
"session_activate_result" => Ok(Self::SessionActivateResult), "session_activate_result" => Ok(Self::SessionActivateResult),
_ => Err(SdkError::frame_error(format!("unknown frame type: {}", s))), _ => Err(SdkError::frame_error(format!(
"unknown frame type: {value}"
))),
} }
} }
} }
/// A frame in the SDK wire protocol. /// A typed JSON-object frame in the SDK wire protocol.
/// ///
/// Frames are JSON objects with a `type` field and optional `id` field for correlation. /// The map remains open-ended to allow operation-specific fields while the frame type,
#[derive(Debug, Clone)] /// correlation UUID, and handshake are validated by the client.
pub struct Frame { #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
/// Underlying JSON value #[serde(transparent)]
pub struct SdkFrame {
/// Underlying JSON object.
pub data: Map<String, Value>, pub data: Map<String, Value>,
} }
impl Frame { impl SdkFrame {
/// Create a new frame from a JSON object. /// Create a frame from a JSON object.
pub fn new(data: Map<String, Value>) -> Self { pub fn new(data: Map<String, Value>) -> Self {
Self { data } Self { data }
} }
/// Create an empty frame. /// Create an empty frame object.
pub fn empty() -> Self { pub fn empty() -> Self {
Self { data: Map::new() } Self::new(Map::new())
} }
/// Get the frame type. /// Parse and validate the frame's `type` field.
pub fn frame_type(&self) -> SdkResult<FrameType> { pub fn frame_type(&self) -> SdkResult<FrameType> {
let frame_type_str = self let frame_type = self
.data .data
.get("type") .get("type")
.and_then(Value::as_str) .and_then(Value::as_str)
.ok_or_else(|| SdkError::frame_error("missing 'type' field"))?; .ok_or_else(|| SdkError::frame_error("missing 'type' field"))?;
FrameType::parse(frame_type_str) FrameType::parse(frame_type)
} }
/// Get the optional request/response ID for correlation. /// Get the optional UUID used for request/response correlation.
pub fn id(&self) -> Option<String> { pub fn id(&self) -> SdkResult<Option<Uuid>> {
match self.data.get("id") {
None => Ok(None),
Some(Value::String(id)) => Uuid::parse_str(id).map(Some).map_err(|error| {
SdkError::correlation_error(format!("invalid frame UUID: {error}"))
}),
Some(_) => Err(SdkError::correlation_error(
"frame 'id' field must be a UUID string",
)),
}
}
/// Set the frame's correlation UUID.
pub fn set_id(&mut self, id: Uuid) {
self.data self.data
.get("id") .insert("id".to_string(), Value::String(id.to_string()));
.and_then(Value::as_str)
.map(String::from)
} }
/// Get the `ok` boolean field from a response frame. /// Read a response correlation UUID from either supported field name.
pub fn response_id(&self) -> SdkResult<Option<Uuid>> {
if self.data.contains_key("id") {
return self.id();
}
match self.data.get("requestId") {
None => Ok(None),
Some(Value::String(id)) => Uuid::parse_str(id).map(Some).map_err(|error| {
SdkError::correlation_error(format!("invalid response UUID: {error}"))
}),
Some(_) => Err(SdkError::correlation_error(
"frame 'requestId' field must be a UUID string",
)),
}
}
/// Read the `ok` boolean from a response frame.
pub fn ok(&self) -> SdkResult<bool> { pub fn ok(&self) -> SdkResult<bool> {
self.data self.data
.get("ok") .get("ok")
@@ -118,33 +146,64 @@ impl Frame {
.ok_or_else(|| SdkError::frame_error("missing or invalid 'ok' field")) .ok_or_else(|| SdkError::frame_error("missing or invalid 'ok' field"))
} }
/// Get the error object from a response frame. /// Read the remote error code and message from an error response.
pub fn error(&self) -> Option<SdkResult<(String, String)>> { pub fn error(&self) -> Option<SdkResult<(String, String)>> {
let error_obj = self.data.get("error")?.as_object()?; let error = self.data.get("error")?.as_object()?;
let code = error_obj.get("code")?.as_str()?.to_string(); let code = error.get("code")?.as_str()?.to_string();
let message = error_obj.get("message")?.as_str()?.to_string(); let message = error.get("message")?.as_str()?.to_string();
Some(Ok((code, message))) Some(Ok((code, message)))
} }
/// Get the result field from a response frame. /// Get the response result field.
pub fn result(&self) -> Option<Value> { pub fn result(&self) -> Option<Value> {
self.data.get("result").cloned() self.data.get("result").cloned()
} }
/// Convert to JSON string. /// Serialize this frame to compact JSON.
pub fn to_json(&self) -> SdkResult<String> { pub fn to_json(&self) -> SdkResult<String> {
serde_json::to_string(&self.data) serde_json::to_string(&self.data)
.map_err(|e| SdkError::frame_error(format!("JSON serialization error: {}", e))) .map_err(|error| SdkError::frame_error(format!("JSON serialization error: {error}")))
} }
/// Parse from JSON string. /// Parse a JSON object as an SDK frame.
pub fn from_json(json_str: &str) -> SdkResult<Self> { pub fn from_json(json: &str) -> SdkResult<Self> {
let data: Map<String, Value> = serde_json::from_str(json_str) match serde_json::from_str::<Value>(json)
.map_err(|e| SdkError::frame_error(format!("JSON parsing error: {}", e)))?; .map_err(|error| SdkError::frame_error(format!("JSON parsing error: {error}")))?
Ok(Self::new(data)) {
Value::Object(data) => Ok(Self::new(data)),
_ => Err(SdkError::frame_error("SDK frame must be a JSON object")),
}
} }
/// Create a broker hello frame. /// Create the protocol-v3 client hello with the advertised capabilities.
pub fn client_hello(capabilities: &[String]) -> Self {
let mut data = Map::new();
data.insert("type".to_string(), Value::String("hello".to_string()));
data.insert(
"protocolVersion".to_string(),
Value::Number(PROTOCOL_VERSION.into()),
);
data.insert(
"capabilities".to_string(),
Value::Array(capabilities.iter().cloned().map(Value::String).collect()),
);
Self::new(data)
}
/// Return whether this frame completes a server hello handshake.
pub fn is_server_hello(&self) -> bool {
matches!(
self.data.get("type").and_then(Value::as_str),
Some("hello" | "server_hello" | "broker_hello")
)
}
/// Read an optional connection identifier from a server hello.
pub fn connection_id(&self) -> Option<&str> {
self.data.get("connectionId").and_then(Value::as_str)
}
/// Create a broker hello frame for protocol tests and mock servers.
pub fn broker_hello() -> Self { pub fn broker_hello() -> Self {
let mut data = Map::new(); let mut data = Map::new();
data.insert( data.insert(
@@ -158,63 +217,63 @@ impl Frame {
Self::new(data) Self::new(data)
} }
/// Create a broker request frame. /// Create a broker operation request frame.
pub fn broker_request(id: String, operation: String, input: Value) -> Self { pub fn broker_request(id: Uuid, operation: String, input: Value) -> Self {
let mut data = Map::new(); let mut data = Map::new();
data.insert( data.insert(
"type".to_string(), "type".to_string(),
Value::String("broker_request".to_string()), Value::String("broker_request".to_string()),
); );
data.insert("id".to_string(), Value::String(id)); data.insert("id".to_string(), Value::String(id.to_string()));
data.insert("operation".to_string(), Value::String(operation)); data.insert("operation".to_string(), Value::String(operation));
data.insert("input".to_string(), input); data.insert("input".to_string(), input);
Self::new(data) Self::new(data)
} }
/// Create a broker response frame. /// Create a successful broker response frame.
pub fn broker_response_ok(id: String, result: Value) -> Self { pub fn broker_response_ok(id: Uuid, result: Value) -> Self {
let mut data = Map::new(); let mut data = Map::new();
data.insert( data.insert(
"type".to_string(), "type".to_string(),
Value::String("broker_response".to_string()), Value::String("broker_response".to_string()),
); );
data.insert("id".to_string(), Value::String(id)); data.insert("id".to_string(), Value::String(id.to_string()));
data.insert("ok".to_string(), Value::Bool(true)); data.insert("ok".to_string(), Value::Bool(true));
data.insert("result".to_string(), result); data.insert("result".to_string(), result);
Self::new(data) Self::new(data)
} }
/// Create a broker error response frame. /// Create an error broker response frame.
pub fn broker_response_error(id: String, code: String, message: String) -> Self { pub fn broker_response_error(id: Uuid, code: String, message: String) -> Self {
let mut error_data = Map::new(); let mut error = Map::new();
error_data.insert("code".to_string(), Value::String(code)); error.insert("code".to_string(), Value::String(code));
error_data.insert("message".to_string(), Value::String(message)); error.insert("message".to_string(), Value::String(message));
let mut data = Map::new(); let mut data = Map::new();
data.insert( data.insert(
"type".to_string(), "type".to_string(),
Value::String("broker_response".to_string()), Value::String("broker_response".to_string()),
); );
data.insert("id".to_string(), Value::String(id)); data.insert("id".to_string(), Value::String(id.to_string()));
data.insert("ok".to_string(), Value::Bool(false)); data.insert("ok".to_string(), Value::Bool(false));
data.insert("error".to_string(), Value::Object(error_data)); data.insert("error".to_string(), Value::Object(error));
Self::new(data) Self::new(data)
} }
/// Get the payload size in bytes. /// Get the serialized frame length in UTF-8 bytes.
pub fn byte_size(&self) -> usize { pub fn byte_size(&self) -> usize {
self.to_json().map(|s| s.len()).unwrap_or(0) self.to_json().map_or(0, |json| json.len())
} }
/// Get the operation name from a broker request frame. /// Read the operation name from a request frame.
pub fn operation(&self) -> Option<String> { pub fn operation(&self) -> Option<String> {
self.data self.data
.get("operation") .get("operation")
.and_then(Value::as_str) .and_then(Value::as_str)
.map(String::from) .map(str::to_owned)
} }
/// Get the input value from a broker request frame. /// Read the request input object.
pub fn input(&self) -> Option<Value> { pub fn input(&self) -> Option<Value> {
self.data.get("input").cloned() self.data.get("input").cloned()
} }
@@ -226,7 +285,7 @@ mod tests {
#[test] #[test]
fn frame_type_round_trip() { fn frame_type_round_trip() {
assert_eq!(FrameType::BrokerHello.as_str(), "broker_hello"); assert_eq!(FrameType::Hello.as_str(), "hello");
assert_eq!( assert_eq!(
FrameType::parse("broker_hello").unwrap(), FrameType::parse("broker_hello").unwrap(),
FrameType::BrokerHello FrameType::BrokerHello
@@ -235,68 +294,76 @@ mod tests {
#[test] #[test]
fn frame_parsing_extracts_type() { fn frame_parsing_extracts_type() {
let frame = Frame::broker_hello(); assert_eq!(
assert_eq!(frame.frame_type().unwrap(), FrameType::BrokerHello); SdkFrame::broker_hello().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] #[test]
fn frame_json_round_trip() { fn frame_id_round_trips_as_a_uuid() {
let frame = Frame::broker_hello(); let id = Uuid::new_v4();
let json_str = frame.to_json().unwrap(); let frame = SdkFrame::broker_request(id, "test.operation".to_string(), Value::Null);
let parsed = Frame::from_json(&json_str).unwrap(); assert_eq!(frame.id().unwrap(), Some(id));
assert_eq!(parsed.frame_type().unwrap(), FrameType::BrokerHello);
} }
#[test] #[test]
fn broker_response_error_contains_code_and_message() { fn json_round_trip_preserves_the_typed_frame() {
let frame = Frame::broker_response_error( let frame = SdkFrame::client_hello(&["capability.one".to_string()]);
"id1".to_string(), let parsed = SdkFrame::from_json(&frame.to_json().unwrap()).unwrap();
assert_eq!(parsed, frame);
assert_eq!(parsed.frame_type().unwrap(), FrameType::Hello);
}
#[test]
fn client_hello_advertises_protocol_and_capabilities() {
let frame = SdkFrame::client_hello(&["capability.one".to_string()]);
assert_eq!(frame.data["protocolVersion"], PROTOCOL_VERSION);
assert_eq!(
frame.data["capabilities"],
serde_json::json!(["capability.one"])
);
}
#[test]
fn broker_error_response_exposes_code_and_message() {
let frame = SdkFrame::broker_response_error(
Uuid::new_v4(),
"invalid_input".to_string(), "invalid_input".to_string(),
"test error".to_string(), "bad input".to_string(),
); );
let (code, message) = frame.error().unwrap().unwrap(); assert_eq!(
assert_eq!(code, "invalid_input"); frame.error().unwrap().unwrap(),
assert_eq!(message, "test error"); ("invalid_input".into(), "bad input".into())
);
assert!(!frame.ok().unwrap());
} }
#[test] #[test]
fn frame_extracts_operation_from_broker_request() { fn operation_and_input_are_recoverable() {
let frame = Frame::broker_request( let input = serde_json::json!({"session_id": "session-1"});
"test-id".to_string(), let frame = SdkFrame::broker_request(
"query.list_sessions".to_string(), Uuid::new_v4(),
Value::Object(Map::new()),
);
assert_eq!(frame.operation(), Some("query.list_sessions".to_string()));
}
#[test]
fn frame_extracts_input_from_broker_request() {
let input = Value::Object({
let mut m = Map::new();
m.insert("session_id".to_string(), Value::String("123".to_string()));
m
});
let frame = Frame::broker_request(
"test-id".to_string(),
"query.get_session".to_string(), "query.get_session".to_string(),
input.clone(), input.clone(),
); );
assert_eq!(frame.operation().as_deref(), Some("query.get_session"));
assert_eq!(frame.input(), Some(input)); assert_eq!(frame.input(), Some(input));
} }
#[test] #[test]
fn frame_without_operation_returns_none() { fn malformed_json_and_non_object_values_are_rejected() {
let frame = Frame::broker_hello(); assert!(SdkFrame::from_json("{").is_err());
assert_eq!(frame.operation(), None); assert!(SdkFrame::from_json("[]").is_err());
assert!(SdkFrame::from_json("null").is_err());
}
#[test]
fn invalid_correlation_uuid_is_rejected() {
let frame = SdkFrame::from_json(r#"{"type":"broker_response","id":"not-a-uuid"}"#).unwrap();
assert_eq!(
frame.id().unwrap_err().code,
crate::error::SdkErrorCode::CorrelationError
);
} }
} }
+3 -3
View File
@@ -16,7 +16,7 @@ pub mod frame;
pub mod registry; pub mod registry;
pub mod transport; pub mod transport;
pub use error::{SdkError, SdkResult}; pub use error::{SdkError, SdkErrorCode, SdkResult, SdkSentRecord};
pub use frame::{Frame, FrameType}; pub use frame::{FrameType, SdkFrame};
pub use registry::{OperationMetadata, OperationRegistry, Schema}; pub use registry::{OperationMetadata, OperationRegistry, Schema};
pub use transport::{SdkTransport, SdkTransportConfig}; pub use transport::{SdkClient, SdkClientConfig};
File diff suppressed because it is too large. Load diff
+22 -27
View File
@@ -10,17 +10,18 @@
//! - Schema validation for operation inputs/outputs //! - Schema validation for operation inputs/outputs
//! - Discovery of available operations //! - Discovery of available operations
use sdk::{Frame, OperationMetadata, OperationRegistry, Schema}; use sdk::{OperationMetadata, OperationRegistry, Schema, SdkFrame};
use serde_json::json; use serde_json::json;
use uuid::Uuid;
fn test_id() -> Uuid {
Uuid::new_v4()
}
#[test] #[test]
fn gap_sdk_allows_undefined_operations() { fn gap_sdk_allows_undefined_operations() {
// BEFORE: No validation - any operation name is accepted // BEFORE: No validation - any operation name is accepted
let frame = Frame::broker_request( let frame = SdkFrame::broker_request(test_id(), "undefined.operation".to_string(), json!({}));
"test-id".to_string(),
"undefined.operation".to_string(),
json!({}),
);
// The frame is created successfully with no validation // The frame is created successfully with no validation
assert_eq!(frame.operation(), Some("undefined.operation".to_string())); assert_eq!(frame.operation(), Some("undefined.operation".to_string()));
@@ -48,9 +49,9 @@ fn fix_operation_registry_validates_operations() {
#[test] #[test]
fn gap_sdk_has_no_operation_discovery() { fn gap_sdk_has_no_operation_discovery() {
// BEFORE: No way to discover what operations are available // BEFORE: No way to discover what operations are available
// The Frame struct has no way to enumerate operations // The SdkFrame struct has no way to enumerate operations
let frame = Frame::broker_request( let frame = SdkFrame::broker_request(
"test-id".to_string(), test_id(),
"query.list_sessions".to_string(), "query.list_sessions".to_string(),
json!({"some": "input"}), json!({"some": "input"}),
); );
@@ -84,14 +85,14 @@ fn fix_operation_registry_enables_discovery() {
fn gap_sdk_has_no_input_schema_validation() { fn gap_sdk_has_no_input_schema_validation() {
// BEFORE: No validation of input schemas // BEFORE: No validation of input schemas
// Any input shape is accepted for any operation // Any input shape is accepted for any operation
let frame_with_valid_input = Frame::broker_request( let frame_with_valid_input = SdkFrame::broker_request(
"test-1".to_string(), test_id(),
"query.get_session".to_string(), "query.get_session".to_string(),
json!({"session_id": "123"}), // Correct structure json!({"session_id": "123"}), // Correct structure
); );
let frame_with_invalid_input = Frame::broker_request( let frame_with_invalid_input = SdkFrame::broker_request(
"test-2".to_string(), test_id(),
"query.get_session".to_string(), "query.get_session".to_string(),
json!({"wrong_field": "value"}), // Missing session_id, wrong field json!({"wrong_field": "value"}), // Missing session_id, wrong field
); );
@@ -126,8 +127,8 @@ fn fix_operation_registry_validates_input_schemas() {
#[test] #[test]
fn gap_sdk_has_no_output_schema_validation() { fn gap_sdk_has_no_output_schema_validation() {
// BEFORE: No validation of response schemas // BEFORE: No validation of response schemas
let response_frame = Frame::broker_response_ok( let response_frame = SdkFrame::broker_response_ok(
"test-id".to_string(), test_id(),
json!({"any": "response"}), // No validation of structure json!({"any": "response"}), // No validation of structure
); );
@@ -158,8 +159,8 @@ fn fix_operation_registry_validates_output_schemas() {
#[test] #[test]
fn gap_sdk_transports_ignore_operation_names() { fn gap_sdk_transports_ignore_operation_names() {
// BEFORE: The transport layer sends any operation without validation // BEFORE: The transport layer sends any operation without validation
let frame = Frame::broker_request( let frame = SdkFrame::broker_request(
"req-123".to_string(), test_id(),
"this.is.not.a.real.operation".to_string(), "this.is.not.a.real.operation".to_string(),
json!({"data": "value"}), json!({"data": "value"}),
); );
@@ -175,11 +176,8 @@ fn fix_registry_integrates_with_frame_operations() {
let registry = OperationRegistry::with_standard_operations(); let registry = OperationRegistry::with_standard_operations();
// Create a frame with a valid operation // Create a frame with a valid operation
let valid_frame = Frame::broker_request( let valid_frame =
"req-123".to_string(), SdkFrame::broker_request(test_id(), "query.list_sessions".to_string(), json!({}));
"query.list_sessions".to_string(),
json!({}),
);
// Extract and validate the operation // Extract and validate the operation
if let Some(op_name) = valid_frame.operation() { if let Some(op_name) = valid_frame.operation() {
@@ -189,11 +187,8 @@ fn fix_registry_integrates_with_frame_operations() {
} }
// Create a frame with an invalid operation // Create a frame with an invalid operation
let invalid_frame = Frame::broker_request( let invalid_frame =
"req-124".to_string(), SdkFrame::broker_request(test_id(), "invalid.operation".to_string(), json!({}));
"invalid.operation".to_string(),
json!({}),
);
// Extract and validate the operation // Extract and validate the operation
if let Some(op_name) = invalid_frame.operation() { if let Some(op_name) = invalid_frame.operation() {