feat: Construct MQTT interface of TA

Construct the rough structure of MQTT interface for tangle-accelerator. The main function is in `mqtt_main.c`.
Callback functions of MQTT client locate in `duplex_callback.[c, h]`, and more important the `mqtt_request_handler()`
locates in `duplex_callback.c` as well.

Other functions for `main()` function locate in `client_common.[c, h]`

CAUTION: The current version is single thread. To support multiple thread, it necessitates reformation of the code.

Doxyfile

@@ -8,7 +8,8 @@ INPUT                  = . \
                         request \
                         response \
                         serializer \
-                        utils
+                        utils \ 
+                        connectivity/mqtt
 FILE_PATTERNS          = *.h \
                         *.md
 EXAMPLE_PATH           = tests

accelerator/BUILD

@@ -28,11 +28,15 @@ cc_binary(
 
 cc_binary(
     name = "accelerator_mqtt",
-    srcs = ["mqtt_main.c"],
+    srcs = ["mqtt_interface.c"],
+    copts = [
+        "-DLOGGER_ENABLE",
+        "-DENABLE_MQTT",
+    ],
     deps = [
         ":ta_config",
         ":ta_errors",
-        "//mqtt_utils",
+        "//connectivity/mqtt:mqtt_utils",
         "@entangled//utils/handles:signal",
     ],
 )

accelerator/config.h

@@ -28,6 +28,10 @@ extern "C" {
 
 #define TA_VERSION "tangle-accelerator/0.5.0"
 #define TA_HOST "localhost"
+#ifdef ENABLE_MQTT
+#define MQTT_HOST "localhost"
+#define TOPIC_ROOT "NB/root/topics"
+#endif
 #define TA_PORT "8000"
 #define TA_THREAD_COUNT 10
 #define IRI_HOST "localhost"

accelerator/errors.h

@@ -54,6 +54,7 @@ extern "C" {
 #define SC_MODULE_CONF (0x07 << SC_MODULE_SHIFT)
 #define SC_MODULE_UTILS (0x08 << SC_MODULE_SHIFT)
 #define SC_MODULE_HTTP (0x09 << SC_MODULE_SHIFT)
+#define SC_MODULE_MQTT (0x0A << SC_MODULE_SHIFT)
 /** @} */
 
 /** @name serverity code */
@@ -168,6 +169,28 @@ typedef enum {
   /**< URL doesn't match regular expression rule */
   SC_HTTP_URL_PARSE_ERROR = 0x05 | SC_MODULE_HTTP | SC_SEVERITY_MAJOR,
   /**< URL parameter parsing error */
+
+  // MQTT module
+  SC_MQTT_OOM = 0x01 | SC_MODULE_MQTT | SC_SEVERITY_FATAL,
+  /**< Fail to create MQTT object */
+  SC_MQTT_NULL = 0x02 | SC_MODULE_MQTT | SC_SEVERITY_FATAL,
+  /**< NULL object in MQTT */
+  SC_MQTT_INVALID_REGEX = 0x03 | SC_MODULE_MQTT | SC_SEVERITY_MAJOR,
+  /**< Invalid URL regular expression rule in MQTT */
+  SC_MQTT_TOPIC_NOT_MATCH = 0x04 | SC_MODULE_MQTT | SC_SEVERITY_MAJOR,
+  /**< Topic doesn't match regular expression rule */
+  SC_MQTT_URL_PARSE_ERROR = 0x05 | SC_MODULE_MQTT | SC_SEVERITY_MAJOR,
+  /**< Topic parameter parsing error */
+  SC_MQTT_INIT = 0x06 | SC_MODULE_MQTT | SC_SEVERITY_MAJOR,
+  /**< Error during initialization*/
+  SC_MOSQ_OBJ_INIT_ERROR = 0x07 | SC_MODULE_MQTT | SC_SEVERITY_MAJOR,
+  /**< Error in initializing mosquitto object */
+  SC_MQTT_TOPIC_SET = 0x08 | SC_MODULE_MQTT | SC_SEVERITY_MAJOR,
+  /**< Error in setting topic */
+  SC_MQTT_OPT_SET = 0x09 | SC_MODULE_MQTT | SC_SEVERITY_MAJOR,
+  /**< Error in setting options of `struct mosquitto` object */
+  SC_CLIENT_CONNTECT = 0x0A | SC_MODULE_MQTT | SC_SEVERITY_MAJOR,
+  /**< Error in connecting to broker */
 } status_t;
 
 typedef enum {

accelerator/mqtt_interface.c

@@ -0,0 +1,47 @@
+#include "config.h"
+#include "connectivity/mqtt/client_common.h"
+#include "connectivity/mqtt/duplex_callback.h"
+#include "connectivity/mqtt/duplex_utils.h"
+#include "errors.h"
+
+int main(int argc, char *argv[]) {
+  status_t ret;
+  mosq_config_t cfg;
+  struct mosquitto *mosq = NULL;
+
+  // Initialize `mosq` and `cfg`
+  // if we want to opertate this program under multi-threading, see https://github.com/eclipse/mosquitto/issues/450
+  ret = duplex_config_init(&mosq, &cfg);
+  if (ret != SC_OK) {
+    goto done;
+  }
+
+  // Set callback functions
+  duplex_callback_func_set(mosq);
+
+  // The following one line is used for testing if this server work fine with requests with given topics.
+  // Uncomment it if it is necessitated
+  // gossip_channel_set(&cfg, MQTT_HOST, "NB/test/room1", "NB/test/room2");
+
+  // Set the configures and message for testing
+  ret = gossip_api_channels_set(&cfg, MQTT_HOST, TOPIC_ROOT);
+  if (ret != SC_OK) {
+    goto done;
+  }
+
+  // Set cfg as `userdata` field of `mosq` which allows the callback functions to use `cfg`.
+  mosquitto_user_data_set(mosq, &cfg);
+
+  // Start listening subscribing topics, once we received a message from the listening topics, we can send corresponding
+  // message.
+  // if we need to take the above task forever, just put it in a infinite loop.
+  do {
+    ret = duplex_client_start(mosq, &cfg);
+  } while (!ret);
+
+done:
+  mosquitto_destroy(mosq);
+  mosquitto_lib_cleanup();
+  mosq_config_free(&cfg);
+  return ret;
+}

mqtt_utils/BUILD → connectivity/mqtt/BUILD

@@ -9,7 +9,10 @@ cc_library(
         "duplex_utils.h",
     ],
     visibility = ["//visibility:public"],
-    deps = [":mqtt_common"],
+    deps = [
+        ":mqtt_common",
+        "//accelerator:ta_errors",
+    ],
 )
 
 cc_library(
@@ -24,5 +27,9 @@ cc_library(
         "pub_utils.h",
         "sub_utils.h",
     ],
-    deps = ["//third_party:mosquitto"],
+    deps = [
+        "//accelerator:ta_config",
+        "//accelerator:ta_errors",
+        "//third_party:mosquitto",
+    ],
 )