feat: sans io quic (#61)

Author: qzhuyan

Cargo.toml

@@ -29,6 +29,7 @@ native-tls = { version = "0.2", optional = true }
 
 # QUIC support (optional)
 quinn = { version = "0.11", optional = true }
+quinn-proto = { version = "0.11", optional = true }
 rustls = { version = "0.23", optional = true, default-features = false, features = ["ring", "std"] }
 rustls-native-certs = { version = "0.7", optional = true }
 rustls-pki-types = { version = "1", optional = true }
@@ -39,7 +40,7 @@ default = ["strict-protocol-compliance", "tls"]
 # TLS/SSL transport support
 tls = ["dep:tokio-native-tls", "dep:native-tls"]
 # QUIC transport support
-quic = ["dep:quinn", "dep:rustls", "dep:rustls-native-certs", "dep:rustls-pki-types"]
+quic = ["dep:quinn", "dep:quinn-proto", "dep:rustls", "dep:rustls-native-certs", "dep:rustls-pki-types"]
 # Rustls-based TLS over TCP (mqtts://) transport support
 rustls-tls = [
 	"dep:tokio-rustls",
@@ -56,3 +57,22 @@ protocol-testing = []
 # usage: cargo build --target aarch64-unknown-linux-musl --examples --features quic
 [target.aarch64-unknown-linux-musl]
 rustflags = ["-C", "target-feature=+crt-static"]
+
+[dev-dependencies]
+futures = "0.3"
+
+[[example]]
+name = "no_io_quic_async_client_example"
+required-features = ["quic"]
+
+[[example]]
+name = "no_io_quic_client_example"
+required-features = ["quic"]
+
+[[example]]
+name = "no_io_tokio_quic_client_example"
+required-features = ["quic"]
+
+[[example]]
+name = "tokio_async_mqtt_quic_example"
+required-features = ["quic"]

examples/mqtt_client_v3.rs examples/mqtt_client_v3_example.rsexamples/mqtt_client_v3.rs renamed to examples/mqtt_client_v3_example.rs

@@ -0,0 +1,143 @@
+use flowsdk::mqtt_client::commands::PublishCommand;
+use flowsdk::mqtt_client::engine::{MqttEvent, QuicMqttEngine};
+use flowsdk::mqtt_client::opts::MqttClientOptions;
+use std::future::Future;
+use std::net::{ToSocketAddrs, UdpSocket};
+use std::pin::Pin;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+use std::task::{Context, Poll};
+use std::time::{Duration, Instant};
+
+struct ProtocolDriver {
+    engine: QuicMqttEngine,
+    socket: UdpSocket,
+    server_addr: std::net::SocketAddr,
+    last_tick: Instant,
+    published: bool,
+    running: Arc<AtomicBool>,
+    exit_when_rcvd: bool,
+}
+
+impl Future for ProtocolDriver {
+    type Output = Result<(), Box<dyn std::error::Error>>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        if !self.running.load(Ordering::SeqCst) {
+            return Poll::Ready(Ok(()));
+        }
+
+        let now = Instant::now();
+        let mut buf = [0u8; 2048];
+        while let Ok((len, remote)) = self.socket.recv_from(&mut buf) {
+            if remote == self.server_addr {
+                self.engine
+                    .handle_datagram(buf[..len].to_vec(), remote, now);
+            }
+        }
+
+        if now.duration_since(self.last_tick) >= Duration::from_millis(10) {
+            let events = self.engine.handle_tick(now);
+            for event in events {
+                match event {
+                    MqttEvent::Connected(_) => println!("Async Driver: MQTT Connected!"),
+                    MqttEvent::Published(res) => {
+                        println!("Async Driver: Message Published: ID={:?}", res.packet_id)
+                    }
+                    MqttEvent::Subscribed(res) => {
+                        println!("Async Driver: Subscribed: ID={:?}", res.packet_id)
+                    }
+                    MqttEvent::Disconnected(reason) => {
+                        println!("Async Driver: MQTT Disconnected: {:?}", reason);
+                        return Poll::Ready(Ok(()));
+                    }
+                    _ => {}
+                }
+            }
+            self.last_tick = now;
+        }
+
+        let mut dags = self.engine.take_outgoing_datagrams();
+        while let Some((dest, data)) = dags.pop_front() {
+            let _ = self.socket.send_to(&data, dest);
+        }
+
+        if self.engine.is_connected() && !self.published {
+            let pub_cmd = PublishCommand::builder()
+                .topic("test/topic/async")
+                .payload("Hello from non-tokio async!".to_string())
+                .qos(1)
+                .build();
+
+            if let Ok(cmd) = pub_cmd {
+                let _ = self.engine.publish(cmd);
+            }
+            self.published = true;
+            if self.exit_when_rcvd {
+                self.running.store(false, Ordering::SeqCst);
+            }
+        }
+
+        cx.waker().wake_by_ref();
+        Poll::Pending
+    }
+}
+
+/// Demonstrates how to use the QuicMqttEngine as a non-tokio async driver.
+pub fn run_example(exit_when_rcvd: bool) -> Result<(), Box<dyn std::error::Error>> {
+    let _ = rustls::crypto::ring::default_provider().install_default();
+    let mqtt_opts = MqttClientOptions::builder()
+        .client_id("no-io-quic-async-client")
+        .peer("broker.emqx.io:14567")
+        .build();
+
+    let mut root_store = rustls::RootCertStore::empty();
+    for cert in rustls_native_certs::load_native_certs()? {
+        root_store.add(cert).ok();
+    }
+    let crypto = rustls::ClientConfig::builder()
+        .with_root_certificates(root_store)
+        .with_no_client_auth();
+
+    let socket = UdpSocket::bind("0.0.0.0:0")?;
+    socket.set_nonblocking(true)?;
+    let server_addr = ("broker.emqx.io", 14567u16)
+        .to_socket_addrs()?
+        .next()
+        .ok_or("DNS Failure")?;
+
+    let mut engine = QuicMqttEngine::new(mqtt_opts)?;
+    engine.connect(server_addr, "broker.emqx.io", crypto, Instant::now())?;
+
+    let running = Arc::new(AtomicBool::new(true));
+    let r = running.clone();
+    ctrlc::set_handler(move || {
+        r.store(false, Ordering::SeqCst);
+    })?;
+
+    let driver = ProtocolDriver {
+        engine,
+        socket,
+        server_addr,
+        last_tick: Instant::now(),
+        published: false,
+        running,
+        exit_when_rcvd,
+    };
+
+    futures::executor::block_on(driver)
+}
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    run_example(false)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_example() {
+        run_example(true).unwrap();
+    }
+}

