wip

accelerator/BUILD

@@ -21,6 +21,7 @@ cc_binary(
     deps = [
         ":ta_config",
         "//connectivity/http",
+        "//accelerator/core:background",
         "@org_iota_common//utils/handles:signal",
     ] + select({
         ":mqtt_enable": [

accelerator/config.c

@@ -302,6 +302,7 @@ status_t ta_core_default_init(ta_core_t* const core) {
   cache->state = false;
   cache->buffer_list_name = BUFFER_LIST_NAME;
   cache->done_list_name = DONE_LIST_NAME;
+  cache->buffer_mam_list_name = BUFFER_MAM_LIST_NAME;
   cache->capacity = CACHE_MAX_CAPACITY;
 
   ta_log_info("Initializing IOTA full node configuration\n");

accelerator/config.h

@@ -61,6 +61,7 @@ extern "C" {
 #define MAM_FILE_PREFIX "/tmp/mam_bin_XXXXXX"
 #define BUFFER_LIST_NAME "txn_buff_list"
 #define DONE_LIST_NAME "done_txn_buff_list"
+#define BUFFER_MAM_LIST_NAME "mam_buff_list"
 #define CACHE_MAX_CAPACITY 170 * 1024 * 1024  // default to 170MB
 #define HEALTH_TRACK_PERIOD 1800              // Check every half hour in default
 #define RESULT_SET_LIMIT 100  // The maximun returned transaction object number when querying transaction object by tag
@@ -98,14 +99,15 @@ 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* done_list_name;       /**< Name of the list to store successfully broadcast transactions from buffer */
+  char* buffer_mam_list_name; /**< Name of the list to buffer MAM requests UUID */
+  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,18 @@ cc_library(
         "@org_iota_common//utils:time",
     ],
 )
+
+cc_library(
+    name = "background",
+    srcs = ["background.c"],
+    hdrs = ["background.h"],
+    linkopts = [
+        "-lpthread",
+    ],
+    visibility = ["//visibility:public"],
+    deps = [
+        ":core",
+        "//common:ta_errors",
+        "//common:ta_logger",
+    ],
+)

accelerator/core/apis.c

@@ -220,31 +220,48 @@ status_t api_recv_mam_message(const iota_config_t* const iconf, const iota_clien
 }
 
 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) {
+                              const ta_cache_t* const cache, const iota_client_service_t* const service,
+                              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;
     ta_log_error("%s\n", ta_error_to_string(ret));
     goto done;
   }
 
-  ret = ta_send_mam_message(info, iconf, service, req, res);
+  // Generate UUID
+  char uuid[UUID_STR_LEN];
+  uuid_t binuuid;
+  uuid_generate_random(binuuid);
+  uuid_unparse(binuuid, uuid);
+  if (!uuid[0]) {
+    ta_log_error("%s\n", "Failed to generate UUID");
+    goto done;
+  }
+
+  // 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;
+  }
+
+  // Buffer the UUID to the list
+  ret = cache_list_push(cache->buffer_mam_list_name, strlen(cache->buffer_mam_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

@@ -91,7 +91,8 @@ status_t api_recv_mam_message(const iota_config_t* const iconf, const iota_clien
  * - 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);
+                              const ta_cache_t* const cache, const iota_client_service_t* const service,
+                              char const* const obj, char** json_result);
 
 /**
  * @brief Send transfer to tangle.