add network transport modules

Author: ZeroBitsTech

examples/HttpPumpDemo/HttpPumpDemo.ino

@@ -0,0 +1,98 @@
+#include <ZeroKernel.h>
+#include <modules/net/ZeroHttpPump.h>
+
+using zerokernel::Kernel;
+using zerokernel::modules::net::ZeroHttpPump;
+
+namespace {
+
+ZeroHttpPump g_httpPump;
+bool g_connectPending = true;
+unsigned long g_requestId = 0;
+const Kernel::TopicKey kCompletionTopic = Kernel::makeTopicKey("http.done");
+
+ZeroHttpPump::StepResult connectStep(const ZeroHttpPump::Request&, void*) {
+  if (g_connectPending) {
+    g_connectPending = false;
+    return ZeroHttpPump::kStepPending;
+  }
+
+  return ZeroHttpPump::kStepComplete;
+}
+
+ZeroHttpPump::StepResult writeStep(const ZeroHttpPump::Request&, void*) {
+  return ZeroHttpPump::kStepComplete;
+}
+
+ZeroHttpPump::StepResult readStep(const ZeroHttpPump::Request&, void*) {
+  return ZeroHttpPump::kStepComplete;
+}
+
+ZeroHttpPump::StepResult closeStep(const ZeroHttpPump::Request&, void*) {
+  g_connectPending = true;
+  return ZeroHttpPump::kStepComplete;
+}
+
+void onCompletion(const char*, const Kernel::EventValue& value) {
+  if (value.type != Kernel::kEventBool) {
+    return;
+  }
+
+  Serial.print("http.done => ");
+  Serial.println(value.boolValue ? "ok" : "fail");
+}
+
+void queueRequestTask() {
+  static char payload[48];
+  const int written = snprintf(payload, sizeof(payload), "{\"seq\":%lu}", g_requestId++);
+  if (written <= 0) {
+    return;
+  }
+
+  ZeroHttpPump::Request request;
+  request.path = "/api/data";
+  request.contentType = "application/json";
+  request.body = payload;
+  request.bodyLength = static_cast<uint16_t>(written);
+  request.completionTopicKey = kCompletionTopic;
+  g_httpPump.enqueue(request);
+}
+
+void reportTask() {
+  const zerokernel::modules::net::ZeroTransportMetrics::Snapshot snapshot =
+      g_httpPump.metrics().snapshot();
+
+  Serial.print("http queue=");
+  Serial.print(g_httpPump.queuedCount());
+  Serial.print(" sent_ok=");
+  Serial.print(snapshot.sendSuccesses);
+  Serial.print(" sent_fail=");
+  Serial.print(snapshot.sendFailures);
+  Serial.print(" connect_attempts=");
+  Serial.println(snapshot.connectAttempts);
+}
+
+}  // namespace
+
+void setup() {
+  Serial.begin(115200);
+  delay(50);
+
+  ZeroKernel.begin();
+  ZeroKernel.subscribeTypedFast(kCompletionTopic, onCompletion);
+
+  ZeroHttpPump::Config config;
+  config.retryBaseMs = 100;
+  config.retryMaxMs = 500;
+  config.maxRetries = 1;
+
+  g_httpPump.begin(ZeroKernel, connectStep, writeStep, readStep, closeStep, config);
+
+  ZeroKernel.addTask("HttpQueue", queueRequestTask, 1500, 0);
+  ZeroKernel.addTask("HttpReport", reportTask, 1000, 0);
+}
+
+void loop() {
+  ZeroKernel.tick();
+  g_httpPump.tick();
+}

examples/MqttPumpDemo/MqttPumpDemo.ino

