gate mqtt on transport readiness

Author: ZeroBitsTech

examples/LiveNetworkNode/LiveNetworkNode.ino

@@ -401,6 +401,7 @@ void setup() {
                    mqttLoopStep,
                    mqttPublishStep,
                    mqttConfig);
+  g_mqttPump.setTransportProbe(isWiFiConnected);
 
 #if defined(ARDUINO_ARCH_ESP8266)
   ZeroKernel.addTask("Sample", sampleTask, 100, 0);

src/modules/net/ZeroMqttPump.h

@@ -26,6 +26,7 @@ class ZeroMqttPump {
   };
 
   typedef bool (*LinkProbe)();
+  typedef bool (*TransportProbe)();
   typedef bool (*ConnectStep)(void* context);
   typedef void (*LoopStep)(void* context);
   typedef bool (*PublishStep)(Kernel::TopicKey topicKey,
@@ -59,6 +60,7 @@ class ZeroMqttPump {
   ZeroMqttPump()
       : kernel_(NULL),
         linkProbe_(NULL),
+        transportProbe_(NULL),
         connectStep_(NULL),
         loopStep_(NULL),
         publishStep_(NULL),
@@ -95,6 +97,10 @@ class ZeroMqttPump {
     currentRetryMs_ = config.retryBaseMs;
   }
 
+  void setTransportProbe(TransportProbe transportProbe) {
+    transportProbe_ = transportProbe;
+  }
+
   void reset() {
     metrics_.reset();
     started_ = false;
@@ -159,6 +165,10 @@ class ZeroMqttPump {
     notifyStateIfChanged_();
 
     if (!connected_) {
+      if (transportProbe_ != NULL && !transportProbe_()) {
+        return;
+      }
+
       if (pendingRetryAtMs_ != 0 && nowMs < pendingRetryAtMs_) {
         return;
       }
@@ -277,6 +287,7 @@ class ZeroMqttPump {
 
   Kernel* kernel_;
   LinkProbe linkProbe_;
+  TransportProbe transportProbe_;
   ConnectStep connectStep_;
   LoopStep loopStep_;
   PublishStep publishStep_;

tests/desktop/KernelTests.cpp

@@ -249,6 +249,10 @@ bool mqttLinkProbe() {
   return g_mqttLinkUp;
 }
 
+bool mqttTransportProbe() {
+  return g_wifiLinkUp;
+}
+
 bool mqttConnectStep(void*) {
   ++g_mqttConnectCalls;
   g_mqttLinkUp = true;
@@ -1058,6 +1062,7 @@ int testMqttPumpModule() {
   g_mqttLoopCalls = 0;
   g_mqttPublishCalls = 0;
   g_mqttStateEvents = 0;
+  g_wifiLinkUp = false;
   g_mqttLinkUp = false;
   g_mqttLastState = false;
   g_mqttPublishShouldFail = true;
@@ -1071,6 +1076,7 @@ int testMqttPumpModule() {
              mqttLoopStep,
              mqttPublishStep,
              config);
+  pump.setTransportProbe(mqttTransportProbe);
 
   expectTrue(pump.enqueue(zerokernel::Kernel::makeTopicKey("mqtt.out"),
                           zerokernel::Kernel::EventValue::fromUnsigned(7UL)),
@@ -1079,22 +1085,28 @@ int testMqttPumpModule() {
   g_fakeNowMs = 1;
   isolatedKernel.tick();
   pump.tick();
+  expectTrue(g_mqttConnectCalls == 0, "mqtt pump waits for transport before connect");
+
+  g_wifiLinkUp = true;
+  g_fakeNowMs = 2;
+  isolatedKernel.tick();
+  pump.tick();
   expectTrue(pump.isConnected(), "mqtt pump connects on first retry");
   expectTrue(g_mqttStateEvents == 1, "mqtt pump publishes connected state");
   expectTrue(g_mqttLastState, "mqtt pump reports connected");
 
-  g_fakeNowMs = 2;
+  g_fakeNowMs = 3;
   isolatedKernel.tick();
   pump.tick();
   expectTrue(g_mqttPublishCalls == 1, "mqtt pump attempts first publish");
   expectTrue(pump.queuedCount() == 1, "mqtt pump keeps message queued after failure");
 
-  g_fakeNowMs = 4;
+  g_fakeNowMs = 5;
   isolatedKernel.tick();
   pump.tick();
   expectTrue(g_mqttPublishCalls == 1, "mqtt pump honors publish backoff");
 
-  g_fakeNowMs = 7;
+  g_fakeNowMs = 8;
   isolatedKernel.tick();
   pump.tick();
   expectTrue(g_mqttPublishCalls == 2, "mqtt pump retries publish after backoff");