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#[allow(deprecated)]
23use pmcp::shared::{
24    OptimizedSseConfig, OptimizedSseTransport, StdioTransport, Transport, WebSocketConfig,
25    WebSocketTransport,
26};
27use std::time::Duration;
28
29/// Create a transport based on configuration
30pub fn create_transport(transport_type: &TransportType) -> Result<Box<dyn Transport>> {
31    match transport_type {
32        TransportType::Stdio => {
33            let transport = StdioTransport::new();
34            Ok(Box::new(transport))
35        }
36        TransportType::Sse => {
37            let config = OptimizedSseConfig {
38                url: "http://localhost:8080/sse".to_string(),
39                connection_timeout: Duration::from_secs(30),
40                keepalive_interval: Duration::from_secs(15),
41                max_reconnects: 5,
42                reconnect_delay: Duration::from_secs(1),
43                buffer_size: 100,
44                flush_interval: Duration::from_millis(100),
45                enable_pooling: true,
46                max_connections: 10,
47                enable_compression: false,
48            };
49            #[allow(deprecated)]
50            let transport = OptimizedSseTransport::new(config);
51            Ok(Box::new(transport))
52        }
53        TransportType::WebSocket => {
54            let url = "ws://localhost:8080/ws"
55                .parse()
56                .map_err(|e| Error::Handler(format!("Invalid WebSocket URL: {}", e)))?;
57
58            let config = WebSocketConfig {
59                url,
60                auto_reconnect: true,
61                reconnect_delay: Duration::from_secs(1),
62                max_reconnect_delay: Duration::from_secs(30),
63                max_reconnect_attempts: Some(5),
64                ping_interval: Some(Duration::from_secs(30)),
65                request_timeout: Duration::from_secs(10),
66            };
67            let transport = WebSocketTransport::new(config);
68            Ok(Box::new(transport))
69        }
70    }
71}
72
73#[cfg(test)]
74mod tests {
75    use super::*;
76
77    #[test]
78    fn test_create_stdio_transport() {
79        let transport = create_transport(&TransportType::Stdio);
80        assert!(transport.is_ok());
81        let t = transport.unwrap();
82        assert_eq!(t.transport_type(), "stdio");
83    }
84
85    #[tokio::test]
86    async fn test_create_sse_transport() {
87        let transport = create_transport(&TransportType::Sse);
88        assert!(transport.is_ok());
89    }
90
91    #[test]
92    fn test_create_websocket_transport() {
93        let transport = create_transport(&TransportType::WebSocket);
94        assert!(transport.is_ok());
95    }
96
97    // Note: SSE and WebSocket tests require server running, so they're integration tests
98}
pforge_runtime/transport.rs

pforge_runtime/
transport.rs