@@ -0,0 +1,88 @@
+#include <ZeroKernel.h>
+#include <modules/net/ZeroMqttPump.h>
+
+using zerokernel::Kernel;
+using zerokernel::modules::net::ZeroMqttPump;
+
+namespace {
+
+ZeroMqttPump g_mqttPump;
+bool g_brokerConnected = false;
+bool g_failNextPublish = true;
+unsigned long g_publishValue = 0;
+const Kernel::TopicKey kBrokerStateTopic = Kernel::makeTopicKey("mqtt.state");
+const Kernel::TopicKey kPublishTopic = Kernel::makeTopicKey("mqtt.out");
+
+bool linkProbe() {
+  return g_brokerConnected;
+}
+
+bool connectStep(void*) {
+  g_brokerConnected = true;
+  return true;
+}
+
+void loopStep(void*) {}
+
+bool publishStep(Kernel::TopicKey, const Kernel::EventValue&, void*) {
+  if (g_failNextPublish) {
+    g_failNextPublish = false;
+    return false;
+  }
+
+  return true;
+}
+
+void onStateEvent(const char*, const Kernel::EventValue& value) {
+  if (value.type != Kernel::kEventBool) {
+    return;
+  }
+
+  Serial.print("mqtt.state => ");
+  Serial.println(value.boolValue ? "connected" : "disconnected");
+}
+
+void queuePublishTask() {
+  g_mqttPump.enqueue(kPublishTopic, Kernel::EventValue::fromUnsigned(g_publishValue++), 1);
+}
+
+void reportTask() {
+  const zerokernel::modules::net::ZeroTransportMetrics::Snapshot snapshot =
+      g_mqttPump.metrics().snapshot();
+
+  Serial.print("mqtt queue=");
+  Serial.print(g_mqttPump.queuedCount());
+  Serial.print(" sent_ok=");
+  Serial.print(snapshot.sendSuccesses);
+  Serial.print(" sent_fail=");
+  Serial.print(snapshot.sendFailures);
+  Serial.print(" loop_calls=");
+  Serial.println(snapshot.loopCalls);
+}
+
+}  // namespace
+
+void setup() {
+  Serial.begin(115200);
+  delay(50);
+
+  ZeroKernel.begin();
+  ZeroKernel.subscribeTypedFast(kBrokerStateTopic, onStateEvent);
+
+  ZeroMqttPump::Config config;
+  config.pollIntervalMs = 50;
+  config.retryBaseMs = 100;
+  config.retryMaxMs = 500;
+  config.maxRetries = 1;
+  config.stateTopicKey = kBrokerStateTopic;
+
+  g_mqttPump.begin(ZeroKernel, linkProbe, connectStep, loopStep, publishStep, config);
+
+  ZeroKernel.addTask("MqttQueue", queuePublishTask, 1200, 0);
+  ZeroKernel.addTask("MqttReport", reportTask, 1000, 0);
+}
+
+void loop() {
+  ZeroKernel.tick();
+  g_mqttPump.tick();
+}

examples/TransportMetricsDemo/TransportMetricsDemo.ino

@@ -0,0 +1,52 @@
+#include <ZeroKernel.h>
+#include <modules/net/ZeroTransportMetrics.h>
+
+using zerokernel::modules::net::ZeroTransportMetrics;
+
+namespace {
+
+ZeroTransportMetrics g_metrics;
+
+void updateMetricsTask() {
+  static bool toggle = false;
+  toggle = !toggle;
+
+  g_metrics.recordConnectAttempt();
+  g_metrics.recordConnectResult(toggle, toggle ? 2 : 5);
+  g_metrics.recordSendQueued(2);
+  g_metrics.recordSendAttempt();
+  g_metrics.recordSendResult(toggle, toggle ? 1 : 4, toggle ? 3 : 8);
+  if (!toggle) {
+    g_metrics.recordQueueDrop();
+    g_metrics.recordBackoffSchedule();
+  }
+  g_metrics.recordLoopCall();
+}
+
+void reportTask() {
+  const ZeroTransportMetrics::Snapshot snapshot = g_metrics.snapshot();
+
+  Serial.print("connect_ok=");
+  Serial.print(snapshot.connectSuccesses);
+  Serial.print(" connect_fail=");
+  Serial.print(snapshot.connectFailures);
+  Serial.print(" send_ok=");
+  Serial.print(snapshot.sendSuccesses);
+  Serial.print(" send_fail=");
+  Serial.println(snapshot.sendFailures);
+}
+
+}  // namespace
+
+void setup() {
+  Serial.begin(115200);
+  delay(50);
+
+  ZeroKernel.begin();
+  ZeroKernel.addTask("MetricsTick", updateMetricsTask, 500, 0);
+  ZeroKernel.addTask("MetricsReport", reportTask, 1000, 0);
+}
+
+void loop() {
+  ZeroKernel.tick();
+}