diff --git a/Cargo.lock b/Cargo.lock index ce820c9..023753a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -397,6 +397,7 @@ version = "0.1.0" dependencies = [ "axum", "crank-core", + "crank-mapping", "reqwest", "roxmltree", "serde", diff --git a/crates/crank-adapter-soap/Cargo.toml b/crates/crank-adapter-soap/Cargo.toml index 0e1d376..9fe65f9 100644 --- a/crates/crank-adapter-soap/Cargo.toml +++ b/crates/crank-adapter-soap/Cargo.toml @@ -7,6 +7,7 @@ version.workspace = true [dependencies] crank-core = { path = "../crank-core" } +crank-mapping = { path = "../crank-mapping" } reqwest.workspace = true roxmltree.workspace = true serde.workspace = true diff --git a/crates/crank-adapter-soap/src/client.rs b/crates/crank-adapter-soap/src/client.rs index bcf67bc..a0b3b96 100644 --- a/crates/crank-adapter-soap/src/client.rs +++ b/crates/crank-adapter-soap/src/client.rs @@ -1,10 +1,12 @@ use std::{collections::BTreeMap, time::Duration}; use crank_core::{SoapTarget, SoapVersion}; +use crank_mapping::JsonPath; use reqwest::{ Client, header::{HeaderMap, HeaderName, HeaderValue}, }; +use serde_json::{Value, json}; use crate::{SoapAdapterError, SoapRequest, SoapResponse, xml}; @@ -36,11 +38,14 @@ impl SoapAdapter { .as_deref() .ok_or(SoapAdapterError::MissingEndpoint)?; let headers = build_headers(target, request)?; + let envelope_headers = resolve_envelope_headers(target, request)?; let envelope = xml::build_envelope( &target.operation_name, target.metadata.namespaces.first().map(String::as_str), &request.body, envelope_namespace(target.soap_version), + target.binding_style, + &envelope_headers, ); let response = self .client @@ -70,6 +75,56 @@ impl SoapAdapter { } } +fn resolve_envelope_headers( + target: &SoapTarget, + request: &SoapRequest, +) -> Result, SoapAdapterError> { + let context = json!({ + "request": { + "body": request.body.clone(), + }, + "mcp": request.body.clone(), + }); + let mut rendered = Vec::new(); + + for header in &target.headers { + let Some(value) = resolve_header_value(header, &context, &request.body)? else { + if header.required { + return Err(SoapAdapterError::MissingRequiredHeader { + name: header.name.clone(), + }); + } + continue; + }; + + rendered.push(xml::SoapEnvelopeHeader { + name: header.name.clone(), + namespace_uri: header.namespace_uri.clone(), + value, + }); + } + + Ok(rendered) +} + +fn resolve_header_value( + header: &crank_core::SoapHeaderConfig, + context: &Value, + body: &Value, +) -> Result, SoapAdapterError> { + if let Some(path) = header.value_path.as_deref() { + let path = JsonPath::parse(path).map_err(|_| SoapAdapterError::InvalidHeaderValuePath { + path: path.to_owned(), + })?; + return Ok(path.read(context).cloned()); + } + + Ok(match body { + Value::Object(map) => map.get(&header.name).cloned(), + _ => None, + }) +} + fn build_headers( target: &SoapTarget, request: &SoapRequest, @@ -209,4 +264,63 @@ mod tests { assert_eq!(response.status_code, 200); assert_eq!(response.body["id"], "lead_123"); } + + #[tokio::test] + async fn renders_soap_headers_and_rpc_literal_children() { + async fn handler(body: Bytes) -> String { + let text = String::from_utf8_lossy(&body); + assert!(text.contains("")); + assert!( + text.contains( + r#"corr-123"# + ) + ); + assert!(text.contains("user@example.com")); + r#" + + + lead_rpc + + + "# + .to_owned() + } + + let app = Router::new().route("/", post(handler)); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + + let target = SoapTarget { + binding_style: SoapBindingStyle::RpcLiteral, + headers: vec![crank_core::SoapHeaderConfig { + name: "CorrelationId".to_owned(), + namespace_uri: Some("urn:headers".to_owned()), + required: true, + value_path: Some("$.request.body.correlation_id".to_owned()), + }], + ..test_target(format!("http://{}", address)) + }; + + let adapter = SoapAdapter::new(); + let response = adapter + .execute( + &target, + &SoapRequest { + headers: BTreeMap::new(), + body: json!({ + "email": "user@example.com", + "correlation_id": "corr-123" + }), + timeout_ms: 1_000, + }, + ) + .await + .unwrap(); + + assert_eq!(response.body["id"], "lead_rpc"); + } } diff --git a/crates/crank-adapter-soap/src/error.rs b/crates/crank-adapter-soap/src/error.rs index 4fb8915..92507c0 100644 --- a/crates/crank-adapter-soap/src/error.rs +++ b/crates/crank-adapter-soap/src/error.rs @@ -5,6 +5,10 @@ use thiserror::Error; pub enum SoapAdapterError { #[error("soap endpoint is missing")] MissingEndpoint, + #[error("invalid SOAP header value path {path}")] + InvalidHeaderValuePath { path: String }, + #[error("required SOAP header {name} is missing")] + MissingRequiredHeader { name: String }, #[error("request failed")] Transport(#[from] reqwest::Error), #[error("invalid xml payload")] diff --git a/crates/crank-adapter-soap/src/xml.rs b/crates/crank-adapter-soap/src/xml.rs index 92e39f2..59adad3 100644 --- a/crates/crank-adapter-soap/src/xml.rs +++ b/crates/crank-adapter-soap/src/xml.rs @@ -1,24 +1,45 @@ +use crank_core::SoapBindingStyle; use roxmltree::{Document, Node}; use serde_json::{Map, Value}; use crate::SoapAdapterError; +#[derive(Clone, Debug, PartialEq)] +pub struct SoapEnvelopeHeader { + pub name: String, + pub namespace_uri: Option, + pub value: Value, +} + pub fn build_envelope( operation_name: &str, namespace: Option<&str>, body: &Value, envelope_namespace: &str, + binding_style: SoapBindingStyle, + headers: &[SoapEnvelopeHeader], ) -> String { let mut xml = String::new(); xml.push_str(&format!( r#""# )); + if !headers.is_empty() { + xml.push_str(""); + for header in headers { + append_header(&mut xml, header); + } + xml.push_str(""); + } xml.push_str(""); match namespace { Some(namespace) => xml.push_str(&format!(r#""#)), None => xml.push_str(&format!("<{operation_name}>")), } - append_value_children(&mut xml, body); + let child_prefix = match binding_style { + SoapBindingStyle::DocumentLiteral => None, + SoapBindingStyle::RpcLiteral => namespace.map(|_| "m"), + }; + append_value_children(&mut xml, body, child_prefix); match namespace { Some(_) => xml.push_str(&format!("")), None => xml.push_str(&format!("")), @@ -64,6 +85,38 @@ fn parse_fault(fault: Node<'_, '_>) -> SoapAdapterError { } } +fn append_header(xml: &mut String, header: &SoapEnvelopeHeader) { + match header.namespace_uri.as_deref() { + Some(namespace_uri) => { + xml.push_str(&format!( + r#""#, + header.name, namespace_uri + )); + append_header_value(xml, &header.value); + xml.push_str(&format!("", header.name)); + } + None => { + xml.push('<'); + xml.push_str(&header.name); + xml.push('>'); + append_header_value(xml, &header.value); + xml.push_str("'); + } + } +} + +fn append_header_value(xml: &mut String, value: &Value) { + match value { + Value::Null => {} + Value::Bool(value) => xml.push_str(if *value { "true" } else { "false" }), + Value::Number(value) => xml.push_str(&value.to_string()), + Value::String(value) => xml.push_str(&escape_xml(value)), + Value::Array(_) | Value::Object(_) => append_value_children(xml, value, None), + } +} + fn find_nested_text(node: Node<'_, '_>, path: &[&str]) -> Option { let mut current = node; for segment in path { @@ -79,70 +132,82 @@ fn find_nested_text(node: Node<'_, '_>, path: &[&str]) -> Option { .map(ToOwned::to_owned) } -fn append_value_children(xml: &mut String, value: &Value) { +fn append_value_children(xml: &mut String, value: &Value, namespace_prefix: Option<&str>) { match value { Value::Object(object) => { for (key, value) in object { - append_named_value(xml, key, value); + append_named_value(xml, key, value, namespace_prefix); } } - other => append_named_value(xml, "value", other), + other => append_named_value(xml, "value", other, namespace_prefix), } } -fn append_named_value(xml: &mut String, name: &str, value: &Value) { +fn append_named_value(xml: &mut String, name: &str, value: &Value, namespace_prefix: Option<&str>) { match value { Value::Array(items) => { for item in items { - append_named_value(xml, name, item); + append_named_value(xml, name, item, namespace_prefix); } } Value::Object(object) => { - xml.push('<'); - xml.push_str(name); - xml.push('>'); + push_start_tag(xml, name, namespace_prefix); for (child_name, child_value) in object { - append_named_value(xml, child_name, child_value); + append_named_value(xml, child_name, child_value, namespace_prefix); } - xml.push_str("'); + push_end_tag(xml, name, namespace_prefix); } Value::Null => { - xml.push('<'); - xml.push_str(name); - xml.push_str("/>"); + push_empty_tag(xml, name, namespace_prefix); } Value::Bool(value) => { - xml.push('<'); - xml.push_str(name); - xml.push('>'); + push_start_tag(xml, name, namespace_prefix); xml.push_str(if *value { "true" } else { "false" }); - xml.push_str("'); + push_end_tag(xml, name, namespace_prefix); } Value::Number(value) => { - xml.push('<'); - xml.push_str(name); - xml.push('>'); + push_start_tag(xml, name, namespace_prefix); xml.push_str(&value.to_string()); - xml.push_str("'); + push_end_tag(xml, name, namespace_prefix); } Value::String(value) => { - xml.push('<'); - xml.push_str(name); - xml.push('>'); + push_start_tag(xml, name, namespace_prefix); xml.push_str(&escape_xml(value)); - xml.push_str("'); + push_end_tag(xml, name, namespace_prefix); } } } +fn push_start_tag(xml: &mut String, name: &str, namespace_prefix: Option<&str>) { + xml.push('<'); + if let Some(prefix) = namespace_prefix { + xml.push_str(prefix); + xml.push(':'); + } + xml.push_str(name); + xml.push('>'); +} + +fn push_end_tag(xml: &mut String, name: &str, namespace_prefix: Option<&str>) { + xml.push_str("'); +} + +fn push_empty_tag(xml: &mut String, name: &str, namespace_prefix: Option<&str>) { + xml.push('<'); + if let Some(prefix) = namespace_prefix { + xml.push_str(prefix); + xml.push(':'); + } + xml.push_str(name); + xml.push_str("/>"); +} + fn escape_xml(value: &str) -> String { value .replace('&', "&") @@ -190,9 +255,10 @@ fn node_to_json(node: Node<'_, '_>, _unwrap_root: bool) -> Value { #[cfg(test)] mod tests { - use serde_json::json; + use serde_json::{Value, json}; - use super::{build_envelope, decode_envelope}; + use super::{SoapEnvelopeHeader, build_envelope, decode_envelope}; + use crank_core::SoapBindingStyle; #[test] fn builds_document_literal_envelope() { @@ -201,6 +267,8 @@ mod tests { Some("urn:crm"), &json!({"email":"user@example.com","source":"mcp"}), "http://schemas.xmlsoap.org/soap/envelope/", + SoapBindingStyle::DocumentLiteral, + &[], ); assert!(xml.contains("")); + assert!( + xml.contains(r#"corr-123"#) + ); + assert!(xml.contains("user@example.com")); + } } diff --git a/crates/crank-adapter-websocket/src/client.rs b/crates/crank-adapter-websocket/src/client.rs index a67486e..bbb3154 100644 --- a/crates/crank-adapter-websocket/src/client.rs +++ b/crates/crank-adapter-websocket/src/client.rs @@ -19,6 +19,11 @@ use crate::{ WebsocketWindowResponse, }; +enum WindowCollectionStatus { + WindowExpired, + MaxItemsReached, +} + #[derive(Clone, Debug)] pub struct WebsocketAdapter; @@ -60,33 +65,51 @@ impl WebsocketAdapter { send_subscribe_message(&mut stream, target).await?; - let completed = collect_window( + let status = match collect_window( &mut stream, request.max_items, deadline, heartbeat.as_ref(), &mut items, ) - .await?; + .await + { + Ok(status) => status, + Err(WebsocketAdapterError::ClosedEarly) => { + if attempts >= reconnect.max_attempts { + return Err(WebsocketAdapterError::ReconnectExhausted); + } + attempts = attempts.saturating_add(1); + reconnect_if_needed(&reconnect, attempts).await; + continue; + } + Err(error) => return Err(error), + }; - if completed { - send_unsubscribe_message(&mut stream, target).await?; - return Ok(WebsocketWindowResponse { - status_code: 101, - headers: connected_headers, - body: serde_json::json!({ - "items": items, - "done": true, - }), - }); + match status { + WindowCollectionStatus::WindowExpired => { + send_unsubscribe_message(&mut stream, target).await?; + return Ok(WebsocketWindowResponse { + status_code: 101, + headers: connected_headers, + body: serde_json::json!({ + "items": items, + "done": false, + }), + }); + } + WindowCollectionStatus::MaxItemsReached => { + send_unsubscribe_message(&mut stream, target).await?; + return Ok(WebsocketWindowResponse { + status_code: 101, + headers: connected_headers, + body: serde_json::json!({ + "items": items, + "done": true, + }), + }); + } } - - if attempts >= reconnect.max_attempts { - return Err(WebsocketAdapterError::ReconnectExhausted); - } - - attempts = attempts.saturating_add(1); - reconnect_if_needed(&reconnect, attempts).await; } } } @@ -189,12 +212,12 @@ async fn collect_window( deadline: Instant, heartbeat: Option<&HeartbeatPolicy>, items: &mut Vec, -) -> Result { +) -> Result { let mut heartbeat_deadline = heartbeat.map(|policy| Instant::now() + policy.interval); loop { if Instant::now() >= deadline { - return Ok(!items.is_empty()); + return Ok(WindowCollectionStatus::WindowExpired); } let now = Instant::now(); @@ -212,17 +235,17 @@ async fn collect_window( continue; } - return Ok(!items.is_empty()); + return Ok(WindowCollectionStatus::WindowExpired); } frame = read_next_frame(stream) => { match frame? { Some(value) => { items.push(value); if max_items.is_some_and(|limit| items.len() as u32 >= limit) { - return Ok(true); + return Ok(WindowCollectionStatus::MaxItemsReached); } } - None => return Ok(!items.is_empty()), + None => return Err(WebsocketAdapterError::ClosedEarly), } heartbeat_deadline = heartbeat.map(|policy| Instant::now() + policy.interval); } @@ -374,6 +397,39 @@ mod tests { assert_eq!(response.body["items"].as_array().unwrap().len(), 3); assert_eq!(response.body["items"][2]["seq"], 3); + assert_eq!(response.body["done"], true); + } + + #[tokio::test] + async fn reconnects_after_partial_close_without_marking_done_early() { + let target_url = spawn_partial_close_server().await; + let adapter = WebsocketAdapter::new(); + let target = WebsocketTarget { + url: target_url, + subprotocols: Vec::new(), + subscribe_message_template: None, + unsubscribe_message_template: None, + static_headers: BTreeMap::new(), + }; + let response = adapter + .execute_window( + &target, + &WebsocketWindowRequest { + headers: BTreeMap::new(), + window_duration_ms: 1_000, + max_items: Some(3), + heartbeat_interval_ms: None, + reconnect_max_attempts: 2, + reconnect_backoff_ms: 10, + }, + ) + .await + .unwrap(); + + assert_eq!(response.body["items"].as_array().unwrap().len(), 3); + assert_eq!(response.body["items"][0]["seq"], 1); + assert_eq!(response.body["items"][2]["seq"], 3); + assert_eq!(response.body["done"], true); } async fn spawn_server(received: Arc>>, close_early: bool) -> String { @@ -439,4 +495,43 @@ mod tests { format!("ws://{}", addr) } + + async fn spawn_partial_close_server() -> String { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + + tokio::spawn(async move { + let mut accepted = 0_u32; + loop { + let (stream, _) = listener.accept().await.unwrap(); + accepted = accepted.saturating_add(1); + tokio::spawn(async move { + let websocket = + accept_hdr_async(stream, |_request: &Request, response: Response| { + Ok(response) + }) + .await + .unwrap(); + let (mut sink, _source) = websocket.split(); + let payloads = if accepted == 1 { + vec![json!({"seq": 1})] + } else { + vec![json!({"seq": 2}), json!({"seq": 3})] + }; + + for payload in payloads { + sink.send(tokio_tungstenite::tungstenite::Message::Text( + payload.to_string().into(), + )) + .await + .unwrap(); + } + + let _ = sink.close().await; + }); + } + }); + + format!("ws://{}", addr) + } }