feature(CI): Add test for AWS IoT Device Advisor

.github/workflows/ci.yml

@@ -72,9 +72,8 @@ jobs:
       AWS_DEFAULT_REGION: ${{ secrets.MGMT_AWS_DEFAULT_REGION }}
       AWS_ACCESS_KEY_ID: ${{ secrets.MGMT_AWS_ACCESS_KEY_ID }}
       AWS_SECRET_ACCESS_KEY: ${{ secrets.MGMT_AWS_SECRET_ACCESS_KEY }}
-      SUITE_ID: greb3uy2wtq3
-      THING_ARN: arn:aws:iot:eu-west-1:274906834921:thing/mqttrust
-      CERTIFICATE_ARN: arn:aws:iot:eu-west-1:274906834921:cert/e7280d8d316b58da3058037a2c1730d9eb15de50e96f4d47e54ea655266b76db
+      SUITE_ID: 1gaev57dq6i5
+      THING_ARN: arn:aws:iot:eu-west-1:411974994697:thing/mqttrust
     steps:
       - name: Checkout
         uses: actions/checkout@v1
@@ -89,7 +88,7 @@ jobs:
       - name: Get AWS_HOSTNAME
         id: hostname
         run: |
-          hostname=$(aws iotdeviceadvisor get-endpoint --output text --query endpoint)
+          hostname=$(aws iotdeviceadvisor get-endpoint --thing-arn ${{ env.THING_ARN }} --output text --query endpoint)
           ret=$?
           echo "::set-output name=AWS_HOSTNAME::$hostname"
           exit $ret
@@ -105,7 +104,7 @@ jobs:
       - name: Start test suite
         id: test_suite
         run: |
-          suite_id=$(aws iotdeviceadvisor start-suite-run --suite-definition-id ${{ env.SUITE_ID }} --suite-run-configuration "primaryDevice={thingArn=${{ env.THING_ARN }},certificateArn=${{ env.CERTIFICATE_ARN }}}" --output text --query suiteRunId)
+          suite_id=$(aws iotdeviceadvisor start-suite-run --suite-definition-id ${{ env.SUITE_ID }} --suite-run-configuration "primaryDevice={thingArn=${{ env.THING_ARN }}},parallelRun=true" --output text --query suiteRunId)
           ret=$?
           echo "::set-output name=SUITE_RUN_ID::$suite_id"
           exit $ret
@@ -121,9 +120,9 @@ jobs:
 
       - name: Monitor test run
         run: |
-          chmod +x ./.github/scripts/da_monitor.sh
+          chmod +x ./scripts/da_monitor.sh
           echo ${{ env.SUITE_ID }} ${{ steps.test_suite.outputs.SUITE_RUN_ID }} ${{ steps.binary.outputs.PID }}
-          ./.github/scripts/da_monitor.sh ${{ env.SUITE_ID }} ${{ steps.test_suite.outputs.SUITE_RUN_ID }} ${{ steps.binary.outputs.PID }}
+          ./scripts/da_monitor.sh ${{ env.SUITE_ID }} ${{ steps.test_suite.outputs.SUITE_RUN_ID }} ${{ steps.binary.outputs.PID }}
 
       - name: Kill test binary process
         if: ${{ always() }}

.gitignore

@@ -3,3 +3,4 @@
 .gdb_history
 Cargo.lock
 target/
+device_advisor_integration.log

mqttrust_core/Cargo.toml

@@ -40,11 +40,14 @@ defmt = { version = "^0.3", optional = true }
 [dev-dependencies]
 native-tls = { version = "^0.2" }
 dns-lookup = "1.0.3"
-env_logger = "0.9.0"
+env_logger = "0.11"
+static_cell = "2.1"
 
 [features]
 default = []
 
 std = []
 
-defmt-impl = ["defmt", "mqttrust/defmt-impl", "heapless/defmt-impl", "fugit/defmt"]
+log = ["dep:log", "mqttrust/log"]
+
+defmt-impl = ["dep:defmt", "mqttrust/defmt-impl", "heapless/defmt-impl", "fugit/defmt"]

mqttrust_core/examples/aws_device_advisor.rs

@@ -1,56 +1,77 @@
 mod common;
 
+use mqttrust::encoding::v4::{Connack, ConnectReturnCode};
 use mqttrust::{Mqtt, QoS, SubscribeTopic};
 use mqttrust_core::bbqueue::BBBuffer;
 use mqttrust_core::{EventLoop, MqttOptions, Notification};
 
 use common::clock::SysClock;
 use common::network::Network;
 use native_tls::TlsConnector;
