feat(mam): Return UUID as the repsonse of send MAM message

accelerator/BUILD

@@ -22,6 +22,7 @@ cc_binary(
         ":ta_config",
         ":ta_cli",
         "//connectivity/http",
+        "//accelerator/core:periodical_task",
         "@org_iota_common//utils/handles:signal",
     ] + select({
         ":mqtt_enable": [

accelerator/cli_info.h

@@ -53,7 +53,7 @@ typedef enum ta_cli_arg_value_e {
   HEALTH_TRACK_PERIOD,
   NO_GTTA,
   BUFFER_LIST,
-  DONE_LIST,
+  COMPLETE_LIST,
   HTTP_THREADS_CLI,
   CACHE_CAPACITY,
 
@@ -95,7 +95,7 @@ static struct ta_cli_argument_s {
      "The period for checking IOTA full node host connection status"},
     {"no-gtta", no_argument, NULL, NO_GTTA, "Disable getTransactionToConfirm (gTTA) when sending transaction"},
     {"buffer_list", required_argument, NULL, BUFFER_LIST, "Set the value of `buffer_list_name`"},
-    {"done_list", required_argument, NULL, DONE_LIST, "Set the value of `done_list_name`"},
+    {"complete_list", required_argument, NULL, COMPLETE_LIST, "Specify the name of complete list`"},
     {"cache_capacity", required_argument, NULL, CACHE_CAPACITY, "Set the maximum capacity of caching server"},
     {"quiet", no_argument, NULL, QUIET, "Disable logger"},
     {"runtime_cli", no_argument, NULL, RUNTIME_CLI, "Enable runtime command line"},

accelerator/config.c

@@ -221,7 +221,7 @@ status_t cli_core_set(ta_core_t* const core, int key, char* const value) {
         ta_log_error("Malformed input character\n");
       }
       break;
-    case CACHE_MAX_CAPACITY:
+    case CACHE_CAPACITY:
       strtol_temp = strtol(value, NULL, 10);
       if (strtol_p != value && errno != ERANGE && strtol_temp >= INT_MIN && strtol_temp <= INT_MAX) {
         if (strtol_temp <= 0) {
@@ -265,8 +265,8 @@ status_t cli_core_set(ta_core_t* const core, int key, char* const value) {
     case BUFFER_LIST:
       cache->buffer_list_name = value;
       break;
-    case DONE_LIST:
-      cache->done_list_name = value;
+    case COMPLETE_LIST:
+      cache->complete_list_name = value;
       break;
 
     // Command line options configuration
@@ -342,8 +342,9 @@ status_t ta_core_default_init(ta_core_t* const core) {
   cache->host = REDIS_HOST;
   cache->port = REDIS_PORT;
   cache->state = false;
-  cache->buffer_list_name = BUFFER_LIST_NAME;
-  cache->done_list_name = DONE_LIST_NAME;
+  cache->complete_list_name = COMPLETE_LIST_NAME;
+  cache->mam_buffer_list_name = MAM_BUFFER_LIST_NAME;
+  cache->mam_complete_list_name = MAM_COMPLETE_LIST_NAME;
   cache->capacity = CACHE_MAX_CAPACITY;
 
   ta_log_info("Initializing IOTA full node configuration\n");

accelerator/config.h

@@ -71,7 +71,9 @@ extern "C" {
 #define DB_HOST "localhost"
 #define MAM_FILE_PREFIX "/tmp/mam_bin_XXXXXX"
 #define BUFFER_LIST_NAME "txn_buff_list"
-#define DONE_LIST_NAME "done_txn_buff_list"
+#define COMPLETE_LIST_NAME "complete_txn_buff_list"
+#define MAM_BUFFER_LIST_NAME "mam_buff_list"
+#define MAM_COMPLETE_LIST_NAME "complete_mam_buff_list"
 #define CACHE_MAX_CAPACITY \
   170 * 1024 * 1024              /**< Default cache server maximum capacity. It is set to 170MB by default. */
 #define HEALTH_TRACK_PERIOD 1800 /**< Check every half hour in default */
@@ -113,14 +115,16 @@ typedef struct iota_config_s {
 
 /** struct type of accelerator cache */
 typedef struct ta_cache_s {
-  char* host;               /**< Binding address of redis server */
-  uint64_t timeout;         /**< Timeout for keys in cache server */
-  char* buffer_list_name;   /**< Name of the list to buffer transactions */
-  char* done_list_name;     /**< Name of the list to store successfully broadcast transactions from buffer */
-  uint16_t port;            /**< Binding port of redis server */
-  bool state;               /**< Set it true to turn on cache server */
-  long int capacity;        /**< The maximum capacity of cache server */
-  pthread_rwlock_t* rwlock; /**< Read/Write lock to avoid data racing in buffering */
+  char* host;                   /**< Binding address of redis server */
+  uint64_t timeout;             /**< Timeout for keys in cache server */
+  char* buffer_list_name;       /**< Name of the list to buffer transactions */
+  char* complete_list_name;     /**< Name of the list to store successfully broadcast transactions from buffer */
+  char* mam_buffer_list_name;   /**< Name of the list to buffer MAM requests UUID */
+  char* mam_complete_list_name; /**< Name of the list to successfullay published MAM requests */
+  uint16_t port;                /**< Binding port of redis server */
+  bool state;                   /**< Set it true to turn on cache server */
+  long int capacity;            /**< The maximum capacity of cache server */
+  pthread_rwlock_t* rwlock;     /**< Read/Write lock to avoid data racing in buffering */
 } ta_cache_t;
 
 /** struct type of accelerator core */

accelerator/core/BUILD

@@ -87,3 +87,19 @@ cc_library(
         "@org_iota_common//utils:time",
     ],
 )
+
+cc_library(
+    name = "periodical_task",
+    srcs = ["periodical_task.c"],
+    hdrs = ["periodical_task.h"],
+    linkopts = [
+        "-lpthread",
+    ],
+    visibility = ["//visibility:public"],
+    deps = [
+        ":core",
+        ":mam_core",
+        "//common:ta_errors",
+        "//common:ta_logger",
+    ],
+)

accelerator/core/apis.c

@@ -214,32 +214,49 @@ status_t api_recv_mam_message(const iota_config_t* const iconf, const iota_clien
   return ret;
 }
 
-status_t api_send_mam_message(const ta_config_t* const info, const iota_config_t* const iconf,
-                              const iota_client_service_t* const service, char const* const obj, char** json_result) {
+status_t api_send_mam_message(const ta_cache_t* const cache, char const* const obj, char** json_result) {
   status_t ret = SC_OK;
   ta_send_mam_req_t* req = send_mam_req_new();
-  ta_send_mam_res_t* res = send_mam_res_new();
 
-  if (send_mam_message_req_deserialize(obj, req)) {
-    ret = SC_MAM_FAILED_INIT;
+  ret = send_mam_message_req_deserialize(obj, req);
+  if (ret) {
+    ta_log_error("%s\n", ta_error_to_string(ret));
+    goto done;
+  }
+
+  // Generate UUID. Each UUID would map to a processing request.
+  char uuid[UUID_STR_LEN];
+  uuid_t bin_uuid;
+  uuid_generate_random(bin_uuid);
+  uuid_unparse(bin_uuid, uuid);
+  if (!uuid[0]) {
+    ta_log_error("%s\n", "Failed to generate UUID");
+    goto done;
+  }
+
+  // TODO Generate the address that TA can quickly generate from the SEED with given parameters.
+
+  // Buffer send_mam_req_t object for publishing it later
+  ret = cache_set(uuid, UUID_STR_LEN - 1, obj, strlen(obj), cache->timeout);
+  if (ret != SC_OK) {
     ta_log_error("%s\n", ta_error_to_string(ret));
     goto done;
   }
 
-  ret = ta_send_mam_message(info, iconf, service, req, res);
+  // Buffer the UUID to the list
+  ret = cache_list_push(cache->mam_buffer_list_name, strlen(cache->mam_buffer_list_name), uuid, UUID_STR_LEN - 1);
   if (ret != SC_OK) {
     ta_log_error("%s\n", ta_error_to_string(ret));
     goto done;
   }
 
-  ret = send_mam_message_res_serialize(res, json_result);
+  ret = send_mam_message_res_serialize(NULL, uuid, json_result);
   if (ret != SC_OK) {
     ta_log_error("%s\n", ta_error_to_string(ret));
   }
 
 done:
   send_mam_req_free(&req);
-  send_mam_res_free(&res);
   return ret;
 }
 

accelerator/core/apis.h

@@ -75,8 +75,7 @@ status_t api_recv_mam_message(const iota_config_t* const iconf, const iota_clien
  * - SC_OK on success
  * - non-zero on error
  */
-status_t api_send_mam_message(const ta_config_t* const info, const iota_config_t* const iconf,
-                              const iota_client_service_t* const service, char const* const obj, char** json_result);
+status_t api_send_mam_message(const ta_cache_t* const cache, char const* const obj, char** json_result);
 
 /**
  * @brief Send transfer to tangle.