flowsdk 0.5.1

Safety-first, realistic, behavior-predictable messaging SDK for MQTT and more.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112

// SPDX-License-Identifier: MPL-2.0

//! Transport abstraction layer for MQTT connections
//!
//! This module provides a clean abstraction for different connection types (TCP, TLS, WebSocket, etc.)
//! allowing the MQTT client to work with any underlying transport without modification.

use async_trait::async_trait;
use std::io;
use tokio::io::{AsyncRead, AsyncWrite};

#[cfg(feature = "async-client")]
pub mod tcp;

#[cfg(feature = "tls")]
pub mod tls;

#[cfg(feature = "quic")]
pub mod quic;

#[cfg(feature = "rustls-tls")]
pub mod rustls_tls;

/// Error type for transport operations
#[derive(Debug, thiserror::Error)]
pub enum TransportError {
    #[error("Connection failed: {0}")]
    ConnectionFailed(String),

    #[error("IO error: {0}")]
    Io(#[from] io::Error),

    #[cfg(any(feature = "tls", feature = "rustls-tls"))]
    #[error("TLS error: {0}")]
    Tls(String),

    #[cfg(feature = "quic")]
    #[error("QUIC error: {0}")]
    Quic(String),

    #[error("Transport not supported: {0}")]
    NotSupported(String),

    #[error("Invalid address: {0}")]
    InvalidAddress(String),
}

/// Parse MQTT address to extract hostname and socket address
///
/// Supports formats:
/// - "mqtts://host:port" → ("host", "host:port")
/// - "host:port" → ("host", "host:port")
pub fn parse_mqtt_address(addr: &str) -> Result<(String, String), TransportError> {
    // Remove mqtts:// prefix if present
    let addr = addr.trim_start_matches("mqtts://");

    // Split host and port
    if let Some((host, port)) = addr.rsplit_once(':') {
        Ok((host.to_string(), format!("{}:{}", host, port)))
    } else {
        Err(TransportError::InvalidAddress(format!(
            "Invalid address format: '{}'. Expected 'host:port' or 'mqtts://host:port'",
            addr
        )))
    }
}

/// Transport trait for different connection types
///
/// This trait provides a unified interface for TCP, TLS, WebSocket, and other
/// transport protocols. All transports must implement AsyncRead and AsyncWrite
/// for compatibility with existing I/O code.
#[async_trait]
pub trait Transport: AsyncRead + AsyncWrite + Send + Sync + Unpin {
    /// Connect to the specified address
    ///
    /// # Arguments
    /// * `addr` - The address to connect to (format depends on transport type)
    ///
    /// # Returns
    /// A new connected transport instance
    async fn connect(addr: &str) -> Result<Self, TransportError>
    where
        Self: Sized;

    /// Gracefully close the connection
    async fn close(&mut self) -> Result<(), TransportError>;

    /// Get the peer address as a string
    fn peer_addr(&self) -> Result<String, TransportError>;

    /// Get the local address as a string
    fn local_addr(&self) -> Result<String, TransportError>;

    /// Set TCP_NODELAY option (no-op for non-TCP transports)
    fn set_nodelay(&self, _nodelay: bool) -> Result<(), TransportError> {
        Ok(()) // Default implementation does nothing
    }
}

/// Boxed transport for dynamic dispatch
pub type BoxedTransport = Box<dyn Transport>;

// Re-export transport types
#[cfg(feature = "async-client")]
pub use tcp::TcpTransport;

#[cfg(feature = "tls")]
pub use tls::TlsTransport;

#[cfg(feature = "rustls-tls")]
pub use rustls_tls::{RustlsTlsConfig, RustlsTlsTransport};