mockserver_client/
breakpoint.rs

1//! Breakpoint matcher registration, callback WebSocket client, and handler routing.
2//!
3//! This module provides the WebSocket callback stack for interactive breakpoints.
4//! It connects to `/_mockserver_callback_websocket`, obtains a `clientId`, and
5//! dispatches paused items to per-breakpoint-id handlers.
6
7use base64::{engine::general_purpose::STANDARD as BASE64, Engine};
8use serde::{Deserialize, Serialize};
9use serde_json::Value;
10use std::collections::HashMap;
11use std::sync::{Arc, Mutex};
12use tungstenite::{connect, Message};
13use url::Url;
14
15use crate::error::{Error, Result};
16use crate::model::HttpRequest;
17
18// ---------------------------------------------------------------------------
19// Breakpoint phase constants
20// ---------------------------------------------------------------------------
21
22/// Breakpoint interception phase.
23pub mod phase {
24    pub const REQUEST: &str = "REQUEST";
25    pub const RESPONSE: &str = "RESPONSE";
26    pub const RESPONSE_STREAM: &str = "RESPONSE_STREAM";
27    pub const INBOUND_STREAM: &str = "INBOUND_STREAM";
28}
29
30// ---------------------------------------------------------------------------
31// Wire-contract DTOs
32// ---------------------------------------------------------------------------
33
34/// Registration request for a breakpoint matcher.
35#[derive(Debug, Clone, Serialize, Deserialize)]
36#[serde(rename_all = "camelCase")]
37pub struct BreakpointMatcherRegistration {
38    pub http_request: HttpRequest,
39    pub phases: Vec<String>,
40    #[serde(skip_serializing_if = "Option::is_none")]
41    pub client_id: Option<String>,
42}
43
44/// Response from registering a breakpoint matcher.
45#[derive(Debug, Clone, Serialize, Deserialize)]
46pub struct BreakpointMatcherResponse {
47    pub id: String,
48    pub phases: Vec<String>,
49}
50
51/// An entry in the list of registered breakpoint matchers.
52#[derive(Debug, Clone, Serialize, Deserialize)]
53#[serde(rename_all = "camelCase")]
54pub struct BreakpointMatcherEntry {
55    pub id: String,
56    pub http_request: Value,
57    pub phases: Vec<String>,
58    #[serde(skip_serializing_if = "Option::is_none")]
59    pub client_id: Option<String>,
60}
61
62/// Response from listing breakpoint matchers.
63#[derive(Debug, Clone, Serialize, Deserialize)]
64pub struct BreakpointMatcherList {
65    pub matchers: Vec<BreakpointMatcherEntry>,
66}
67
68/// A paused stream frame pushed by the server over the callback WebSocket.
69#[derive(Debug, Clone, Serialize, Deserialize)]
70#[serde(rename_all = "camelCase")]
71pub struct PausedStreamFrame {
72    pub correlation_id: String,
73    #[serde(default)]
74    pub stream_id: String,
75    #[serde(default)]
76    pub sequence_number: i64,
77    #[serde(default)]
78    pub direction: String,
79    #[serde(default)]
80    pub phase: String,
81    #[serde(default)]
82    pub body: String,
83    #[serde(skip_serializing_if = "Option::is_none")]
84    pub request_method: Option<String>,
85    #[serde(skip_serializing_if = "Option::is_none")]
86    pub request_path: Option<String>,
87    #[serde(skip_serializing_if = "Option::is_none")]
88    pub breakpoint_id: Option<String>,
89}
90
91impl PausedStreamFrame {
92    /// Decode the Base64-encoded body to bytes.
93    pub fn body_bytes(&self) -> std::result::Result<Vec<u8>, base64::DecodeError> {
94        BASE64.decode(&self.body)
95    }
96}
97
98/// Client-to-server reply for a stream frame decision.
99#[derive(Debug, Clone, Serialize, Deserialize)]
100#[serde(rename_all = "camelCase")]
101pub struct StreamFrameDecision {
102    pub correlation_id: String,
103    pub action: String,
104    #[serde(skip_serializing_if = "Option::is_none")]
105    pub body: Option<String>,
106}
107
108impl StreamFrameDecision {
109    /// Create a CONTINUE decision.
110    pub fn continue_frame(correlation_id: impl Into<String>) -> Self {
111        Self {
112            correlation_id: correlation_id.into(),
113            action: "CONTINUE".to_string(),
114            body: None,
115        }
116    }
117
118    /// Create a MODIFY decision with replacement bytes.
119    pub fn modify(correlation_id: impl Into<String>, body: &[u8]) -> Self {
120        Self {
121            correlation_id: correlation_id.into(),
122            action: "MODIFY".to_string(),
123            body: Some(BASE64.encode(body)),
124        }
125    }
126
127    /// Create a DROP decision.
128    pub fn drop_frame(correlation_id: impl Into<String>) -> Self {
129        Self {
130            correlation_id: correlation_id.into(),
131            action: "DROP".to_string(),
132            body: None,
133        }
134    }
135
136    /// Create an INJECT decision.
137    pub fn inject(correlation_id: impl Into<String>, extra_body: &[u8]) -> Self {
138        Self {
139            correlation_id: correlation_id.into(),
140            action: "INJECT".to_string(),
141            body: Some(BASE64.encode(extra_body)),
142        }
143    }
144
145    /// Create a CLOSE decision.
146    pub fn close(correlation_id: impl Into<String>) -> Self {
147        Self {
148            correlation_id: correlation_id.into(),
149            action: "CLOSE".to_string(),
150            body: None,
151        }
152    }
153}
154
155// ---------------------------------------------------------------------------
156// WebSocket message envelope
157// ---------------------------------------------------------------------------
158
159#[derive(Debug, Clone, Serialize, Deserialize)]
160pub struct WsEnvelope {
161    #[serde(rename = "type")]
162    pub msg_type: String,
163    pub value: String,
164}
165
166#[derive(Debug, Clone, Deserialize)]
167#[serde(rename_all = "camelCase")]
168struct WsClientIdDTO {
169    client_id: String,
170}
171
172// ---------------------------------------------------------------------------
173// Handler types
174// ---------------------------------------------------------------------------
175
176/// Handler for REQUEST phase breakpoints.
177/// Receives the parsed request JSON. Return a Value that is either:
178/// - An HttpRequest JSON object (continue/modify), or
179/// - An HttpResponse JSON object with a "statusCode" field (abort).
180///
181/// Return None to auto-continue with the original request.
182pub type BreakpointRequestHandler = Box<dyn Fn(Value) -> Option<Value> + Send + Sync>;
183
184/// Handler for RESPONSE phase breakpoints.
185/// Receives the request and response JSON objects.
186/// Return a response Value, or None to auto-continue.
187pub type BreakpointResponseHandler = Box<dyn Fn(Value, Value) -> Option<Value> + Send + Sync>;
188
189/// Handler for stream frame breakpoints.
190/// Receives the parsed PausedStreamFrame.
191/// Return a StreamFrameDecision, or None to auto-continue.
192pub type BreakpointStreamFrameHandler =
193    Box<dyn Fn(&PausedStreamFrame) -> Option<StreamFrameDecision> + Send + Sync>;
194
195// ---------------------------------------------------------------------------
196// BreakpointWebSocketClient
197// ---------------------------------------------------------------------------
198
199/// Internal WebSocket client for breakpoint callback resolution.
200pub(crate) struct BreakpointWebSocketClient {
201    pub(crate) client_id: String,
202    socket: Arc<Mutex<tungstenite::WebSocket<tungstenite::stream::MaybeTlsStream<std::net::TcpStream>>>>,
203    request_handlers: Arc<Mutex<HashMap<String, BreakpointRequestHandler>>>,
204    response_handlers: Arc<Mutex<HashMap<String, BreakpointResponseHandler>>>,
205    stream_frame_handlers: Arc<Mutex<HashMap<String, BreakpointStreamFrameHandler>>>,
206    dead: Arc<std::sync::atomic::AtomicBool>,
207    _read_thread: Option<std::thread::JoinHandle<()>>,
208}
209
210impl BreakpointWebSocketClient {
211    /// Returns true if the read loop has exited (connection no longer usable).
212    pub(crate) fn is_dead(&self) -> bool {
213        self.dead.load(std::sync::atomic::Ordering::Relaxed)
214    }
215
216    /// Connect to the callback WebSocket and obtain a clientId.
217    pub(crate) fn connect(base_url: &str) -> Result<Self> {
218        let parsed = Url::parse(base_url)
219            .map_err(|e| Error::InvalidRequest(format!("bad base URL: {e}")))?;
220
221        let ws_scheme = if parsed.scheme() == "https" { "wss" } else { "ws" };
222        let host = parsed.host_str().unwrap_or("localhost");
223        let port = parsed.port().unwrap_or(if parsed.scheme() == "https" { 443 } else { 80 });
224        let path = parsed.path().trim_end_matches('/');
225        let ws_url = format!("{ws_scheme}://{host}:{port}{path}/_mockserver_callback_websocket");
226
227        // NOTE: tungstenite's blocking `connect` does not expose a handshake timeout.
228        // A connect timeout would require manually creating a TcpStream with
229        // `TcpStream::connect_timeout`, then upgrading via `tungstenite::client`,
230        // which adds complexity and fragility. Leaving this as a deferred follow-up;
231        // the OS-level TCP connect timeout (typically ~60-120s) applies instead.
232        let (mut socket, _response) = connect(&ws_url)
233            .map_err(|e| Error::InvalidRequest(format!("WebSocket connect failed: {e}")))?;
234
235        // Read registration reply
236        let reg_msg = socket
237            .read()
238            .map_err(|e| Error::InvalidRequest(format!("WebSocket read registration: {e}")))?;
239
240        let reg_text = match reg_msg {
241            Message::Text(t) => t,
242            _ => return Err(Error::InvalidRequest("Expected text registration message".into())),
243        };
244
245        let envelope: WsEnvelope = serde_json::from_str(&reg_text)?;
246        let client_id = if envelope.msg_type
247            == "org.mockserver.serialization.model.WebSocketClientIdDTO"
248        {
249            let dto: WsClientIdDTO = serde_json::from_str(&envelope.value)?;
250            dto.client_id
251        } else {
252            return Err(Error::InvalidRequest(format!(
253                "Unexpected registration type: {}",
254                envelope.msg_type
255            )));
256        };
257
258        let request_handlers: Arc<Mutex<HashMap<String, BreakpointRequestHandler>>> =
259            Arc::new(Mutex::new(HashMap::new()));
260        let response_handlers: Arc<Mutex<HashMap<String, BreakpointResponseHandler>>> =
261            Arc::new(Mutex::new(HashMap::new()));
262        let stream_frame_handlers: Arc<Mutex<HashMap<String, BreakpointStreamFrameHandler>>> =
263            Arc::new(Mutex::new(HashMap::new()));
264
265        let socket = Arc::new(Mutex::new(socket));
266        let dead = Arc::new(std::sync::atomic::AtomicBool::new(false));
267
268        // Start read loop in a background thread
269        let rh = Arc::clone(&request_handlers);
270        let resh = Arc::clone(&response_handlers);
271        let sfh = Arc::clone(&stream_frame_handlers);
272        let sock = Arc::clone(&socket);
273        let dead_flag = Arc::clone(&dead);
274
275        let read_thread = std::thread::spawn(move || {
276            read_loop(sock, rh, resh, sfh, dead_flag);
277        });
278
279        Ok(Self {
280            client_id,
281            socket,
282            request_handlers,
283            response_handlers,
284            stream_frame_handlers,
285            dead,
286            _read_thread: Some(read_thread),
287        })
288    }
289
290    pub(crate) fn set_request_handler(&self, breakpoint_id: &str, handler: BreakpointRequestHandler) {
291        self.request_handlers
292            .lock()
293            .unwrap()
294            .insert(breakpoint_id.to_string(), handler);
295    }
296
297    pub(crate) fn set_response_handler(&self, breakpoint_id: &str, handler: BreakpointResponseHandler) {
298        self.response_handlers
299            .lock()
300            .unwrap()
301            .insert(breakpoint_id.to_string(), handler);
302    }
303
304    pub(crate) fn set_stream_frame_handler(
305        &self,
306        breakpoint_id: &str,
307        handler: BreakpointStreamFrameHandler,
308    ) {
309        self.stream_frame_handlers
310            .lock()
311            .unwrap()
312            .insert(breakpoint_id.to_string(), handler);
313    }
314
315    pub(crate) fn remove_handlers(&self, breakpoint_id: &str) {
316        self.request_handlers.lock().unwrap().remove(breakpoint_id);
317        self.response_handlers.lock().unwrap().remove(breakpoint_id);
318        self.stream_frame_handlers
319            .lock()
320            .unwrap()
321            .remove(breakpoint_id);
322    }
323
324    pub(crate) fn clear_handlers(&self) {
325        self.request_handlers.lock().unwrap().clear();
326        self.response_handlers.lock().unwrap().clear();
327        self.stream_frame_handlers.lock().unwrap().clear();
328    }
329
330    pub(crate) fn close(&self) {
331        if let Ok(mut sock) = self.socket.lock() {
332            let _ = sock.close(None);
333        }
334    }
335}
336
337// ---------------------------------------------------------------------------
338// Read loop (runs on a background thread)
339// ---------------------------------------------------------------------------
340
341/// Drop guard that sets the `dead` flag when the read loop exits for any reason.
342struct ReadLoopDeadGuard(Arc<std::sync::atomic::AtomicBool>);
343impl Drop for ReadLoopDeadGuard {
344    fn drop(&mut self) {
345        self.0.store(true, std::sync::atomic::Ordering::Relaxed);
346    }
347}
348
349fn read_loop(
350    socket: Arc<Mutex<tungstenite::WebSocket<tungstenite::stream::MaybeTlsStream<std::net::TcpStream>>>>,
351    request_handlers: Arc<Mutex<HashMap<String, BreakpointRequestHandler>>>,
352    response_handlers: Arc<Mutex<HashMap<String, BreakpointResponseHandler>>>,
353    stream_frame_handlers: Arc<Mutex<HashMap<String, BreakpointStreamFrameHandler>>>,
354    dead: Arc<std::sync::atomic::AtomicBool>,
355) {
356    let _guard = ReadLoopDeadGuard(dead);
357    loop {
358        let msg = {
359            let mut sock = match socket.lock() {
360                Ok(s) => s,
361                Err(e) => {
362                    eprintln!("mockserver: breakpoint ws read loop terminated: mutex poisoned: {e}");
363                    return;
364                }
365            };
366            match sock.read() {
367                Ok(m) => m,
368                Err(e) => {
369                    eprintln!("mockserver: breakpoint ws read loop terminated: {e}");
370                    return;
371                }
372            }
373        };
374
375        let text = match msg {
376            Message::Text(t) => t,
377            Message::Close(_) => {
378                eprintln!("mockserver: breakpoint ws read loop terminated: received close frame");
379                return;
380            }
381            _ => continue,
382        };
383
384        let envelope: WsEnvelope = match serde_json::from_str(&text) {
385            Ok(e) => e,
386            Err(_) => continue,
387        };
388
389        match envelope.msg_type.as_str() {
390            "org.mockserver.model.HttpRequest" => {
391                handle_request(
392                    &envelope.value,
393                    &request_handlers,
394                    &socket,
395                );
396            }
397            "org.mockserver.model.HttpRequestAndHttpResponse" => {
398                handle_response(
399                    &envelope.value,
400                    &response_handlers,
401                    &socket,
402                );
403            }
404            "org.mockserver.serialization.model.PausedStreamFrameDTO" => {
405                handle_stream_frame(
406                    &envelope.value,
407                    &stream_frame_handlers,
408                    &socket,
409                );
410            }
411            "org.mockserver.serialization.model.WebSocketClientIdDTO" => {
412                // Already handled during connect
413            }
414            _ => {}
415        }
416    }
417}
418
419fn handle_request(
420    value_json: &str,
421    handlers: &Arc<Mutex<HashMap<String, BreakpointRequestHandler>>>,
422    socket: &Arc<Mutex<tungstenite::WebSocket<tungstenite::stream::MaybeTlsStream<std::net::TcpStream>>>>,
423) {
424    if let Some((type_name, result)) = route_request(value_json, handlers) {
425        send_envelope(socket, &type_name, &result);
426    }
427}
428
429fn handle_response(
430    value_json: &str,
431    handlers: &Arc<Mutex<HashMap<String, BreakpointResponseHandler>>>,
432    socket: &Arc<Mutex<tungstenite::WebSocket<tungstenite::stream::MaybeTlsStream<std::net::TcpStream>>>>,
433) {
434    if let Some((type_name, result)) = route_response(value_json, handlers) {
435        send_envelope(socket, &type_name, &result);
436    }
437}
438
439fn handle_stream_frame(
440    value_json: &str,
441    handlers: &Arc<Mutex<HashMap<String, BreakpointStreamFrameHandler>>>,
442    socket: &Arc<Mutex<tungstenite::WebSocket<tungstenite::stream::MaybeTlsStream<std::net::TcpStream>>>>,
443) {
444    if let Some((type_name, decision)) = route_stream_frame(value_json, handlers) {
445        send_envelope(socket, &type_name, &decision);
446    }
447}
448
449// ---------------------------------------------------------------------------
450// Pure routing functions (testable without a live WebSocket)
451// ---------------------------------------------------------------------------
452
453/// Route a REQUEST-phase message: dispatch to the per-breakpoint-id handler,
454/// auto-continue on missing handler or panic. Returns (type_name, reply_value).
455pub fn route_request(
456    value_json: &str,
457    handlers: &Arc<Mutex<HashMap<String, BreakpointRequestHandler>>>,
458) -> Option<(String, Value)> {
459    let request: Value = serde_json::from_str(value_json).ok()?;
460
461    let correlation_id = extract_header(&request, "WebSocketCorrelationId");
462    let breakpoint_id = extract_header(&request, "X-MockServer-BreakpointId");
463
464    let mut result: Option<Value> = None;
465    if let Some(bp_id) = &breakpoint_id {
466        if let Ok(h) = handlers.lock() {
467            if let Some(handler) = h.get(bp_id) {
468                result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
469                    handler(request.clone())
470                }))
471                .unwrap_or(None);
472            }
473        }
474    }
475
476    // Auto-continue with original request if no result
477    let mut result = result.unwrap_or_else(|| request.clone());
478
479    // Determine type: statusCode present => HttpResponse (abort)
480    let type_name = if result.get("statusCode").is_some() {
481        "org.mockserver.model.HttpResponse"
482    } else {
483        "org.mockserver.model.HttpRequest"
484    };
485
486    if let Some(corr) = &correlation_id {
487        set_header(&mut result, "WebSocketCorrelationId", corr);
488    }
489
490    Some((type_name.to_string(), result))
491}
492
493/// Route a RESPONSE-phase message: dispatch to the per-breakpoint-id handler,
494/// auto-continue on missing handler or panic. Returns (type_name, reply_value).
495pub fn route_response(
496    value_json: &str,
497    handlers: &Arc<Mutex<HashMap<String, BreakpointResponseHandler>>>,
498) -> Option<(String, Value)> {
499    let req_and_resp: Value = serde_json::from_str(value_json).ok()?;
500
501    let http_request = req_and_resp.get("httpRequest").cloned().unwrap_or(Value::Null);
502    let http_response = req_and_resp.get("httpResponse").cloned().unwrap_or(Value::Null);
503
504    let correlation_id = extract_header(&http_request, "WebSocketCorrelationId");
505    let breakpoint_id = extract_header(&http_request, "X-MockServer-BreakpointId");
506
507    let mut result: Option<Value> = None;
508    if let Some(bp_id) = &breakpoint_id {
509        if let Ok(h) = handlers.lock() {
510            if let Some(handler) = h.get(bp_id) {
511                result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
512                    handler(http_request.clone(), http_response.clone())
513                }))
514                .unwrap_or(None);
515            }
516        }
517    }
518
519    let mut result = result.unwrap_or(http_response);
520
521    if let Some(corr) = &correlation_id {
522        set_header(&mut result, "WebSocketCorrelationId", corr);
523    }
524
525    Some(("org.mockserver.model.HttpResponse".to_string(), result))
526}
527
528/// Route a stream-frame message: dispatch to the per-breakpoint-id handler,
529/// auto-continue on missing handler or panic. Returns (type_name, decision).
530pub fn route_stream_frame(
531    value_json: &str,
532    handlers: &Arc<Mutex<HashMap<String, BreakpointStreamFrameHandler>>>,
533) -> Option<(String, StreamFrameDecision)> {
534    let frame: PausedStreamFrame = serde_json::from_str(value_json).ok()?;
535
536    let mut decision: Option<StreamFrameDecision> = None;
537    if let Some(bp_id) = &frame.breakpoint_id {
538        if let Ok(h) = handlers.lock() {
539            if let Some(handler) = h.get(bp_id) {
540                decision = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
541                    handler(&frame)
542                }))
543                .unwrap_or(None);
544            }
545        }
546    }
547
548    let mut decision = decision.unwrap_or_else(|| StreamFrameDecision::continue_frame(&frame.correlation_id));
549    decision.correlation_id = frame.correlation_id;
550
551    Some((
552        "org.mockserver.serialization.model.StreamFrameDecisionDTO".to_string(),
553        decision,
554    ))
555}
556
557fn send_envelope<T: Serialize>(
558    socket: &Arc<Mutex<tungstenite::WebSocket<tungstenite::stream::MaybeTlsStream<std::net::TcpStream>>>>,
559    type_name: &str,
560    value: &T,
561) {
562    let value_json = match serde_json::to_string(value) {
563        Ok(j) => j,
564        Err(_) => return,
565    };
566    let envelope = WsEnvelope {
567        msg_type: type_name.to_string(),
568        value: value_json,
569    };
570    let env_json = match serde_json::to_string(&envelope) {
571        Ok(j) => j,
572        Err(_) => return,
573    };
574    if let Ok(mut sock) = socket.lock() {
575        let _ = sock.send(Message::Text(env_json));
576    }
577}
578
579// ---------------------------------------------------------------------------
580// Header helpers
581// ---------------------------------------------------------------------------
582
583/// Extract the first value of a header from a JSON request object.
584pub fn extract_header(obj: &Value, name: &str) -> Option<String> {
585    let headers = obj.get("headers")?.as_object()?;
586    let name_lower = name.to_lowercase();
587    for (k, v) in headers {
588        if k.to_lowercase() == name_lower {
589            if let Some(arr) = v.as_array() {
590                return arr.first()?.as_str().map(|s| s.to_string());
591            }
592            return v.as_str().map(|s| s.to_string());
593        }
594    }
595    None
596}
597
598/// Set a header value on a JSON request/response object.
599pub fn set_header(obj: &mut Value, name: &str, value: &str) {
600    if let Some(obj_map) = obj.as_object_mut() {
601        let headers = obj_map
602            .entry("headers")
603            .or_insert_with(|| Value::Object(serde_json::Map::new()));
604        if let Some(h) = headers.as_object_mut() {
605            h.insert(
606                name.to_string(),
607                serde_json::json!([value]),
608            );
609        }
610    }
611}
mockserver_client/breakpoint.rs

mockserver_client/
breakpoint.rs