+use static_cell::StaticCell;
+use std::sync::atomic::AtomicBool;
 use std::sync::Arc;
 use std::thread;
 
 use crate::common::credentials;
 
-static mut Q: BBBuffer<{ 1024 * 6 }> = BBBuffer::new();
+static mut Q: BBBuffer<{ 1024 * 60 }> = BBBuffer::new();
 
 fn main() {
     env_logger::init();
 
     let (p, c) = unsafe { Q.try_split_framed().unwrap() };
 
-    let hostname = credentials::HOSTNAME.unwrap();
+    static HOSTNAME: StaticCell<String> = StaticCell::new();
+    let hostname = HOSTNAME.init(credentials::hostname());
+
+    log::info!(
+        "Starting device advisor test on endpoint {}",
+        hostname.as_str()
+    );
 
     let connector = TlsConnector::builder()
         .identity(credentials::identity())
         .add_root_certificate(credentials::root_ca())
         .build()
         .unwrap();
 
-    let mut network = Network::new_tls(connector, String::from(hostname));
+    let mut network = Network::new_tls(connector, hostname.clone());
 
     let thing_name = "mqttrust";
 
     let mut mqtt_eventloop = EventLoop::new(
         c,
         SysClock::new(),
-        MqttOptions::new(thing_name, hostname.into(), 8883),
+        MqttOptions::new(thing_name, hostname.as_str().into(), 8883),
     );
 
     let mqtt_client = mqttrust_core::Client::new(p, thing_name);
 
+    let connected = Arc::new(AtomicBool::new(false));
+    let con = connected.clone();
+
     thread::Builder::new()
         .name("eventloop".to_string())
         .spawn(move || loop {
             match nb::block!(mqtt_eventloop.connect(&mut network)) {
                 Err(_) => continue,
-                Ok(true) => {
-                    log::info!("Successfully connected to broker");
+                Ok(Some(Notification::ConnAck(Connack {
+                    session_present,
+                    code: ConnectReturnCode::Accepted,
+                }))) => {
+                    log::info!(
+                        "Successfully connected to broker. session_present: {}",
+                        session_present
+                    );
+                    con.store(true, std::sync::atomic::Ordering::Release);
+                }
+                Ok(n) => {
+                    log::info!("Received {:?} during connect", n);
                 }
-                Ok(false) => {}
             }
 
-            match mqtt_eventloop.yield_event(&mut network) {
+            match nb::block!(mqtt_eventloop.yield_event(&mut network)) {
                 Ok(Notification::Publish(_)) => {}
                 Ok(n) => {
                     log::trace!("{:?}", n);
@@ -62,19 +83,21 @@ fn main() {
 
     loop {
         thread::sleep(std::time::Duration::from_millis(5000));
-        mqtt_client
-            .subscribe(&[SubscribeTopic {
-                topic_path: format!("{}/device/advisor", thing_name).as_str(),
-                qos: QoS::AtLeastOnce,
-            }])
-            .unwrap();
-
-        mqtt_client
-            .publish(
-                format!("{}/device/advisor/hello", thing_name).as_str(),
-                format!("Hello from {}", thing_name).as_bytes(),
-                QoS::AtLeastOnce,
-            )
-            .unwrap();
+        if connected.load(std::sync::atomic::Ordering::Acquire) {
+            mqtt_client
+                .subscribe(&[SubscribeTopic {
+                    topic_path: format!("plc/output/{}", thing_name).as_str(),
+                    qos: QoS::AtLeastOnce,
+                }])
+                .unwrap();
+
+            mqtt_client
+                .publish(
+                    format!("plc/input/{}", thing_name).as_str(),
+                    format!("Hello from {}", thing_name).as_bytes(),
+                    QoS::AtLeastOnce,
+                )
+                .unwrap();
+        }
     }
 }

mqttrust_core/examples/common/clock.rs

@@ -5,14 +5,12 @@ use std::{
 };
 pub struct SysClock {
     start_time: u32,
-    countdown_end: Option<u32>,
 }
 
 impl SysClock {
     pub fn new() -> Self {
         Self {
             start_time: Self::epoch(),
-            countdown_end: None,
         }
     }
 

mqttrust_core/examples/common/credentials.rs

@@ -2,13 +2,15 @@
 
 use native_tls::{Certificate, Identity};
 
 pub fn identity() -> Identity {
     let pw = env::var("DEVICE_ADVISOR_PASSWORD").unwrap();
     Identity::from_pkcs12(include_bytes!("../secrets/identity.pfx"), pw.as_str()).unwrap()
 }
 
 pub fn root_ca() -> Certificate {
     Certificate::from_pem(include_bytes!("../secrets/root-ca.pem")).unwrap()
 }
 
-pub const HOSTNAME: Option<&'static str> = option_env!("AWS_HOSTNAME");
+pub fn hostname() -> String {
+    env::var("AWS_HOSTNAME").unwrap()
+}

mqttrust_core/src/eventloop.rs

@@ -7,7 +7,7 @@
 use core::ops::DerefMut;
 use core::ops::RangeTo;
 use embedded_nal::{AddrType, Dns, SocketAddr, TcpClientStack};
-use fugit::ExtU32;
+use fugit::{ExtU32, TimerDurationU32};
 use heapless::{String, Vec};
 use mqttrust::encoding::v4::{decode_slice, encode_slice, Connect, Packet, Protocol};
 
@@ -24,6 +24,7 @@
     /// Request stream
     pub(crate) requests: Option<FrameConsumer<'a, L>>,
     network_handle: NetworkHandle<S>,
+    connect_counter: u8,
 }
 
 impl<'a, 'b, S, O, const TIMER_HZ: u32, const L: usize> EventLoop<'a, 'b, S, O, TIMER_HZ, L>
@@ -41,6 +42,7 @@
             options,
             requests: Some(requests),
             network_handle: NetworkHandle::new(),
+            connect_counter: 0,
         }
     }
 
@@ -51,10 +53,10 @@
         self.requests.take()
     }
 
     pub fn connect<N: Dns + TcpClientStack<TcpSocket = S> + ?Sized>(
         &mut self,
         network: &mut N,
-    ) -> nb::Result<bool, EventError> {
+    ) -> nb::Result<Option<Notification>, EventError> {
         // connect to the broker
         match self.network_handle.is_connected(network) {
             Ok(false) => {
@@ -112,7 +114,7 @@
         // };
 
         // qos_0 || (self.requests.ready() && qos_space)
         qos_space
     }
 
     /// Selects an event from the client's requests, incoming packets from the
@@ -128,7 +130,7 @@
             match &mut self.requests {
                 Some(requests) => {
                     if let Some(mut grant) = requests.read() {
                         let mut packet = SerializedPacket(grant.deref_mut());
                         match self.state.handle_outgoing_request(&mut packet, &now) {
                             Ok(()) => {
                                 self.network_handle.send(network, packet.to_inner())?;
@@ -180,7 +182,7 @@
             // Update inflight's timestamp for later retrials
             inflight.last_touch_entry().insert(now);
             let packet = inflight.packet(*pid).map_err(EventError::from)?;
             self.network_handle.send(network, &packet)?;
         }
 
         notification.ok_or(nb::Error::WouldBlock)
@@ -189,10 +191,10 @@
     /// Yields notification from events. All the error raised while processing
     /// event is reported as an `Ok` value of `Notification::Abort`.
     #[must_use = "Eventloop should be iterated over a loop to make progress"]
     pub fn yield_event<N: TcpClientStack<TcpSocket = S> + ?Sized>(
         &mut self,
         network: &mut N,
     ) -> nb::Result<Notification, Infallible> {
         if self.network_handle.socket.is_none() {
             return Ok(Notification::Abort(EventError::Network(
                 NetworkError::NoSocket,
@@ -216,16 +218,23 @@
         }
     }
 
+    fn backoff(&self) -> TimerDurationU32<TIMER_HZ> {
+        let base_time_ms: u32 = 200;
+        let backoff = base_time_ms.saturating_mul(u32::pow(2, self.connect_counter as u32));
+
+        core::cmp::min(50.secs(), backoff.millis())
+    }
+
     fn mqtt_connect<N: TcpClientStack<TcpSocket = S> + ?Sized>(
         &mut self,
         network: &mut N,
-    ) -> nb::Result<bool, EventError> {
+    ) -> nb::Result<Option<Notification>, EventError> {
         match self.state.connection_status {
-            MqttConnectionStatus::Connected => Ok(false),
+            MqttConnectionStatus::Connected => Ok(None),
             MqttConnectionStatus::Disconnected => {
-                info!("MQTT connecting..");
                 let now = self.last_outgoing_timer.now();
                 self.state.last_ping_entry().insert(now);
+                self.connect_counter += 1;
 
                 self.state.await_pingresp = false;
                 self.network_handle.rx_buf.init();
@@ -234,7 +243,7 @@
 
                 let connect = Packet::Connect(Connect {
                     protocol: Protocol::MQTT311,
                     keep_alive: (self.options.keep_alive_ms() / 1000) as u16,
                     client_id: self.options.client_id(),
                     clean_session: self.options.clean_session(),
                     last_will: self.options.last_will(),
@@ -242,6 +251,12 @@
                     password,
                 });
 
+                info!(
+                    "MQTT connecting.. Attempt: {}. Backoff time: {}",
+                    self.connect_counter,
+                    self.backoff().to_millis()
+                );
+
                 // mqtt connection with timeout
                 self.network_handle.send_packet(network, &connect)?;
                 self.state.handle_outgoing_connect();
@@ -250,25 +265,37 @@
             MqttConnectionStatus::Handshake => {
                 let now = self.last_outgoing_timer.now();
 
+                let backoff_time = self.backoff();
+
                 if self
                     .state
                     .last_ping_entry()
                     .or_insert(now)
-                    .has_elapsed(&now, 50.secs())
+                    .has_elapsed(&now, backoff_time)
                 {
+                    warn!("Timed out waiting for connect packet!");
                     return Err(nb::Error::Other(EventError::Timeout));
                 }
 
-                self.network_handle
+                let res = self
+                    .network_handle
                     .receive(network)
                     .map_err(|e| e.map(EventError::Network))?
                     .decode(&mut self.state)
                     .and_then(|(n, p)| {
                         if n.is_none() && p.is_none() {
                             return Err(nb::Error::WouldBlock);
                         }
-                        Ok(n.map(|n| n == Notification::ConnAck).unwrap_or(false))
-                    })
+                        Ok(n)
+                    });
+
+                match res {
+                    Ok(r) => {
+                        self.connect_counter = 0;
+                        Ok(r)
+                    }
+                    Err(e) => Err(e),
+                }
             }
         }
     }
@@ -284,7 +311,7 @@
 impl<S> NetworkHandle<S> {
     fn lookup_host<N: Dns + TcpClientStack<TcpSocket = S> + ?Sized>(
         network: &mut N,
         broker: Broker,
         port: u16,
     ) -> Result<(String<256>, SocketAddr), NetworkError> {
         match broker {
@@ -324,12 +351,12 @@
         &self,
         network: &mut N,
     ) -> Result<bool, NetworkError> {
         match self.socket {
             Some(ref socket) => network
                 .is_connected(socket)
                 .map_err(|_e| NetworkError::SocketClosed),
             None => Err(NetworkError::SocketClosed),
         }
     }
 
     fn connect<N: Dns + TcpClientStack<TcpSocket = S> + ?Sized>(
@@ -346,7 +373,7 @@
         };
 
         let (broker, port) = broker;
         let (_hostname, socket_addr) = NetworkHandle::<S>::lookup_host(network, broker, port)?;
 
         nb::block!(network.connect(socket, socket_addr)).map_err(|_| {
             if let Some(socket) = self.socket.take() {
@@ -356,7 +383,7 @@
         })
     }
 
     pub fn send_packet<'d, N: TcpClientStack<TcpSocket = S> + ?Sized>(
         &mut self,
         network: &mut N,
         pkt: &Packet,
@@ -366,7 +393,7 @@
             .resize_default(self.tx_buf.capacity())
             .unwrap_or_else(|()| unreachable!("Input length equals to the current capacity."));
 
         let size = encode_slice(&pkt, self.tx_buf.as_mut()).map_err(EventError::Encoding)?;
 
         let socket = self
             .socket
@@ -381,7 +408,7 @@
         Ok(length)
     }
 
     pub fn send<'d, N: TcpClientStack<TcpSocket = S> + ?Sized>(
         &mut self,
         network: &mut N,
         pkt: &[u8],
@@ -391,7 +418,7 @@
             .as_mut()
             .ok_or(EventError::Network(NetworkError::NoSocket))?;
 
         let length = nb::block!(network.send(socket, &pkt)).map_err(|_| {
             error!("[send] NetworkError::Write");
             EventError::Network(NetworkError::Write)
         })?;
@@ -417,7 +444,7 @@
 #[derive(Debug)]
 struct PacketBuffer {
     range: RangeTo<usize>,
-    buffer: Vec<u8, 4096>,
+    buffer: Vec<u8, { 1024 * 16 }>,
 }
 
 impl PacketBuffer {
@@ -518,7 +545,7 @@
             })
             .enumerate()
             .fold(1, |acc, (i, length)| {
                 acc + 1 + length * 0x80_usize.pow(i as u32)
             })
             .into()
     }
@@ -536,6 +563,7 @@
                 Err(EventError::Encoding(e).into())
             }
             Ok(Some(packet)) => {
+                warn!("Got packet! {:?}", packet);
                 self.is_err.replace(false);
                 state
                     .handle_incoming_packet(packet)
@@ -679,7 +707,7 @@
         let mut state = MqttState::<1000>::new();
         const LEN: usize = 1024 * 10;
         static mut PUBLISH_MEM: [u8; LEN] = [0u8; LEN];
         BoxedPublish::grow(unsafe { &mut PUBLISH_MEM });
 
         let mut rx_buf = PacketBuffer::new();
         let connack = Connack {
@@ -699,12 +727,12 @@
         rx_buf.range.end += connack_len;
         let publish_len = encode_slice(&Packet::from(publish.clone()), rx_buf.buffer()).unwrap();
         rx_buf.range.end += publish_len;
-        assert_eq!(rx_buf.range.end, rx_buf.buffer.capacity());
+        assert_eq!(rx_buf.range.end, 4096);
 
         // Decode the first Connack packet on the Handshake state.
         state.connection_status = MqttConnectionStatus::Handshake;
         let (n, p) = PacketDecoder::new(&mut rx_buf).decode(&mut state).unwrap();
-        assert_eq!(n, Some(Notification::ConnAck));
+        assert!(matches!(n, Some(Notification::ConnAck(_))));
         assert_eq!(p, None);
 
         let mut pkg = SerializedPacket(&mut rx_buf.buffer[rx_buf.range]);
@@ -728,7 +756,7 @@
         let mut state = MqttState::<1000>::new();
         const LEN: usize = 1024 * 10;
         static mut PUBLISH_MEM: [u8; LEN] = [0u8; LEN];
         BoxedPublish::grow(unsafe { &mut PUBLISH_MEM });
 
         let mut rx_buf = PacketBuffer::new();
         let connack_malformed = Connack {
@@ -750,7 +778,7 @@
         rx_buf.range.end += connack_malformed_len;
         let publish_len = encode_slice(&Packet::from(publish.clone()), rx_buf.buffer()).unwrap();
         rx_buf.range.end += publish_len;
-        assert_eq!(rx_buf.range.end, rx_buf.buffer.capacity());
+        assert_eq!(rx_buf.range.end, 4096);
 
         // When a packet is malformed, we cannot tell its length. The decoder
         // discards the entire buffer.

mqttrust_core/src/lib.rs

@@ -18,12 +18,13 @@
 use core::convert::TryFrom;
 pub use eventloop::EventLoop;
 use heapless::{String, Vec};
+use mqttrust::encoding::v4::Connack;
 pub use mqttrust::encoding::v4::{Pid, Publish, QoS, QosPid, Suback};
 pub use mqttrust::*;
 pub use options::{Broker, MqttOptions};
 use state::StateError;
 
 #[derive(Debug, PartialEq)]
 #[cfg_attr(feature = "defmt-impl", derive(defmt::Format))]
 pub struct PublishNotification {
     pub dup: bool,
@@ -39,7 +40,7 @@
 // #[cfg_attr(feature = "defmt-impl", derive(defmt::Format))]
 pub enum Notification {
     /// Incoming connection acknowledge
-    ConnAck,
+    ConnAck(Connack),
     /// Incoming publish from the broker
     #[cfg(not(feature = "std"))]
     Publish(heapless::pool::singleton::Box<state::BoxedPublish, heapless::pool::Init>),
@@ -65,12 +66,12 @@
     type Error = StateError;
 
     fn try_from(p: Publish<'a>) -> Result<Self, Self::Error> {
         Ok(PublishNotification {
             dup: p.dup,
             qospid: p.qos,
             retain: p.retain,
             topic_name: String::from(p.topic_name),
             payload: Vec::from_slice(p.payload).map_err(|_| {
                 error!("Failed to convert payload to notification!");
                 StateError::PayloadEncoding
             })?,
@@ -91,7 +92,7 @@
     RequestsNotAvailable,
 }
 
 #[derive(Debug, PartialEq)]
 #[cfg_attr(feature = "defmt-impl", derive(defmt::Format))]
 pub enum NetworkError {
     Read,
@@ -105,12 +106,12 @@
 
 impl From<mqttrust::encoding::v4::Error> for EventError {
     fn from(e: mqttrust::encoding::v4::Error) -> Self {
         EventError::Encoding(e)
     }
 }
 
 impl From<StateError> for EventError {
     fn from(e: StateError) -> Self {
         EventError::MqttState(e)
     }
 }