test: add async versiosn of network loopback tests

fix: remove tokio version cap for dev-dep

mavlink-core/Cargo.toml

@@ -46,5 +46,4 @@ async-trait = { version = "0.1.18", optional = true }
 default = ["std", "tcp", "udp", "direct-serial", "serde"]
 
 [dev-dependencies]
-# 1.39 tokio macros seems to be currently broken
-tokio = { version = "1.0, <=1.38", default-features = false, features = ["io-util", "net", "sync", "fs","macros", "rt" ] }
+tokio = { version = "1.0", default-features = false, features = ["io-util", "net", "sync", "fs","macros", "rt" ] }

mavlink/Cargo.toml

@@ -115,3 +115,6 @@ features = [
     "tokio-1",
     "signing"
 ]
+
+[dev-dependencies]
+tokio = { version = "1.0", default-features = false, features = ["macros", "rt", "time" ] }

mavlink/tests/tcp_loopback_async_tests.rs

@@ -0,0 +1,69 @@
+mod test_shared;
+
+#[cfg(all(feature = "tokio-1", feature = "tcp", feature = "common"))]
+mod test_tcp_connections {
+    #[cfg(feature = "signing")]
+    use crate::test_shared;
+    #[cfg(feature = "signing")]
+    use mavlink::SigningConfig;
+
+    /// Test whether we can send a message via TCP and receive it OK using async_connect.
+    /// This also test signing as a property of a MavConnection if the signing feature is enabled.
+    #[tokio::test]
+    pub async fn test_tcp_loopback() {
+        const RECEIVE_CHECK_COUNT: i32 = 5;
+
+        #[cfg(feature = "signing")]
+        let singing_cfg_server = SigningConfig::new(test_shared::SECRET_KEY, 0, true, false);
+        #[cfg(feature = "signing")]
+        let singing_cfg_client = singing_cfg_server.clone();
+
+        let server_thread = tokio::spawn(async move {
+            //TODO consider using get_available_port to use a random port
+            let mut server =
+                mavlink::connect_async("tcpin:0.0.0.0:14551").await.expect("Couldn't create server");
+
+            #[cfg(feature = "signing")]
+            server.setup_signing(Some(singing_cfg_server));
+
+            let mut recv_count = 0;
+            for _i in 0..RECEIVE_CHECK_COUNT {
+                match server.recv().await {
+                    Ok((_header, msg)) => {
+                        if let mavlink::common::MavMessage::HEARTBEAT(_heartbeat_msg) = msg {
+                            recv_count += 1;
+                        } else {
+                            // one message parse failure fails the test
+                            break;
+                        }
+                    }
+                    Err(..) => {
+                        // one message read failure fails the test
+                        break;
+                    }
+                }
+            }
+            assert_eq!(recv_count, RECEIVE_CHECK_COUNT);
+        });
+
+        // Give some time for the server to connect
+        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+
+        // have the client send a few hearbeats
+        tokio::spawn(async move {
+            let msg =
+                mavlink::common::MavMessage::HEARTBEAT(crate::test_shared::get_heartbeat_msg());
+            let mut client =
+                mavlink::connect_async("tcpout:127.0.0.1:14551").await.expect("Couldn't create client");
+
+            #[cfg(feature = "signing")]
+            client.setup_signing(Some(singing_cfg_client));
+
+            for _i in 0..RECEIVE_CHECK_COUNT {
+                client.send_default(&msg).await.ok();
+            }
+        });
+
+        server_thread.await.unwrap();
+    }
+}

mavlink/tests/udp_loopback_async_tests.rs

@@ -0,0 +1,46 @@
+mod test_shared;
+
+#[cfg(all(feature = "tokio-1", feature = "udp", feature = "common"))]
+mod test_udp_connections {
+
+    /// Test whether we can send a message via UDP and receive it OK using async_connect
+    #[tokio::test]
+    pub async fn test_udp_loopback() {
+        const RECEIVE_CHECK_COUNT: i32 = 3;
+
+        let server = mavlink::connect_async("udpin:0.0.0.0:14552").await.expect("Couldn't create server");
+
+        // have the client send one heartbeat per second
+        tokio::spawn({
+            async move {
+                let msg =
+                    mavlink::common::MavMessage::HEARTBEAT(crate::test_shared::get_heartbeat_msg());
+                let client =
+                    mavlink::connect_async("udpout:127.0.0.1:14552").await.expect("Couldn't create client");
+                loop {
+                    client.send_default(&msg).await.ok();
+                }
+            }
+        });
+
+        //TODO use std::sync::WaitTimeoutResult to timeout ourselves if recv fails?
+        let mut recv_count = 0;
+        for _i in 0..RECEIVE_CHECK_COUNT {
+            match server.recv().await {
+                Ok((_header, msg)) => {
+                    if let mavlink::common::MavMessage::HEARTBEAT(_heartbeat_msg) = msg {
+                        recv_count += 1;
+                    } else {
+                        // one message parse failure fails the test
+                        break;
+                    }
+                }
+                Err(..) => {
+                    // one message read failure fails the test
+                    break;
+                }
+            }
+        }
+        assert_eq!(recv_count, RECEIVE_CHECK_COUNT);
+    }
+}