MQTT/Protobuf Middleware API

.gitmodules

@@ -2,3 +2,6 @@
 [submodule "dev/OpenOCD"]
 	path = dev/OpenOCD
 	url = https://github.com/STMicroelectronics/OpenOCD.git
+[submodule "dev/nanopb/nanopb"]
+	path = dev/nanopb/nanopb
+	url = https://github.com/nanopb/nanopb.git

NetX/inc/u_nx_protobuf.h

@@ -0,0 +1,83 @@
+#pragma once
+
+// clang-format off
+
+/*
+*   Wrapper for structuring and sending protobuf Ethernet messages over MQTT. 
+*   This API sits above the lower-level u_nx_ethernet.h driver layer. Messages with nonstandard formats can still be sent using the u_nx_ethernet.h API directly.
+*/
+
+#include <stddef.h>
+#include <stdbool.h>
+#include "serverdata.pb.h"
+
+/* Helper macros. */
+#define PB_STR_HELPER(x) #x                                                   // Helper for PB_TOSTR(). Probably should never use directly.
+#define PB_TOSTR(x) PB_STR_HELPER(x)                                          // Converts a macro's value into a string.
+#define PB_COUNT_ARGS(...) (sizeof((float[]){ __VA_ARGS__ }) / sizeof(float)) // Returns the number of arguments passed into it.
+#define PB_STR_LEN(s) (sizeof(s) - 1)                                         // Returns the length of a string literal.
+
+/* CONFIG: Compile-time validation of topic size, unit size, and number of values. */
+#define PB_MAX_TOPIC_LENGTH 100 // Maximum length of topic string literal (in characters).
+#define PB_MAX_UNIT_LENGTH  15  // Maximum length of unit string literal (in characters).
+#define PB_MIN_DATAPOINTS   1   // Minimum number of datapoints (i.e., variable `...` arguments passed into `nx_protobuf_mqtt_message_create()`).
+#define PB_MAX_DATAPOINTS   5   // Maximum number of datapoints (i.e., variable `...` arguments passed into `nx_protobuf_mqtt_message_create()`).
+#define PB_VALIDATE_ARGS(topic, unit, num_values)                                                                                             \
+    do {                                                                                                                                      \
+        _Static_assert(                                                                                                                       \
+            PB_STR_LEN(topic) <= PB_MAX_TOPIC_LENGTH,                                                                                         \
+            "MQTT topic parameter exceeds maximum length of " PB_TOSTR(PB_MAX_TOPIC_LENGTH) " allowed by `nx_protobuf_mqtt_message_create()`."\
+        );                                                                                                                                    \
+        _Static_assert(                                                                                                                       \
+            PB_STR_LEN(unit) <= PB_MAX_UNIT_LENGTH,                                                                                           \
+            "MQTT unit parameter exceeds maximum length of " PB_TOSTR(PB_MAX_UNIT_LENGTH) " allowed by `nx_protobuf_mqtt_message_create()`."  \
+        );                                                                                                                                    \
+        _Static_assert(                                                                                                                       \
+            (num_values) >= PB_MIN_DATAPOINTS,                                                                                                \
+            "Must pass at least " PB_TOSTR(PB_MIN_DATAPOINTS) " value into the variable argument of `nx_protobuf_mqtt_message_create()`."     \
+        );                                                                                                                                    \
+        _Static_assert(                                                                                                                       \
+            (num_values) <= PB_MAX_DATAPOINTS,                                                                                                \
+            "Cannot pass more than " PB_TOSTR(PB_MAX_DATAPOINTS) " values into the variable argument of `nx_protobuf_mqtt_message_create()`." \
+        );                                                                                                                                    \
+    } while (0)
+
+/**
+ * @brief Creates and formats a `ethernet_mqtt_message_t` object, and returns it to the caller.
+ * @param topic (const char*) String literal representing the message's MQTT topic name.
+ * @param unit (const char*) String literal representing the unit of the message's data.
+ * @param ... (float) The data to be sent in the message. This is a variable argument, so it can be repeated depending on how many datapoints you want to send. If you pass in more datapoints than allowed, you will get a compile-time error.
+ * @return An `ethernet_mqtt_message_t` object. 
+ * @note If message creation was not completed for any reason, .initialized will be false in the returned `ethernet_mqtt_message_t` object. You may still use the object as you please (including attempting to initialize it again), but attempting to send the message (via `nx_protobuf_mqtt_message_send()`) will return an error.
+ */
+#define nx_protobuf_mqtt_message_create(topic, unit, ...)           \
+    ({                                                              \
+        PB_VALIDATE_ARGS(topic, unit, PB_COUNT_ARGS(__VA_ARGS__));  \
+        _nx_protobuf_mqtt_message_create(                           \
+            (topic), PB_STR_LEN(topic),                             \
+            (unit),  PB_STR_LEN(unit),                              \
+            (float[]){ __VA_ARGS__ },                               \
+            PB_COUNT_ARGS(__VA_ARGS__)                              \
+        );                                                          \
+    })
+
+
+/* Ethernet MQTT Message. */
+typedef struct {
+    const char* topic;
+    int topic_size;
+    serverdata_v2_ServerData protobuf;
+    bool initialized;
+} ethernet_mqtt_message_t;
+
+/**
+ * @brief Dispatches a `ethernet_mqtt_message_t` message over MQTT.
+ * @param message The message to send.
+ * @return U_SUCCESS if successful, U_ERROR is not successful.
+ */
+int nx_protobuf_mqtt_message_send(ethernet_mqtt_message_t* message);
+
+/* MACRO IMPLEMENTATIONS */
+ethernet_mqtt_message_t _nx_protobuf_mqtt_message_create(const char* topic, size_t topic_size, const char* unit,  size_t unit_len, const float values[], int values_count);
+
+// clang-format on

