pforge_runtime/
transport.rs

1//! Transport layer implementation
2//!
3//! This module provides transport creation based on configuration.
4
5use crate::{Error, Result};
6use pforge_config::TransportType;
7// `OptimizedSseTransport` is deprecated in pmcp 2.x in favour of
8// `StreamableHttpTransport`, "which bounds every peer-controlled read".
9//
10// NOT migrated here, deliberately. The two are different wire protocols with
11// unrelated configs — `OptimizedSseConfig` carries keepalive, reconnect,
12// pooling and compression knobs that `StreamableHttpTransportConfig` (url,
13// extra_headers, auth_provider, session) has no equivalent for. Swapping them
14// changes what `transport: sse` actually speaks, so it breaks every client
15// configured against a pforge SSE endpoint. That is a product decision for a
16// pforge release, not a side effect of a dependency bump. pmcp keeps the type
17// "for 2.x compatibility", so it remains available meanwhile.
18//
19// The deprecation does flag a real exposure — an unbounded peer-controlled
20// read is a DoS vector — so this should not sit indefinitely. Tracked
21// separately.
22#[cfg(feature = "sse")]
23#[allow(deprecated)]
24use pmcp::shared::{OptimizedSseConfig, OptimizedSseTransport};
25use pmcp::shared::{StdioTransport, Transport};
26#[cfg(feature = "websocket")]
27use pmcp::shared::{WebSocketConfig, WebSocketTransport};
28#[cfg(any(feature = "sse", feature = "websocket"))]
29use std::time::Duration;
30
31/// Create a transport based on configuration
32pub fn create_transport(transport_type: &TransportType) -> Result<Box<dyn Transport>> {
33    match transport_type {
34        TransportType::Stdio => {
35            let transport = StdioTransport::new();
36            Ok(Box::new(transport))
37        }
38        TransportType::Sse => create_sse_transport(),
39        TransportType::WebSocket => create_websocket_transport(),
40    }
41}
42
43// Infallible in this configuration, but the signature must match the
44// `not(feature = "sse")` variant, which is the whole point of the split.
45#[allow(clippy::unnecessary_wraps)]
46#[cfg(feature = "sse")]
47fn create_sse_transport() -> Result<Box<dyn Transport>> {
48    let config = OptimizedSseConfig {
49        url: "http://localhost:8080/sse".to_string(),
50        connection_timeout: Duration::from_secs(30),
51        keepalive_interval: Duration::from_secs(15),
52        max_reconnects: 5,
53        reconnect_delay: Duration::from_secs(1),
54        buffer_size: 100,
55        flush_interval: Duration::from_millis(100),
56        enable_pooling: true,
57        max_connections: 10,
58        enable_compression: false,
59    };
60    #[allow(deprecated)]
61    let transport = OptimizedSseTransport::new(config);
62    Ok(Box::new(transport))
63}
64
65#[cfg(not(feature = "sse"))]
66fn create_sse_transport() -> Result<Box<dyn Transport>> {
67    Err(Error::feature_disabled("sse", "transport `sse`"))
68}
69
70#[cfg(feature = "websocket")]
71fn create_websocket_transport() -> Result<Box<dyn Transport>> {
72    let url = "ws://localhost:8080/ws"
73        .parse()
74        .map_err(|e| Error::Handler(format!("Invalid WebSocket URL: {}", e)))?;
75
76    let config = WebSocketConfig {
77        url,
78        auto_reconnect: true,
79        reconnect_delay: Duration::from_secs(1),
80        max_reconnect_delay: Duration::from_secs(30),
81        max_reconnect_attempts: Some(5),
82        ping_interval: Some(Duration::from_secs(30)),
83        request_timeout: Duration::from_secs(10),
84    };
85    let transport = WebSocketTransport::new(config);
86    Ok(Box::new(transport))
87}
88
89#[cfg(not(feature = "websocket"))]
90fn create_websocket_transport() -> Result<Box<dyn Transport>> {
91    Err(Error::feature_disabled(
92        "websocket",
93        "transport `websocket`",
94    ))
95}
96
97#[cfg(test)]
98mod tests {
99    use super::*;
100
101    #[test]
102    fn test_create_stdio_transport() {
103        let transport = create_transport(&TransportType::Stdio);
104        assert!(transport.is_ok());
105        let t = transport.unwrap();
106        assert_eq!(t.transport_type(), "stdio");
107    }
108
109    // The SSE/WebSocket cases are asserted in BOTH feature configurations, because
110    // the bug being guarded here was a build that did not exist rather than a
111    // behaviour that was wrong: `--no-default-features` failed to COMPILE, so a
112    // test written only for the default build could never have caught it.
113
114    #[cfg(feature = "sse")]
115    #[tokio::test]
116    async fn test_create_sse_transport() {
117        let transport = create_transport(&TransportType::Sse);
118        assert!(transport.is_ok());
119    }
120
121    #[cfg(not(feature = "sse"))]
122    #[test]
123    fn test_sse_without_feature_errors_and_names_the_feature() {
124        let msg = create_transport(&TransportType::Sse)
125            .expect_err("sse must fail when compiled out, not fall back to stdio")
126            .to_string();
127        assert!(
128            msg.contains("sse") && msg.contains("--features"),
129            "the error must tell the operator how to fix it, got: {msg}"
130        );
131    }
132
133    #[cfg(feature = "websocket")]
134    #[test]
135    fn test_create_websocket_transport() {
136        let transport = create_transport(&TransportType::WebSocket);
137        assert!(transport.is_ok());
138    }
139
140    #[cfg(not(feature = "websocket"))]
141    #[test]
142    fn test_websocket_without_feature_errors_and_names_the_feature() {
143        let msg = create_transport(&TransportType::WebSocket)
144            .expect_err("websocket must fail when compiled out, not fall back to stdio")
145            .to_string();
146        assert!(
147            msg.contains("websocket") && msg.contains("--features"),
148            "the error must tell the operator how to fix it, got: {msg}"
149        );
150    }
151
152    // Stdio is unconditional, so it must work in EVERY configuration. If a future
153    // refactor puts stdio behind a feature, this fails in the minimal build.
154    #[test]
155    fn test_stdio_works_in_every_feature_configuration() {
156        assert!(create_transport(&TransportType::Stdio).is_ok());
157    }
158}
pforge_runtime/transport.rs

pforge_runtime/
transport.rs
