p2p_foundation/
transport.rs

1//! Transport Layer
2//!
3//! This module provides transport protocol implementations for the P2P Foundation.
4//! It supports QUIC and TCP transports with automatic selection, connection pooling,
5//! and performance monitoring.
6
7pub mod tcp;
8pub mod quic;
9pub mod tunneled;
10
11use crate::{PeerId, Multiaddr, P2PError, Result};
12use async_trait::async_trait;
13use serde::{Deserialize, Serialize};
14use std::collections::HashMap;
15use std::fmt;
16use std::net::SocketAddr;
17use std::sync::Arc;
18use std::time::{Duration, Instant};
19use tokio::sync::{Mutex, RwLock};
20use tracing::{debug, info, warn};
21
22/// Transport protocol types
23#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
24pub enum TransportType {
25    /// QUIC transport protocol
26    QUIC,
27    /// TCP transport protocol  
28    TCP,
29}
30
31/// Transport selection strategy
32#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
33pub enum TransportSelection {
34    /// Automatically select best transport
35    Auto,
36    /// Prefer specific transport with fallback
37    Prefer(TransportType),
38    /// Force specific transport only
39    Force(TransportType),
40}
41
42/// Connection quality metrics
43#[derive(Debug, Clone)]
44pub struct ConnectionQuality {
45    /// Round-trip latency
46    pub latency: Duration,
47    /// Throughput in Mbps
48    pub throughput_mbps: f64,
49    /// Packet loss percentage
50    pub packet_loss: f64,
51    /// Jitter (latency variation)
52    pub jitter: Duration,
53    /// Connection establishment time
54    pub connect_time: Duration,
55}
56
57/// Connection information
58#[derive(Debug, Clone)]
59pub struct ConnectionInfo {
60    /// Transport type being used
61    pub transport_type: TransportType,
62    /// Local address
63    pub local_addr: Multiaddr,
64    /// Remote address
65    pub remote_addr: Multiaddr,
66    /// Whether connection is encrypted
67    pub is_encrypted: bool,
68    /// Cipher suite being used
69    pub cipher_suite: String,
70    /// Whether 0-RTT was used
71    pub used_0rtt: bool,
72    /// Connection establishment time
73    pub established_at: Instant,
74    /// Last activity timestamp
75    pub last_activity: Instant,
76}
77
78/// Connection pool information
79#[derive(Debug, Clone)]
80pub struct ConnectionPoolInfo {
81    /// Number of active connections
82    pub active_connections: usize,
83    /// Total connections ever created
84    pub total_connections: usize,
85    /// Bytes sent through pool
86    pub bytes_sent: u64,
87    /// Bytes received through pool
88    pub bytes_received: u64,
89}
90
91/// Connection pool statistics
92#[derive(Debug, Clone)]
93pub struct ConnectionPoolStats {
94    /// Messages sent per connection
95    pub messages_per_connection: HashMap<String, usize>,
96    /// Bytes per connection
97    pub bytes_per_connection: HashMap<String, u64>,
98    /// Average latency per connection
99    pub latency_per_connection: HashMap<String, Duration>,
100}
101
102/// Message received from transport
103#[derive(Debug, Clone)]
104pub struct TransportMessage {
105    /// Sender peer ID
106    pub sender: PeerId,
107    /// Message data
108    pub data: Vec<u8>,
109    /// Protocol identifier
110    pub protocol: String,
111    /// Timestamp when received
112    pub received_at: Instant,
113}
114
115/// Transport trait for protocol implementations
116#[async_trait]
117pub trait Transport: Send + Sync {
118    /// Start listening on the given address
119    async fn listen(&self, addr: SocketAddr) -> Result<Vec<Multiaddr>>;
120    
121    /// Accept incoming connections (for server-side)
122    async fn accept(&self) -> Result<Box<dyn Connection>>;
123    
124    /// Connect to a remote peer
125    async fn connect(&self, addr: &Multiaddr) -> Result<Box<dyn Connection>>;
126    
127    /// Connect with specific transport options
128    async fn connect_with_options(&self, addr: &Multiaddr, options: TransportOptions) -> Result<Box<dyn Connection>>;
129    
130    /// Get supported addresses for this transport
131    fn supported_addresses(&self) -> Vec<String>;
132    
133    /// Get transport type
134    fn transport_type(&self) -> TransportType;
135    
136    /// Check if address is supported
137    fn supports_address(&self, addr: &Multiaddr) -> bool;
138}
139
140/// Connection trait for active connections
141#[async_trait]
142pub trait Connection: Send + Sync {
143    /// Send data over the connection
144    async fn send(&mut self, data: &[u8]) -> Result<()>;
145    
146    /// Receive data from the connection
147    async fn receive(&mut self) -> Result<Vec<u8>>;
148    
149    /// Get connection info
150    async fn info(&self) -> ConnectionInfo;
151    
152    /// Close the connection
153    async fn close(&mut self) -> Result<()>;
154    
155    /// Check if connection is alive
156    async fn is_alive(&self) -> bool;
157    
158    /// Measure connection quality
159    async fn measure_quality(&self) -> Result<ConnectionQuality>;
160    
161    /// Get local address
162    fn local_addr(&self) -> Multiaddr;
163    
164    /// Get remote address
165    fn remote_addr(&self) -> Multiaddr;
166}
167
168/// Transport configuration options
169#[derive(Debug, Clone)]
170pub struct TransportOptions {
171    /// Enable 0-RTT for QUIC
172    pub enable_0rtt: bool,
173    /// Force encryption
174    pub require_encryption: bool,
175    /// Connection timeout
176    pub connect_timeout: Duration,
177    /// Keep-alive interval
178    pub keep_alive: Duration,
179    /// Maximum message size
180    pub max_message_size: usize,
181}
182
183/// Transport manager coordinates different transport protocols
184pub struct TransportManager {
185    /// Available transports
186    transports: HashMap<TransportType, Arc<dyn Transport>>,
187    /// Active connections
188    connections: Arc<RwLock<HashMap<PeerId, Arc<Mutex<ConnectionPool>>>>>,
189    /// Transport selection strategy
190    selection: TransportSelection,
191    /// Configuration options
192    options: TransportOptions,
193}
194
195/// Connection pool for a specific peer
196struct ConnectionPool {
197    /// Active connections
198    connections: Vec<Arc<Mutex<Box<dyn Connection>>>>,
199    /// Connection info cache (reserved for future use)
200    _info_cache: HashMap<String, ConnectionInfo>,
201    /// Pool statistics
202    stats: ConnectionPoolStats,
203    /// Pool configuration
204    max_connections: usize,
205    /// Round-robin index for load balancing
206    round_robin_index: usize,
207}
208
209impl TransportManager {
210    /// Create a new transport manager
211    pub fn new(selection: TransportSelection, options: TransportOptions) -> Self {
212        Self {
213            transports: HashMap::new(),
214            connections: Arc::new(RwLock::new(HashMap::new())),
215            selection,
216            options,
217        }
218    }
219    
220    /// Register a transport implementation
221    pub fn register_transport(&mut self, transport: Arc<dyn Transport>) {
222        let transport_type = transport.transport_type();
223        self.transports.insert(transport_type, transport);
224        info!("Registered transport: {:?}", transport_type);
225    }
226    
227    /// Connect to a peer using the best available transport
228    pub async fn connect(&self, addr: &Multiaddr) -> Result<PeerId> {
229        let transport_type = self.select_transport(addr).await?;
230        let transport = self.transports.get(&transport_type)
231            .ok_or_else(|| P2PError::Transport(format!("Transport {:?} not available", transport_type)))?;
232        
233        debug!("Connecting to {} using {:?}", addr, transport_type);
234        
235        let connection = transport.connect_with_options(addr, self.options.clone()).await?;
236        let peer_id = format!("peer_from_{}", addr); // Placeholder peer ID extraction
237        
238        // Add to connection pool
239        self.add_connection(peer_id.clone(), connection).await?;
240        
241        info!("Connected to peer {} via {:?}", peer_id, transport_type);
242        Ok(peer_id)
243    }
244    
245    /// Connect with specific transport
246    pub async fn connect_with_transport(&self, addr: &Multiaddr, transport_type: TransportType) -> Result<PeerId> {
247        let transport = self.transports.get(&transport_type)
248            .ok_or_else(|| P2PError::Transport(format!("Transport {:?} not available", transport_type)))?;
249        
250        let connection = transport.connect_with_options(addr, self.options.clone()).await?;
251        let peer_id = format!("peer_from_{}", addr);
252        
253        self.add_connection(peer_id.clone(), connection).await?;
254        Ok(peer_id)
255    }
256    
257    /// Send message to a peer
258    pub async fn send_message(&self, peer_id: &PeerId, data: Vec<u8>) -> Result<()> {
259        let connections = self.connections.read().await;
260        let pool = connections.get(peer_id)
261            .ok_or_else(|| P2PError::Network(format!("No connection to peer {}", peer_id)))?;
262        
263        let mut pool_guard = pool.lock().await;
264        let connection = pool_guard.get_connection()?;
265        
266        let mut conn_guard = connection.lock().await;
267        conn_guard.send(&data).await?;
268        
269        debug!("Sent {} bytes to peer {}", data.len(), peer_id);
270        Ok(())
271    }
272    
273    /// Get connection info for a peer
274    pub async fn get_connection_info(&self, peer_id: &PeerId) -> Result<ConnectionInfo> {
275        let connections = self.connections.read().await;
276        let pool = connections.get(peer_id)
277            .ok_or_else(|| P2PError::Network(format!("No connection to peer {}", peer_id)))?;
278        
279        let mut pool_guard = pool.lock().await;
280        let connection = pool_guard.get_connection()?;
281        let conn_guard = connection.lock().await;
282        
283        Ok(conn_guard.info().await)
284    }
285    
286    /// Get connection pool info
287    pub async fn get_connection_pool_info(&self, peer_id: &PeerId) -> Result<ConnectionPoolInfo> {
288        let connections = self.connections.read().await;
289        let pool = connections.get(peer_id)
290            .ok_or_else(|| P2PError::Network(format!("No connection to peer {}", peer_id)))?;
291        
292        let pool_guard = pool.lock().await;
293        Ok(ConnectionPoolInfo {
294            active_connections: pool_guard.connections.len(),
295            total_connections: pool_guard.stats.messages_per_connection.len(),
296            bytes_sent: pool_guard.stats.bytes_per_connection.values().sum(),
297            bytes_received: 0, // TODO: Track separately
298        })
299    }
300    
301    /// Get connection pool statistics
302    pub async fn get_connection_pool_stats(&self, peer_id: &PeerId) -> Result<ConnectionPoolStats> {
303        let connections = self.connections.read().await;
304        let pool = connections.get(peer_id)
305            .ok_or_else(|| P2PError::Network(format!("No connection to peer {}", peer_id)))?;
306        
307        let pool_guard = pool.lock().await;
308        Ok(pool_guard.stats.clone())
309    }
310    
311    /// Measure connection quality
312    pub async fn measure_connection_quality(&self, peer_id: &PeerId) -> Result<ConnectionQuality> {
313        let connections = self.connections.read().await;
314        let pool = connections.get(peer_id)
315            .ok_or_else(|| P2PError::Network(format!("No connection to peer {}", peer_id)))?;
316        
317        let mut pool_guard = pool.lock().await;
318        let connection = pool_guard.get_connection()?;
319        let conn_guard = connection.lock().await;
320        
321        conn_guard.measure_quality().await
322    }
323    
324    /// Switch transport for a peer
325    pub async fn switch_transport(&self, peer_id: &PeerId, _new_transport: TransportType) -> Result<()> {
326        // This is a placeholder implementation
327        // In reality, this would establish a new connection with the new transport
328        // and gracefully migrate the existing connection
329        
330        warn!("Transport switching not yet fully implemented for peer {}", peer_id);
331        Ok(())
332    }
333    
334    /// Select best transport for an address
335    async fn select_transport(&self, addr: &Multiaddr) -> Result<TransportType> {
336        match &self.selection {
337            TransportSelection::Force(transport_type) => {
338                if self.transports.contains_key(transport_type) {
339                    Ok(*transport_type)
340                } else {
341                    Err(P2PError::Transport(format!("Forced transport {:?} not available", transport_type)))
342                }
343            }
344            TransportSelection::Prefer(preferred) => {
345                if self.transports.contains_key(preferred) {
346                    Ok(*preferred)
347                } else {
348                    // Fall back to any available transport
349                    self.auto_select_transport(addr).await
350                }
351            }
352            TransportSelection::Auto => {
353                self.auto_select_transport(addr).await
354            }
355        }
356    }
357    
358    /// Auto-select best transport based on address and conditions
359    async fn auto_select_transport(&self, addr: &Multiaddr) -> Result<TransportType> {
360        // Strongly prefer QUIC if available (better performance, 0-RTT, multiplexing)
361        if self.transports.contains_key(&TransportType::QUIC) {
362            if let Some(transport) = self.transports.get(&TransportType::QUIC) {
363                if transport.supports_address(addr) {
364                    debug!("Selected QUIC transport for {} (preferred for P2P)", addr);
365                    return Ok(TransportType::QUIC);
366                }
367            }
368        }
369        
370        // Fall back to TCP only as last resort
371        if self.transports.contains_key(&TransportType::TCP) {
372            if let Some(transport) = self.transports.get(&TransportType::TCP) {
373                if transport.supports_address(addr) {
374                    warn!("Falling back to TCP transport for {}. QUIC would provide better performance.", addr);
375                    return Ok(TransportType::TCP);
376                }
377            }
378        }
379        
380        Err(P2PError::Transport("No suitable transport available. Consider using QUIC-compatible addresses.".to_string()))
381    }
382    
383    /// Add connection to pool
384    async fn add_connection(&self, peer_id: PeerId, connection: Box<dyn Connection>) -> Result<()> {
385        let mut connections = self.connections.write().await;
386        
387        let pool = connections.entry(peer_id.clone()).or_insert_with(|| {
388            Arc::new(Mutex::new(ConnectionPool::new(3))) // Default max 3 connections per peer
389        });
390        
391        let mut pool_guard = pool.lock().await;
392        pool_guard.add_connection(connection).await?;
393        
394        Ok(())
395    }
396}
397
398impl ConnectionPool {
399    /// Create a new connection pool
400    fn new(max_connections: usize) -> Self {
401        Self {
402            connections: Vec::new(),
403            _info_cache: HashMap::new(),
404            stats: ConnectionPoolStats {
405                messages_per_connection: HashMap::new(),
406                bytes_per_connection: HashMap::new(),
407                latency_per_connection: HashMap::new(),
408            },
409            max_connections,
410            round_robin_index: 0,
411        }
412    }
413    
414    /// Add a connection to the pool
415    async fn add_connection(&mut self, connection: Box<dyn Connection>) -> Result<()> {
416        if self.connections.len() >= self.max_connections {
417            // Remove oldest connection
418            self.connections.remove(0);
419        }
420        
421        let conn_id = format!("conn_{}", self.connections.len());
422        self.stats.messages_per_connection.insert(conn_id.clone(), 0);
423        self.stats.bytes_per_connection.insert(conn_id.clone(), 0);
424        self.stats.latency_per_connection.insert(conn_id, Duration::from_millis(0));
425        
426        self.connections.push(Arc::new(Mutex::new(connection)));
427        Ok(())
428    }
429    
430    /// Get a connection using round-robin load balancing
431    fn get_connection(&mut self) -> Result<Arc<Mutex<Box<dyn Connection>>>> {
432        if self.connections.is_empty() {
433            return Err(P2PError::Network("No connections available".to_string()));
434        }
435        
436        let connection = self.connections[self.round_robin_index % self.connections.len()].clone();
437        self.round_robin_index += 1;
438        
439        // Update stats
440        let conn_id = format!("conn_{}", self.round_robin_index % self.connections.len());
441        if let Some(count) = self.stats.messages_per_connection.get_mut(&conn_id) {
442            *count += 1;
443        }
444        
445        Ok(connection)
446    }
447}
448
449impl fmt::Display for TransportType {
450    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
451        match self {
452            TransportType::QUIC => write!(f, "quic"),
453            TransportType::TCP => write!(f, "tcp"),
454        }
455    }
456}
457
458impl Default for TransportSelection {
459    fn default() -> Self {
460        // Default to preferring QUIC with TCP fallback
461        TransportSelection::Prefer(TransportType::QUIC)
462    }
463}
464
465impl Default for TransportOptions {
466    fn default() -> Self {
467        Self {
468            enable_0rtt: true,
469            require_encryption: true,
470            connect_timeout: Duration::from_secs(30),
471            keep_alive: Duration::from_secs(60),
472            max_message_size: 64 * 1024 * 1024, // 64MB
473        }
474    }
475}
476
477impl Default for ConnectionQuality {
478    fn default() -> Self {
479        Self {
480            latency: Duration::from_millis(50),
481            throughput_mbps: 100.0,
482            packet_loss: 0.0,
483            jitter: Duration::from_millis(5),
484            connect_time: Duration::from_millis(100),
485        }
486    }
487}
488
489
490/// Legacy transport types module for backward compatibility
491pub mod transport_types {
492    pub use super::TransportType;
493}
494
495// Re-export transport implementations
496pub use tcp::TcpTransport;
497pub use quic::QuicTransport;
498
499#[cfg(test)]
500mod tests {
501    use super::*;
502    use async_trait::async_trait;
503    use std::sync::atomic::{AtomicUsize, Ordering};
504    use tokio::time::Duration;
505
506    /// Mock transport implementation for testing
507    struct MockTransport {
508        transport_type: TransportType,
509        should_fail: bool,
510        supports_all: bool,
511    }
512
513    impl MockTransport {
514        fn new(transport_type: TransportType) -> Self {
515            Self {
516                transport_type,
517                should_fail: false,
518                supports_all: true,
519            }
520        }
521
522        fn with_failure(mut self) -> Self {
523            self.should_fail = true;
524            self
525        }
526
527        fn with_limited_support(mut self) -> Self {
528            self.supports_all = false;
529            self
530        }
531    }
532
533    #[async_trait]
534    impl Transport for MockTransport {
535        async fn listen(&self, _addr: SocketAddr) -> Result<Vec<Multiaddr>> {
536            if self.should_fail {
537                return Err(P2PError::Transport("Listen failed".to_string()));
538            }
539            Ok(vec!["/ip4/127.0.0.1/tcp/9000".to_string()])
540        }
541
542        async fn connect(&self, addr: &Multiaddr) -> Result<Box<dyn Connection>> {
543            if self.should_fail {
544                return Err(P2PError::Transport("Connection failed".to_string()));
545            }
546            Ok(Box::new(MockConnection::new(addr.clone())))
547        }
548
549        async fn connect_with_options(&self, addr: &Multiaddr, _options: TransportOptions) -> Result<Box<dyn Connection>> {
550            self.connect(addr).await
551        }
552
553        async fn accept(&self) -> Result<Box<dyn Connection>> {
554            if self.should_fail {
555                return Err(P2PError::Transport("Accept failed".to_string()));
556            }
557            Ok(Box::new(MockConnection::new("/ip4/127.0.0.1/tcp/9000".to_string())))
558        }
559
560        fn supported_addresses(&self) -> Vec<String> {
561            if self.supports_all {
562                vec!["/ip4/0.0.0.0/tcp/0".to_string(), "/ip6/::/tcp/0".to_string()]
563            } else {
564                vec!["/ip4/0.0.0.0/tcp/0".to_string()]
565            }
566        }
567
568        fn transport_type(&self) -> TransportType {
569            self.transport_type
570        }
571
572        fn supports_address(&self, addr: &Multiaddr) -> bool {
573            if !self.supports_all && addr.contains("ip6") {
574                return false;
575            }
576            addr.contains("tcp") || addr.contains("quic")
577        }
578    }
579
580    /// Mock connection implementation for testing
581    struct MockConnection {
582        remote_addr: Multiaddr,
583        is_alive: bool,
584        bytes_sent: AtomicUsize,
585        bytes_received: AtomicUsize,
586    }
587
588    impl MockConnection {
589        fn new(remote_addr: Multiaddr) -> Self {
590            Self {
591                remote_addr,
592                is_alive: true,
593                bytes_sent: AtomicUsize::new(0),
594                bytes_received: AtomicUsize::new(0),
595            }
596        }
597    }
598
599    #[async_trait]
600    impl Connection for MockConnection {
601        async fn send(&mut self, data: &[u8]) -> Result<()> {
602            if !self.is_alive {
603                return Err(P2PError::Network("Connection closed".to_string()));
604            }
605            self.bytes_sent.fetch_add(data.len(), Ordering::Relaxed);
606            Ok(())
607        }
608
609        async fn receive(&mut self) -> Result<Vec<u8>> {
610            if !self.is_alive {
611                return Err(P2PError::Network("Connection closed".to_string()));
612            }
613            let data = b"mock_response".to_vec();
614            self.bytes_received.fetch_add(data.len(), Ordering::Relaxed);
615            Ok(data)
616        }
617
618        async fn info(&self) -> ConnectionInfo {
619            ConnectionInfo {
620                transport_type: TransportType::QUIC,
621                local_addr: "/ip4/127.0.0.1/tcp/9000".to_string(),
622                remote_addr: self.remote_addr.clone(),
623                is_encrypted: true,
624                cipher_suite: "TLS_AES_256_GCM_SHA384".to_string(),
625                used_0rtt: false,
626                established_at: Instant::now(),
627                last_activity: Instant::now(),
628            }
629        }
630
631        async fn close(&mut self) -> Result<()> {
632            self.is_alive = false;
633            Ok(())
634        }
635
636        async fn is_alive(&self) -> bool {
637            self.is_alive
638        }
639
640        async fn measure_quality(&self) -> Result<ConnectionQuality> {
641            Ok(ConnectionQuality {
642                latency: Duration::from_millis(10),
643                throughput_mbps: 1000.0,
644                packet_loss: 0.1,
645                jitter: Duration::from_millis(2),
646                connect_time: Duration::from_millis(50),
647            })
648        }
649
650        fn local_addr(&self) -> Multiaddr {
651            "/ip4/127.0.0.1/tcp/9000".to_string()
652        }
653
654        fn remote_addr(&self) -> Multiaddr {
655            self.remote_addr.clone()
656        }
657    }
658
659    fn create_test_transport_manager() -> TransportManager {
660        let options = TransportOptions::default();
661        TransportManager::new(TransportSelection::Auto, options)
662    }
663
664    #[test]
665    fn test_transport_type_display() {
666        assert_eq!(format!("{}", TransportType::QUIC), "quic");
667        assert_eq!(format!("{}", TransportType::TCP), "tcp");
668    }
669
670    #[test]
671    fn test_transport_type_serialization() {
672        let quic_type = TransportType::QUIC;
673        let tcp_type = TransportType::TCP;
674
675        assert_eq!(quic_type, TransportType::QUIC);
676        assert_eq!(tcp_type, TransportType::TCP);
677        assert_ne!(quic_type, tcp_type);
678    }
679
680    #[test]
681    fn test_transport_selection_variants() {
682        let auto = TransportSelection::Auto;
683        let prefer_quic = TransportSelection::Prefer(TransportType::QUIC);
684        let force_tcp = TransportSelection::Force(TransportType::TCP);
685
686        assert!(matches!(auto, TransportSelection::Auto));
687        assert!(matches!(prefer_quic, TransportSelection::Prefer(TransportType::QUIC)));
688        assert!(matches!(force_tcp, TransportSelection::Force(TransportType::TCP)));
689    }
690
691    #[test]
692    fn test_transport_selection_default() {
693        let default = TransportSelection::default();
694        assert!(matches!(default, TransportSelection::Prefer(TransportType::QUIC)));
695    }
696
697    #[test]
698    fn test_transport_options_default() {
699        let options = TransportOptions::default();
700        
701        assert!(options.enable_0rtt);
702        assert!(options.require_encryption);
703        assert_eq!(options.connect_timeout, Duration::from_secs(30));
704        assert_eq!(options.keep_alive, Duration::from_secs(60));
705        assert_eq!(options.max_message_size, 64 * 1024 * 1024);
706    }
707
708    #[test]
709    fn test_connection_quality_default() {
710        let quality = ConnectionQuality::default();
711        
712        assert_eq!(quality.latency, Duration::from_millis(50));
713        assert_eq!(quality.throughput_mbps, 100.0);
714        assert_eq!(quality.packet_loss, 0.0);
715        assert_eq!(quality.jitter, Duration::from_millis(5));
716        assert_eq!(quality.connect_time, Duration::from_millis(100));
717    }
718
719    #[tokio::test]
720    async fn test_transport_manager_creation() {
721        let manager = create_test_transport_manager();
722        assert!(manager.transports.is_empty());
723    }
724
725    #[tokio::test]
726    async fn test_transport_registration() {
727        let mut manager = create_test_transport_manager();
728        let quic_transport = Arc::new(MockTransport::new(TransportType::QUIC));
729        let tcp_transport = Arc::new(MockTransport::new(TransportType::TCP));
730
731        manager.register_transport(quic_transport.clone());
732        manager.register_transport(tcp_transport.clone());
733
734        assert_eq!(manager.transports.len(), 2);
735        assert!(manager.transports.contains_key(&TransportType::QUIC));
736        assert!(manager.transports.contains_key(&TransportType::TCP));
737    }
738
739    #[tokio::test]
740    async fn test_connection_establishment() -> Result<()> {
741        let mut manager = create_test_transport_manager();
742        let transport = Arc::new(MockTransport::new(TransportType::QUIC));
743        manager.register_transport(transport);
744
745        let peer_id = manager.connect(&"/ip4/127.0.0.1/tcp/9001".to_string()).await?;
746        assert_eq!(peer_id, "peer_from_/ip4/127.0.0.1/tcp/9001");
747
748        let connections = manager.connections.read().await;
749        assert!(connections.contains_key(&peer_id));
750
751        Ok(())
752    }
753
754    #[tokio::test]
755    async fn test_connection_with_specific_transport() -> Result<()> {
756        let mut manager = create_test_transport_manager();
757        let transport = Arc::new(MockTransport::new(TransportType::TCP));
758        manager.register_transport(transport);
759
760        let peer_id = manager.connect_with_transport(
761            &"/ip4/127.0.0.1/tcp/9002".to_string(),
762            TransportType::TCP
763        ).await?;
764
765        assert_eq!(peer_id, "peer_from_/ip4/127.0.0.1/tcp/9002");
766        Ok(())
767    }
768
769    #[tokio::test]
770    async fn test_connection_failure_handling() {
771        let mut manager = create_test_transport_manager();
772        let failing_transport = Arc::new(MockTransport::new(TransportType::QUIC).with_failure());
773        manager.register_transport(failing_transport);
774
775        let result = manager.connect(&"/ip4/127.0.0.1/tcp/9003".to_string()).await;
776        assert!(result.is_err());
777        assert!(result.unwrap_err().to_string().contains("Connection failed"));
778    }
779
780    #[tokio::test]
781    async fn test_message_sending() -> Result<()> {
782        let mut manager = create_test_transport_manager();
783        let transport = Arc::new(MockTransport::new(TransportType::QUIC));
784        manager.register_transport(transport);
785
786        let peer_id = manager.connect(&"/ip4/127.0.0.1/tcp/9004".to_string()).await?;
787        let message = b"Hello, transport!".to_vec();
788        
789        manager.send_message(&peer_id, message.clone()).await?;
790
791        // Verify message was processed
792        let pool_info = manager.get_connection_pool_info(&peer_id).await?;
793        assert_eq!(pool_info.active_connections, 1);
794
795        Ok(())
796    }
797
798    #[tokio::test]
799    async fn test_message_sending_no_connection() {
800        let manager = create_test_transport_manager();
801        let result = manager.send_message(&"nonexistent_peer".to_string(), vec![1, 2, 3]).await;
802        
803        assert!(result.is_err());
804        assert!(result.unwrap_err().to_string().contains("No connection to peer"));
805    }
806
807    #[tokio::test]
808    async fn test_connection_info_retrieval() -> Result<()> {
809        let mut manager = create_test_transport_manager();
810        let transport = Arc::new(MockTransport::new(TransportType::QUIC));
811        manager.register_transport(transport);
812
813        let peer_id = manager.connect(&"/ip4/127.0.0.1/tcp/9005".to_string()).await?;
814        let info = manager.get_connection_info(&peer_id).await?;
815
816        assert_eq!(info.transport_type, TransportType::QUIC);
817        assert_eq!(info.remote_addr, "/ip4/127.0.0.1/tcp/9005");
818        assert!(info.is_encrypted);
819        assert_eq!(info.cipher_suite, "TLS_AES_256_GCM_SHA384");
820
821        Ok(())
822    }
823
824    #[tokio::test]
825    async fn test_connection_pool_info() -> Result<()> {
826        let mut manager = create_test_transport_manager();
827        let transport = Arc::new(MockTransport::new(TransportType::QUIC));
828        manager.register_transport(transport);
829
830        let peer_id = manager.connect(&"/ip4/127.0.0.1/tcp/9006".to_string()).await?;
831        let pool_info = manager.get_connection_pool_info(&peer_id).await?;
832
833        assert_eq!(pool_info.active_connections, 1);
834        assert_eq!(pool_info.total_connections, 1);
835        assert_eq!(pool_info.bytes_sent, 0);
836
837        Ok(())
838    }
839
840    #[tokio::test]
841    async fn test_connection_pool_stats() -> Result<()> {
842        let mut manager = create_test_transport_manager();
843        let transport = Arc::new(MockTransport::new(TransportType::QUIC));
844        manager.register_transport(transport);
845
846        let peer_id = manager.connect(&"/ip4/127.0.0.1/tcp/9007".to_string()).await?;
847        let stats = manager.get_connection_pool_stats(&peer_id).await?;
848
849        assert_eq!(stats.messages_per_connection.len(), 1);
850        assert_eq!(stats.bytes_per_connection.len(), 1);
851        assert_eq!(stats.latency_per_connection.len(), 1);
852
853        Ok(())
854    }
855
856    #[tokio::test]
857    async fn test_connection_quality_measurement() -> Result<()> {
858        let mut manager = create_test_transport_manager();
859        let transport = Arc::new(MockTransport::new(TransportType::QUIC));
860        manager.register_transport(transport);
861
862        let peer_id = manager.connect(&"/ip4/127.0.0.1/tcp/9008".to_string()).await?;
863        let quality = manager.measure_connection_quality(&peer_id).await?;
864
865        assert_eq!(quality.latency, Duration::from_millis(10));
866        assert_eq!(quality.throughput_mbps, 1000.0);
867        assert_eq!(quality.packet_loss, 0.1);
868        assert_eq!(quality.jitter, Duration::from_millis(2));
869
870        Ok(())
871    }
872
873    #[tokio::test]
874    async fn test_transport_switching() -> Result<()> {
875        let mut manager = create_test_transport_manager();
876        let transport = Arc::new(MockTransport::new(TransportType::QUIC));
877        manager.register_transport(transport);
878
879        let peer_id = manager.connect(&"/ip4/127.0.0.1/tcp/9009".to_string()).await?;
880        
881        // Transport switching is not fully implemented, but should not error
882        let result = manager.switch_transport(&peer_id, TransportType::TCP).await;
883        assert!(result.is_ok());
884
885        Ok(())
886    }
887
888    #[tokio::test]
889    async fn test_auto_transport_selection_prefer_quic() -> Result<()> {
890        let mut manager = create_test_transport_manager();
891        let quic_transport = Arc::new(MockTransport::new(TransportType::QUIC));
892        let tcp_transport = Arc::new(MockTransport::new(TransportType::TCP));
893        
894        manager.register_transport(quic_transport);
895        manager.register_transport(tcp_transport);
896
897        let addr = "/ip4/127.0.0.1/tcp/9010".to_string();
898        let selected = manager.auto_select_transport(&addr).await?;
899        
900        // Should prefer QUIC when available
901        assert_eq!(selected, TransportType::QUIC);
902
903        Ok(())
904    }
905
906    #[tokio::test]
907    async fn test_transport_selection_fallback_to_tcp() -> Result<()> {
908        let mut manager = create_test_transport_manager();
909        let tcp_transport = Arc::new(MockTransport::new(TransportType::TCP));
910        
911        manager.register_transport(tcp_transport);
912
913        let addr = "/ip4/127.0.0.1/tcp/9011".to_string();
914        let selected = manager.auto_select_transport(&addr).await?;
915        
916        // Should fall back to TCP when QUIC not available
917        assert_eq!(selected, TransportType::TCP);
918
919        Ok(())
920    }
921
922    #[tokio::test]
923    async fn test_transport_selection_no_suitable_transport() {
924        let manager = create_test_transport_manager();
925        let addr = "/ip4/127.0.0.1/tcp/9012".to_string();
926        
927        let result = manager.auto_select_transport(&addr).await;
928        assert!(result.is_err());
929        assert!(result.unwrap_err().to_string().contains("No suitable transport available"));
930    }
931
932    #[tokio::test]
933    async fn test_forced_transport_selection() -> Result<()> {
934        let mut manager = TransportManager::new(
935            TransportSelection::Force(TransportType::TCP),
936            TransportOptions::default()
937        );
938        let tcp_transport = Arc::new(MockTransport::new(TransportType::TCP));
939        
940        manager.register_transport(tcp_transport);
941
942        let addr = "/ip4/127.0.0.1/tcp/9013".to_string();
943        let selected = manager.select_transport(&addr).await?;
944        
945        assert_eq!(selected, TransportType::TCP);
946
947        Ok(())
948    }
949
950    #[tokio::test]
951    async fn test_forced_transport_unavailable() {
952        let manager = TransportManager::new(
953            TransportSelection::Force(TransportType::QUIC),
954            TransportOptions::default()
955        );
956
957        let addr = "/ip4/127.0.0.1/tcp/9014".to_string();
958        let result = manager.select_transport(&addr).await;
959        
960        assert!(result.is_err());
961        assert!(result.unwrap_err().to_string().contains("Forced transport QUIC not available"));
962    }
963
964    #[tokio::test]
965    async fn test_preferred_transport_with_fallback() -> Result<()> {
966        let mut manager = TransportManager::new(
967            TransportSelection::Prefer(TransportType::QUIC),
968            TransportOptions::default()
969        );
970        let tcp_transport = Arc::new(MockTransport::new(TransportType::TCP));
971        
972        manager.register_transport(tcp_transport);
973
974        let addr = "/ip4/127.0.0.1/tcp/9015".to_string();
975        let selected = manager.select_transport(&addr).await?;
976        
977        // Should fall back to TCP when preferred QUIC is not available
978        assert_eq!(selected, TransportType::TCP);
979
980        Ok(())
981    }
982
983    #[tokio::test]
984    async fn test_mock_connection_lifecycle() -> Result<()> {
985        let mut conn = MockConnection::new("/ip4/127.0.0.1/tcp/9016".to_string());
986
987        assert!(conn.is_alive().await);
988
989        // Test sending
990        conn.send(b"test message").await?;
991        assert_eq!(conn.bytes_sent.load(Ordering::Relaxed), 12);
992
993        // Test receiving
994        let received = conn.receive().await?;
995        assert_eq!(received, b"mock_response");
996        assert_eq!(conn.bytes_received.load(Ordering::Relaxed), 13);
997
998        // Test connection info
999        let info = conn.info().await;
1000        assert_eq!(info.transport_type, TransportType::QUIC);
1001        assert!(info.is_encrypted);
1002
1003        // Test quality measurement
1004        let quality = conn.measure_quality().await?;
1005        assert_eq!(quality.latency, Duration::from_millis(10));
1006
1007        // Test close
1008        conn.close().await?;
1009        assert!(!conn.is_alive().await);
1010
1011        // Operations should fail after close
1012        let result = conn.send(b"test").await;
1013        assert!(result.is_err());
1014
1015        Ok(())
1016    }
1017
1018    #[tokio::test]
1019    async fn test_connection_pool_max_connections() -> Result<()> {
1020        let mut pool = ConnectionPool::new(2); // Max 2 connections
1021
1022        // Add first connection
1023        let conn1 = Box::new(MockConnection::new("/ip4/127.0.0.1/tcp/9017".to_string()));
1024        pool.add_connection(conn1).await?;
1025        assert_eq!(pool.connections.len(), 1);
1026
1027        // Add second connection
1028        let conn2 = Box::new(MockConnection::new("/ip4/127.0.0.1/tcp/9018".to_string()));
1029        pool.add_connection(conn2).await?;
1030        assert_eq!(pool.connections.len(), 2);
1031
1032        // Add third connection (should remove first)
1033        let conn3 = Box::new(MockConnection::new("/ip4/127.0.0.1/tcp/9019".to_string()));
1034        pool.add_connection(conn3).await?;
1035        assert_eq!(pool.connections.len(), 2);
1036
1037        Ok(())
1038    }
1039
1040    #[tokio::test]
1041    async fn test_connection_pool_round_robin() -> Result<()> {
1042        let mut pool = ConnectionPool::new(3);
1043
1044        // Add connections
1045        for i in 0..3 {
1046            let conn = Box::new(MockConnection::new(format!("/ip4/127.0.0.1/tcp/{}", 9020 + i)));
1047            pool.add_connection(conn).await?;
1048        }
1049
1050        // Test round-robin selection
1051        let conn1 = pool.get_connection()?;
1052        let conn2 = pool.get_connection()?;
1053        let conn3 = pool.get_connection()?;
1054        let conn4 = pool.get_connection()?; // Should wrap around
1055
1056        // All connections should be different (until wraparound)
1057        assert_ne!(Arc::as_ptr(&conn1), Arc::as_ptr(&conn2));
1058        assert_ne!(Arc::as_ptr(&conn2), Arc::as_ptr(&conn3));
1059        // Fourth should be same as first (round-robin)
1060        assert_eq!(Arc::as_ptr(&conn1), Arc::as_ptr(&conn4));
1061
1062        Ok(())
1063    }
1064
1065    #[tokio::test]
1066    async fn test_connection_pool_empty() {
1067        let mut pool = ConnectionPool::new(3);
1068        let result = pool.get_connection();
1069        
1070        assert!(result.is_err());
1071        if let Err(e) = result {
1072            assert!(e.to_string().contains("No connections available"));
1073        }
1074    }
1075
1076    #[tokio::test]
1077    async fn test_transport_message_structure() {
1078        let message = TransportMessage {
1079            sender: "test_peer".to_string(),
1080            data: vec![1, 2, 3, 4],
1081            protocol: "/p2p/test/1.0.0".to_string(),
1082            received_at: Instant::now(),
1083        };
1084
1085        assert_eq!(message.sender, "test_peer");
1086        assert_eq!(message.data, vec![1, 2, 3, 4]);
1087        assert_eq!(message.protocol, "/p2p/test/1.0.0");
1088    }
1089
1090    #[tokio::test]
1091    async fn test_mock_transport_address_support() {
1092        let transport = MockTransport::new(TransportType::QUIC);
1093        
1094        assert!(transport.supports_address(&"/ip4/127.0.0.1/tcp/9000".to_string()));
1095        assert!(transport.supports_address(&"/ip6/::1/tcp/9000".to_string()));
1096        assert!(!transport.supports_address(&"/ip4/127.0.0.1/udp/9000".to_string()));
1097
1098        let limited_transport = MockTransport::new(TransportType::QUIC).with_limited_support();
1099        assert!(limited_transport.supports_address(&"/ip4/127.0.0.1/tcp/9000".to_string()));
1100        assert!(!limited_transport.supports_address(&"/ip6/::1/tcp/9000".to_string()));
1101    }
1102
1103    #[tokio::test]
1104    async fn test_mock_transport_supported_addresses() {
1105        let transport = MockTransport::new(TransportType::QUIC);
1106        let addresses = transport.supported_addresses();
1107        
1108        assert_eq!(addresses.len(), 2);
1109        assert!(addresses.contains(&"/ip4/0.0.0.0/tcp/0".to_string()));
1110        assert!(addresses.contains(&"/ip6/::/tcp/0".to_string()));
1111
1112        let limited_transport = MockTransport::new(TransportType::QUIC).with_limited_support();
1113        let limited_addresses = limited_transport.supported_addresses();
1114        
1115        assert_eq!(limited_addresses.len(), 1);
1116        assert!(limited_addresses.contains(&"/ip4/0.0.0.0/tcp/0".to_string()));
1117    }
1118
1119    #[tokio::test]
1120    async fn test_transport_options_configuration() {
1121        let options = TransportOptions {
1122            enable_0rtt: false,
1123            require_encryption: false,
1124            connect_timeout: Duration::from_secs(10),
1125            keep_alive: Duration::from_secs(30),
1126            max_message_size: 1024,
1127        };
1128
1129        assert!(!options.enable_0rtt);
1130        assert!(!options.require_encryption);
1131        assert_eq!(options.connect_timeout, Duration::from_secs(10));
1132        assert_eq!(options.keep_alive, Duration::from_secs(30));
1133        assert_eq!(options.max_message_size, 1024);
1134    }
1135}
p2p_foundation/transport.rs

p2p_foundation/
transport.rs