NetX/src/u_nx_debug.c

@@ -1,6 +1,17 @@
 #include "u_nx_debug.h"
 #include "nxd_ptp_client.h"
 
+#if defined(__has_include)
+#if __has_include("nxd_mqtt_client.h")
+#include "nxd_mqtt_client.h"
+#define U_NX_DEBUG_HAS_MQTT 1
+#endif
+#endif
+
+#ifndef U_NX_DEBUG_HAS_MQTT
+#define U_NX_DEBUG_HAS_MQTT 0
+#endif
+
 // clang-format off
 
 /* Converts a NetX status macro to a printable string. */
@@ -74,8 +85,11 @@ const char* nx_status_toString(UINT status) {
         case NX_CONTINUE: return "NX_CONTINUE";
         case NX_TCPIP_OFFLOAD_ERROR: return "NX_TCPIP_OFFLOAD_ERROR";
 
-        /* MQTT-specific stuff. */
+    /* MQTT-specific stuff. */
+#if U_NX_DEBUG_HAS_MQTT
+#if (NXD_MQTT_SUCCESS != NX_SUCCESS)
         case NXD_MQTT_SUCCESS: return "NXD_MQTT_SUCCESS";
+#endif
         case NXD_MQTT_ALREADY_CONNECTED: return "NXD_MQTT_ALREADY_CONNECTED";
         case NXD_MQTT_NOT_CONNECTED: return "NXD_MQTT_NOT_CONNECTED";
         case NXD_MQTT_MUTEX_FAILURE: return "NXD_MQTT_MUTEX_FAILURE";
@@ -100,6 +114,7 @@ const char* nx_status_toString(UINT status) {
         case NXD_MQTT_ERROR_SERVER_UNAVAILABLE: return "NXD_MQTT_ERROR_SERVER_UNAVAILABLE";
         case NXD_MQTT_ERROR_BAD_USERNAME_PASSWORD: return "NXD_MQTT_ERROR_BAD_USERNAME_PASSWORD";
         case NXD_MQTT_ERROR_NOT_AUTHORIZED: return "NXD_MQTT_ERROR_NOT_AUTHORIZED";
+#endif
 
         /* PTP-specific stuff. */
         case NX_PTP_CLIENT_NOT_STARTED: return "NX_PTP_CLIENT_NOT_STARTED";

NetX/src/u_nx_ethernet.c

@@ -382,6 +382,7 @@ UINT ethernet_init(ethernet_node_t node_id, DriverFunction driver, OnRecieve on_
     /* Mark device as initialized. */
     device.is_initialized = true;
 
+    PRINTLN_INFO("Ran ethernet_init()");
     return NX_SUCCESS;
 }
 

NetX/src/u_nx_protobuf.c

@@ -0,0 +1,116 @@
+#include <string.h>
+#include "u_tx_debug.h"
+#include "u_nx_debug.h"
+#include "u_nx_protobuf.h"
+#include "u_nx_ethernet.h"
+#include "pb.h"
+#include "pb_encode.h"
+#include "tx_api.h"
+#include "nxd_mqtt_client.h"
+
+ethernet_mqtt_message_t _nx_protobuf_mqtt_message_create(const char* topic, size_t topic_len, const char* unit,  size_t unit_len, const float values[], int values_size) {
+    /* Zero-initialize the protobuf struct and the sendable ethernet_mqtt_message_t message. */
+    serverdata_v2_ServerData protobuf =  serverdata_v2_ServerData_init_zero;
+    ethernet_mqtt_message_t message = { 0 };
+    message.initialized = false;
+
+    /* Enforce topic length. */
+    if(topic_len > PB_MAX_TOPIC_LENGTH) {
+        PRINTLN_ERROR("MQTT Message topic exceeds maximum length of %d (Topic: %s, Current Topic Length: %d).", PB_MAX_TOPIC_LENGTH, topic, topic_len);
+        return message; // Return empty, uninitialized message.
+    }
+    // NOTE: If using the `nx_protobuf_mqtt_message_create()` macro (as intended), the static asserts should catch this. This is just an extra check in case a caller uses this function directly for whatever reason.
+
+    /* Enforce unit length. */
+    if(unit_len > PB_MAX_UNIT_LENGTH) {
+        PRINTLN_ERROR("MQTT Unit string length exceeds maximum length of %d (Topic: %s, Current Unit String Length: %d).", PB_MAX_UNIT_LENGTH, topic, unit_len);
+        return message; // Return empty, uninitialized message.
+    }
+    // NOTE: If using the `nx_protobuf_mqtt_message_create()` macro (as intended), the static asserts should catch this. This is just an extra check in case a caller uses this function directly for whatever reason.
+
+    /* Enforce minimum number of datapoints. */
+    if(values_size < PB_MIN_DATAPOINTS) {
+        PRINTLN_ERROR("Message must have at least %d datapoints (Topic: %s, Current values_size: %d).", PB_MIN_DATAPOINTS, topic, values_size);
+        return message; // Return empty, uninitialized message.
+    }
+    // NOTE: If using the `nx_protobuf_mqtt_message_create()` macro (as intended), the static asserts should catch this. This is just an extra check in case a caller uses this function directly for whatever reason.
+
+    /* Enforce maximum number of datapoints. */
+    if(values_size > PB_MAX_DATAPOINTS) {
+        PRINTLN_ERROR("Message cannot have more than %d datapoints (Topic: %s, Current values_size: %d).", PB_MAX_DATAPOINTS, topic, values_size);
+        return message; // Return empty, uninitialized message.
+    }
+    // NOTE: If using the `nx_protobuf_mqtt_message_create()` macro (as intended), the static asserts should catch this. This is just an extra check in case a caller uses this function directly for whatever reason.
+
+    /* Get the PTP time and convert to appropriate protobuf time. */
+    //NX_PTP_DATE_TIME datetime = ethernet_get_time(); u_TODO - for some reason ethernet_get_time() is blocking! kind of weird. it doesn't always block either, just when called from certain areas.
+    uint64_t time_us = 10; // u_TODO - obviously this is temporary, but can't use ethernet_get_time() since it's blocking
+
+    /* Pack the protobuf message. */
+    // CURRENT PROTOBUF SCHEMA LOOKS LIKE THIS:
+    // typedef struct _serverdata_v2_ServerData {
+    //     char unit[15];
+    //     uint64_t time_us;
+    //     pb_size_t values_count;
+    //     float values[5];
+    // } serverdata_v2_ServerData;
+    memcpy(protobuf.unit, unit, unit_len);
+    protobuf.time_us = time_us;
+    protobuf.values_count = values_size;
+    memcpy(protobuf.values, values, values_size * sizeof(float));
+
+    /* Pack the `ethernet_mqtt_message_t` object and return it as successfully initialized. */
+    message.topic = topic;
+    message.topic_size = topic_len + 1; // u_TODO - for some reason you need to do + 1 here or else it will cut the last letter off of the topic in MQTT ui. The macro probably just calculates the topic length with one less than it should or something
+    message.protobuf = protobuf;
+    message.initialized = true;
+    return message;
+}
+
+int nx_protobuf_mqtt_message_send(ethernet_mqtt_message_t* message) {
+    /* Make sure message isn't nullptr. */
+    if(!message) {
+        PRINTLN_ERROR("Null pointer to `ethernet_mqtt_message_t` message.");
+        return U_ERROR;
+    }
+
+    /* Make sure message is initialized. */
+    if(!message->initialized) {
+        PRINTLN_ERROR("Attempting to send an uninitialized `ethernet_mqtt_message_t` message.");
+        return U_ERROR;
+    }
+
+    /* Set up the buffer. */
+    unsigned char buffer[serverdata_v2_ServerData_size];
+    pb_ostream_t stream = pb_ostream_from_buffer(buffer, sizeof(buffer));
+
+    /* Encode the protobuf. */
+    int status = pb_encode(&stream, serverdata_v2_ServerData_fields, &message->protobuf);
+    if(status != true) {
+        PRINTLN_ERROR("Failed to serialize protobuf message (Topic: %s): %s", message->topic, PB_GET_ERROR(&stream));
+        return U_ERROR;
+    }
+
+    /* Publish over MQTT. */
+    status = ethernet_mqtt_publish(message->topic, message->topic_size, (char*)buffer, stream.bytes_written); // u_TODO - ethernet_mqtt_publish should return U_SUCCESS/U_ERROR instead of the internal netx error macro
+    if(status != NXD_MQTT_SUCCESS) {
+        PRINTLN_WARNING("Failed to publish MQTT message (Topic: %s, Status: %d).", message->topic, status);
+
+        /* If disconnected, attempt reconnection. */
+        if(status == NXD_MQTT_NOT_CONNECTED) {
+            PRINTLN_WARNING("Detected disconnect from MQTT. Attempting reconnection...");
+            do {
+                tx_thread_sleep(1000);
+                status = ethernet_mqtt_reconnect();
+                PRINTLN_WARNING("Attempting MQTT reconnection (Status: %d/%s).", status, nx_status_toString(status));
+            } while ((status != NXD_MQTT_SUCCESS) && (status != NXD_MQTT_ALREADY_CONNECTED));
+            PRINTLN_WARNING("MQTT reconnection successful.");
+        }
+
+        return U_ERROR;
+    }
+
+    /* Return successful! */
+    PRINTLN_INFO("Sent MQTT message (Topic: %s).", message->topic);
+    return U_SUCCESS;
+}

dev/nanopb/CMakeLists.txt

@@ -0,0 +1,8 @@
+cmake_minimum_required(VERSION 3.22)
+
+set(CMAKE_MODULE_PATH ${CMAKE_CURRENT_SOURCE_DIR}/nanopb/extra)
+find_package(Nanopb REQUIRED)
+
+nanopb_generate_cpp(TARGET proto serverdata.proto)
+
+target_link_libraries(${CMAKE_PROJECT_NAME} proto)

dev/nanopb/serverdata.options

@@ -0,0 +1,2 @@
+serverdata.v2.ServerData.unit             max_size: 15
+serverdata.v2.ServerData.values           max_count: 5

dev/nanopb/serverdata.proto

@@ -0,0 +1,14 @@
+syntax = "proto3";
+
+package serverdata.v2;
+
+message ServerData {
+  // ensure old type is reserved
+  reserved 1;
+  reserved "value";
+
+  string unit = 2;
+  // time since unix epoch in MICROSECONDS
+  uint64 time_us = 3;
+  repeated float values = 4;
+}