examples/mqtt_client_v5.rs examples/mqtt_client_v5_example.rsexamples/mqtt_client_v5.rs renamed to examples/mqtt_client_v5_example.rs

@@ -0,0 +1,106 @@
+use flowsdk::mqtt_client::commands::PublishCommand;
+use flowsdk::mqtt_client::engine::{MqttEvent, QuicMqttEngine};
+use flowsdk::mqtt_client::opts::MqttClientOptions;
+use flowsdk::mqtt_client::SubscribeCommand;
+use std::net::{ToSocketAddrs, UdpSocket};
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let _ = rustls::crypto::ring::default_provider().install_default();
+    let mqtt_opts = MqttClientOptions::builder()
+        .client_id("no-io-quic-client")
+        .peer("broker.emqx.io:14567")
+        .keep_alive(30)
+        .build();
+
+    let mut root_store = rustls::RootCertStore::empty();
+    for cert in rustls_native_certs::load_native_certs()? {
+        root_store.add(cert).ok();
+    }
+
+    let crypto = rustls::ClientConfig::builder()
+        .with_root_certificates(root_store)
+        .with_no_client_auth();
+
+    let mut engine = QuicMqttEngine::new(mqtt_opts)?;
+    let socket = UdpSocket::bind("0.0.0.0:0")?;
+    socket.set_nonblocking(true)?;
+    let server_addr = ("broker.emqx.io", 14567u16)
+        .to_socket_addrs()?
+        .next()
+        .ok_or("DNS Failure")?;
+
+    engine.connect(server_addr, "broker.emqx.io", crypto, Instant::now())?;
+
+    let running = Arc::new(AtomicBool::new(true));
+    let r = running.clone();
+    ctrlc::set_handler(move || {
+        r.store(false, Ordering::SeqCst);
+    })?;
+
+    let mut last_tick = Instant::now();
+    let mut published = false;
+    let mut subscribed = false;
+
+    loop {
+        if !running.load(Ordering::SeqCst) {
+            break Ok(());
+        }
+        let now = Instant::now();
+
+        let mut buf = [0u8; 2048];
+        while let Ok((len, remote)) = socket.recv_from(&mut buf) {
+            if remote == server_addr {
+                engine.handle_datagram(buf[..len].to_vec(), remote, now);
+            }
+        }
+
+        if now.duration_since(last_tick) >= Duration::from_millis(10) {
+            let events = engine.handle_tick(now);
+            for event in events {
+                match event {
+                    MqttEvent::Connected(_) => println!("MQTT Connected over QUIC!"),
+                    MqttEvent::Subscribed(res) => println!("Subscribed: ID={:?}", res.packet_id),
+                    MqttEvent::Published(res) => {
+                        println!("Message Published: ID={:?}", res.packet_id)
+                    }
+                    MqttEvent::Disconnected(reason) => {
+                        println!("MQTT Disconnected: {:?}", reason);
+                        return Ok(());
+                    }
+                    _ => {}
+                }
+            }
+            last_tick = now;
+        }
+
+        let mut dags = engine.take_outgoing_datagrams();
+        while let Some((dest, data)) = dags.pop_front() {
+            let _ = socket.send_to(&data, dest);
+        }
+
+        if engine.is_connected() && !subscribed {
+            let sub_cmd = SubscribeCommand::builder()
+                .add_topic("test/quic/topic", 1)
+                .build()
+                .unwrap();
+            let _ = engine.subscribe(sub_cmd);
+            subscribed = true;
+        }
+
+        if engine.is_connected() && !published {
+            let pub_cmd = PublishCommand::builder()
+                .topic("test/quic/topic")
+                .payload("Hello!".to_string())
+                .qos(1)
+                .build();
+            if let Ok(cmd) = pub_cmd {
+                let _ = engine.publish(cmd);
+            }
+            published = true;
+        }
+        std::thread::sleep(Duration::from_millis(1));
+    }
+}

