Improved rumqttc event loop check.

Signed-off-by: Suneet Nangia <[email protected]>

Author: suneetnangia

crates/outbound-mqtt/src/lib.rs

@@ -97,16 +97,31 @@ impl v2::HostConnection for OutboundMqtt {
                 .await
                 .map_err(other_error)?;
 
-            // Poll EventLoop once to send the message to MQTT broker or capture/throw error
+            // Poll event loop until outgoing publish event is iterated over to send the message to MQTT broker or capture/throw error.
             // We may revisit this later to manage long running connections and their issues in the connection pool.
-            eventloop
-                .poll()
-                .await
-                .map_err(|err: rumqttc::ConnectionError| {
-                    v2::Error::ConnectionFailed(err.to_string())
-                })?;
-
-            Ok(())
+            loop {
+                let event = eventloop
+                    .poll()
+                    .await
+                    .map_err(|err| v2::Error::ConnectionFailed(err.to_string()))?;
+
+                match event {
+                    rumqttc::Event::Outgoing(outgoing_event) => {
+                        match outgoing_event {
+                            rumqttc::Outgoing::Publish(_) => {
+                                return Ok(());
+                            }
+                            _ => {
+                                // We don't care about other outgoing event types in this loop check.
+                                continue;
+                            }
+                        }
+                    }
+                    rumqttc::Event::Incoming(_) => {
+                        // We don't care about incoming event types in this loop check.
+                    }
+                }
+            }
         }
         .await)
     }