examples/no_io_quic_async_client_example.rs

@@ -0,0 +1,87 @@
+use flowsdk::mqtt_client::commands::{PublishCommand, SubscribeCommand};
+use flowsdk::mqtt_client::engine::MqttEvent;
+use flowsdk::mqtt_client::opts::MqttClientOptions;
+use flowsdk::mqtt_client::tokio_quic_client::TokioQuicMqttClient;
+use std::net::ToSocketAddrs;
+use std::time::Duration;
+
+/// Demonstrates how to use the `TokioQuicMqttClient`, which is a wrapper around the `QuicMqttEngine`,
+/// in a more low-level, event-driven style where you manually drive the client's event loop.
+/// In most cases, users should prefer higher-level helpers, but this shows direct use of `TokioQuicMqttClient`.
+async fn run_example() -> Result<(), Box<dyn std::error::Error>> {
+    let _ = rustls::crypto::ring::default_provider().install_default();
+    let mqtt_opts = MqttClientOptions::builder()
+        .client_id("quic-async-wrapper-client")
+        .peer("broker.emqx.io:14567")
+        .build();
+
+    let mut root_store = rustls::RootCertStore::empty();
+    for cert in rustls_native_certs::load_native_certs()? {
+        root_store.add(cert).ok();
+    }
+    let crypto = rustls::ClientConfig::builder()
+        .with_root_certificates(root_store)
+        .with_no_client_auth();
+
+    let server_addr = ("broker.emqx.io", 14567u16)
+        .to_socket_addrs()?
+        .next()
+        .ok_or("DNS Failure")?;
+    let mut client = TokioQuicMqttClient::new(mqtt_opts)?;
+
+    client
+        .connect(server_addr, "broker.emqx.io".to_string(), crypto)
+        .await
+        .map_err(|e| e.to_string())?;
+
+    let mut subscribed = false;
+    let mut published = false;
+
+    loop {
+        tokio::select! {
+            event = client.next_event() => {
+                let Some(event) = event else { break; };
+                match event {
+                    MqttEvent::Connected(_) => {
+                        if !subscribed {
+                            let sub_cmd = SubscribeCommand::builder().add_topic("test/topic/wrapper", 1).build()?;
+                            client.subscribe(sub_cmd).await.map_err(|e| e.to_string())?;
+                            subscribed = true;
+                        }
+                    }
+                    MqttEvent::Subscribed(res) => {
+                        println!("Subscribed: ID={:?}", res.packet_id);
+                        if !published {
+                            let pub_cmd = PublishCommand::builder().topic("test/topic/wrapper").payload("Hello!".to_string()).qos(1).build()?;
+                            client.publish(pub_cmd).await.map_err(|e| e.to_string())?;
+                            published = true;
+                        }
+                    }
+                    MqttEvent::Published(res) => {
+                        println!("Published: ID={:?}", res.packet_id);
+                        return Ok(());
+                    }
+                    MqttEvent::Disconnected(_) => return Ok(()),
+                    _ => {}
+                }
+            }
+            _ = tokio::time::sleep(Duration::from_secs(30)) => return Ok(()),
+        }
+    }
+    Ok(())
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    run_example().await
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn test_example() {
+        run_example().await.unwrap();
+    }
+}