feat(driver): support tag name duplication in different groups

include/neuron/neuron.h

@@ -28,27 +28,34 @@ extern "C" {
 #include <stdint.h>
 #include <stdlib.h>
 
-#include "adapter.h"
 #include "errcodes.h"
-#include "file.h"
-#include "plugin.h"
-#include "tag.h"
+
+#include "utils/async_queue.h"
+#include "utils/protocol_buf.h"
+
 #include "utils/base64.h"
+#include "utils/log.h"
+#include "utils/zlog.h"
+
+#include "utils/utarray.h"
+#include "utils/utextend.h"
+#include "utils/uthash.h"
+#include "utils/utlist.h"
 
 #include "connection/mqtt_client_intf.h"
 #include "connection/neu_connection.h"
 #include "event/event.h"
-
 #include "json/neu_json_param.h"
 
-#include "utils/async_queue.h"
-#include "utils/log.h"
-#include "utils/protocol_buf.h"
-#include "utils/utextend.h"
-
 #include "define.h"
+
+#include "adapter.h"
+#include "plugin.h"
+#include "tag.h"
 #include "tag_sort.h"
 
+#include "file.h"
+
 #ifdef __cplusplus
 }
 #endif

plugins/modbus/modbus_req.c

@@ -22,6 +22,7 @@
 #include "modbus_req.h"
 
 struct modbus_group_data {
+    char *                  group;
     UT_array *              tags;
     modbus_read_cmd_sort_t *cmd_sort;
 };
@@ -75,6 +76,7 @@ int modbus_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group,
             utarray_push_back(gd->tags, &p);
         }
 
+        gd->group    = strdup(group->group_name);
         gd->cmd_sort = modbus_tag_sort(gd->tags, max_byte);
     } else {
         struct modbus_group_data *gd =
@@ -251,8 +253,8 @@ int modbus_value_handle(void *ctx, uint8_t slave_id, uint16_t n_byte,
             }
         }
 
-        plugin->common.adapter_callbacks->driver.update(plugin->common.adapter,
-                                                        (*p_tag)->name, dvalue);
+        plugin->common.adapter_callbacks->driver.update(
+            plugin->common.adapter, gd->group, (*p_tag)->name, dvalue);
     }
     return 0;
 }

src/adapter/driver/cache.c

@@ -18,9 +18,11 @@
  **/
 
 #include <assert.h>
-#include <pthread.h>
 #include <string.h>
 
+#include <nng/nng.h>
+#include <nng/supplemental/util/platform.h>
+
 #include "utils/uthash.h"
 
 #include "define.h"
@@ -28,6 +30,11 @@
 
 #include "cache.h"
 
+typedef struct {
+    char group[NEU_GROUP_NAME_LEN];
+    char tag[NEU_TAG_NAME_LEN];
+} tkey_t;
+
 struct elem {
     int32_t error;
     int64_t timestamp;
@@ -35,21 +42,31 @@ struct elem {
     uint16_t n_byte;
     uint8_t *bytes;
 
-    char           name[NEU_TAG_NAME_LEN];
+    tkey_t         key;
     UT_hash_handle hh;
 };
 
 struct neu_driver_cache {
-    pthread_mutex_t mtx;
+    nng_mtx *mtx;
 
     struct elem *table;
 };
 
+inline static tkey_t to_key(const char *group, const char *tag)
+{
+    tkey_t key = { 0 };
+
+    strcpy(key.group, group);
+    strcpy(key.tag, tag);
+
+    return key;
+}
+
 neu_driver_cache_t *neu_driver_cache_new()
 {
     neu_driver_cache_t *cache = calloc(1, sizeof(neu_driver_cache_t));
 
-    pthread_mutex_init(&cache->mtx, NULL);
+    nng_mtx_alloc(&cache->mtx);
 
     return cache;
 }
@@ -59,7 +76,7 @@ void neu_driver_cache_destroy(neu_driver_cache_t *cache)
     struct elem *elem = NULL;
     struct elem *tmp  = NULL;
 
-    pthread_mutex_lock(&cache->mtx);
+    nng_mtx_lock(cache->mtx);
     HASH_ITER(hh, cache->table, elem, tmp)
     {
         HASH_DEL(cache->table, elem);
@@ -68,48 +85,57 @@ void neu_driver_cache_destroy(neu_driver_cache_t *cache)
         }
         free(elem);
     }
-    pthread_mutex_unlock(&cache->mtx);
+    nng_mtx_unlock(cache->mtx);
 
-    pthread_mutex_destroy(&cache->mtx);
+    nng_mtx_free(cache->mtx);
 
     free(cache);
 }
 
-void neu_driver_cache_error(neu_driver_cache_t *cache, const char *key,
-                            int64_t timestamp, int32_t error)
+void neu_driver_cache_error(neu_driver_cache_t *cache, const char *group,
+                            const char *tag, int64_t timestamp, int32_t error)
 {
     struct elem *elem = NULL;
+    tkey_t       key  = to_key(group, tag);
 
-    pthread_mutex_lock(&cache->mtx);
-    HASH_FIND_STR(cache->table, key, elem);
+    nng_mtx_lock(cache->mtx);
+    HASH_FIND(hh, cache->table, &key, sizeof(tkey_t), elem);
 
     if (elem == NULL) {
         elem = calloc(1, sizeof(struct elem));
-        strncpy(elem->name, key, sizeof(elem->name));
-        HASH_ADD_STR(cache->table, name, elem);
+        strcpy(elem->key.group, group);
+        strcpy(elem->key.tag, tag);
+
+        HASH_ADD(hh, cache->table, key, sizeof(tkey_t), elem);
     }
     elem->timestamp = timestamp;
     elem->error     = error;
 
-    pthread_mutex_unlock(&cache->mtx);
+    nng_mtx_unlock(cache->mtx);
 }
 
-void neu_driver_cache_update(neu_driver_cache_t *cache, const char *key,
-                             int64_t timestamp, uint16_t n_byte, uint8_t *bytes)
+void neu_driver_cache_update(neu_driver_cache_t *cache, const char *group,
+                             const char *tag, int64_t timestamp,
+                             uint16_t n_byte, uint8_t *bytes)
 {
     struct elem *elem = NULL;
+    tkey_t       key  = to_key(group, tag);
 
-    pthread_mutex_lock(&cache->mtx);
-    HASH_FIND_STR(cache->table, key, elem);
+    nng_mtx_lock(cache->mtx);
+    HASH_FIND(hh, cache->table, &key, sizeof(tkey_t), elem);
 
     if (elem == NULL) {
         elem = calloc(1, sizeof(struct elem));
-        strncpy(elem->name, key, sizeof(elem->name));
+
+        strcpy(elem->key.group, group);
+        strcpy(elem->key.tag, tag);
+
         assert(n_byte != 0);
         assert(n_byte <= NEU_VALUE_SIZE);
         elem->n_byte = n_byte;
         elem->bytes  = calloc(elem->n_byte, 1);
-        HASH_ADD_STR(cache->table, name, elem);
+
+        HASH_ADD(hh, cache->table, key, sizeof(tkey_t), elem);
     }
 
     elem->error     = 0;
@@ -130,17 +156,18 @@ void neu_driver_cache_update(neu_driver_cache_t *cache, const char *key,
 
     memcpy(elem->bytes, bytes, elem->n_byte);
 
-    pthread_mutex_unlock(&cache->mtx);
+    nng_mtx_unlock(cache->mtx);
 }
 
-int neu_driver_cache_get(neu_driver_cache_t *cache, const char *key,
-                         neu_driver_cache_value_t *value)
+int neu_driver_cache_get(neu_driver_cache_t *cache, const char *group,
+                         const char *tag, neu_driver_cache_value_t *value)
 {
     struct elem *elem = NULL;
     int          ret  = -1;
+    tkey_t       key  = to_key(group, tag);
 
-    pthread_mutex_lock(&cache->mtx);
-    HASH_FIND_STR(cache->table, key, elem);
+    nng_mtx_lock(cache->mtx);
+    HASH_FIND(hh, cache->table, &key, sizeof(tkey_t), elem);
 
     if (elem != NULL) {
         value->error     = elem->error;
@@ -151,17 +178,19 @@ int neu_driver_cache_get(neu_driver_cache_t *cache, const char *key,
         ret = 0;
     }
 
-    pthread_mutex_unlock(&cache->mtx);
+    nng_mtx_unlock(cache->mtx);
 
     return ret;
 }
 
-void neu_driver_cache_del(neu_driver_cache_t *cache, const char *key)
+void neu_driver_cache_del(neu_driver_cache_t *cache, const char *group,
+                          const char *tag)
 {
     struct elem *elem = NULL;
+    tkey_t       key  = to_key(group, tag);
 
-    pthread_mutex_lock(&cache->mtx);
-    HASH_FIND_STR(cache->table, key, elem);
+    nng_mtx_lock(cache->mtx);
+    HASH_FIND(hh, cache->table, &key, sizeof(tkey_t), elem);
 
     if (elem != NULL) {
         HASH_DEL(cache->table, elem);
@@ -171,5 +200,5 @@ void neu_driver_cache_del(neu_driver_cache_t *cache, const char *key)
         free(elem);
     }
 
-    pthread_mutex_unlock(&cache->mtx);
+    nng_mtx_unlock(cache->mtx);
 }

src/adapter/driver/cache.h

@@ -29,12 +29,13 @@ typedef struct neu_driver_cache neu_driver_cache_t;
 neu_driver_cache_t *neu_driver_cache_new();
 void                neu_driver_cache_destroy(neu_driver_cache_t *cache);
 
-void neu_driver_cache_error(neu_driver_cache_t *cache, const char *key,
-                            int64_t timestamp, int32_t error);
-void neu_driver_cache_update(neu_driver_cache_t *cache, const char *key,
-                             int64_t timestamp, uint16_t n_byte,
-                             uint8_t *bytes);
-void neu_driver_cache_del(neu_driver_cache_t *cache, const char *key);
+void neu_driver_cache_error(neu_driver_cache_t *cache, const char *group,
+                            const char *tag, int64_t timestamp, int32_t error);
+void neu_driver_cache_update(neu_driver_cache_t *cache, const char *group,
+                             const char *tag, int64_t timestamp,
+                             uint16_t n_byte, uint8_t *bytes);
+void neu_driver_cache_del(neu_driver_cache_t *cache, const char *group,
+                          const char *tag);
 
 typedef struct {
     uint16_t    n_byte;
@@ -44,7 +45,7 @@ typedef struct {
     int64_t timestamp;
 } neu_driver_cache_value_t;
 
-int neu_driver_cache_get(neu_driver_cache_t *cache, const char *key,
-                         neu_driver_cache_value_t *value);
+int neu_driver_cache_get(neu_driver_cache_t *cache, const char *group,
+                         const char *tag, neu_driver_cache_value_t *value);
 
 #endif

src/adapter/driver/driver.c

@@ -61,12 +61,12 @@ struct neu_adapter_driver {
 
 static int       report_callback(void *usr_data);
 static int       read_callback(void *usr_data);
-static void      read_group(neu_driver_cache_t *cache, UT_array *tags,
-                            neu_resp_tag_value_t *datas);
+static void      read_group(neu_driver_cache_t *cache, const char *group,
+                            UT_array *tags, neu_resp_tag_value_t *datas);
 static UT_array *load_tags(UT_array *tags);
 static void      free_tags(UT_array *tags);
-static void      update(neu_adapter_t *adapter, const char *name,
-                        neu_dvalue_t value);
+static void update(neu_adapter_t *adapter, const char *group, const char *tag,
+                   neu_dvalue_t value);
 static void write_response(neu_adapter_t *adapter, void *r, neu_error error);
 static group_t *find_group(neu_adapter_driver_t *driver, const char *name);
 static void     write_response(neu_adapter_t *adapter, void *r, neu_error error)
@@ -79,7 +79,8 @@ static void     write_response(neu_adapter_t *adapter, void *r, neu_error error)
     adapter->cb_funs.response(adapter, req, &nerror);
 }
 
-static void update(neu_adapter_t *adapter, const char *name, neu_dvalue_t value)
+static void update(neu_adapter_t *adapter, const char *group, const char *tag,
+                   neu_dvalue_t value)
 {
     neu_adapter_driver_t *driver = (neu_adapter_driver_t *) adapter;
     uint16_t              n_byte = 0;
@@ -118,9 +119,9 @@ static void update(neu_adapter_t *adapter, const char *name, neu_dvalue_t value)
     }
 
     if (value.type == NEU_TYPE_ERROR) {
-        neu_driver_cache_error(driver->cache, name, 0, value.value.i32);
+        neu_driver_cache_error(driver->cache, group, tag, 0, value.value.i32);
     } else {
-        neu_driver_cache_update(driver->cache, name, 0, n_byte,
+        neu_driver_cache_update(driver->cache, group, tag, 0, n_byte,
                                 value.value.bytes);
     }
 }
@@ -204,7 +205,7 @@ void neu_adapter_driver_read_group(neu_adapter_driver_t *driver,
     resp.n_tag = utarray_len(tags);
     resp.tags  = calloc(1, utarray_len(tags) * sizeof(neu_resp_tag_value_t));
 
-    read_group(driver->cache, tags, resp.tags);
+    read_group(driver->cache, cmd->group, tags, resp.tags);
     strcpy(resp.driver, cmd->driver);
     strcpy(resp.group, cmd->group);
 
@@ -328,7 +329,7 @@ int neu_adapter_driver_del_group(neu_adapter_driver_t *driver, const char *name)
 
         utarray_foreach(find->grp.tags, neu_datatag_t **, p_tag)
         {
-            neu_driver_cache_del(driver->cache, (*p_tag)->name);
+            neu_driver_cache_del(driver->cache, name, (*p_tag)->name);
         }
 
         free_tags(find->grp.tags);
@@ -516,7 +517,8 @@ static int report_callback(void *usr_data)
     strcpy(data->driver, neu_adapter_get_name((neu_adapter_t *) group->driver));
     strcpy(data->group, group->name);
     data->n_tag = utarray_len(tags);
-    read_group(group->driver->cache, tags, (neu_resp_tag_value_t *) &data[1]);
+    read_group(group->driver->cache, group->name, tags,
+               (neu_resp_tag_value_t *) &data[1]);
 
     group->driver->adapter.cb_funs.response(&group->driver->adapter, &header,
                                             data);
@@ -538,7 +540,7 @@ static void group_change(void *arg, int64_t timestamp, UT_array *tags,
 
     utarray_foreach(group->grp.tags, neu_datatag_t **, p_tag)
     {
-        neu_driver_cache_del(group->driver->cache, (*p_tag)->name);
+        neu_driver_cache_del(group->driver->cache, group->name, (*p_tag)->name);
     }
 
     neu_plugin_group_t grp = {
@@ -583,8 +585,8 @@ static int read_callback(void *usr_data)
     return 0;
 }
 
-static void read_group(neu_driver_cache_t *cache, UT_array *tags,
-                       neu_resp_tag_value_t *datas)
+static void read_group(neu_driver_cache_t *cache, const char *group,
+                       UT_array *tags, neu_resp_tag_value_t *datas)
 {
     utarray_foreach(tags, neu_datatag_t *, tag)
     {
@@ -593,7 +595,7 @@ static void read_group(neu_driver_cache_t *cache, UT_array *tags,
 
         strcpy(datas[index].tag, tag->name);
 
-        if (neu_driver_cache_get(cache, tag->name, &value) != 0) {
+        if (neu_driver_cache_get(cache, group, tag->name, &value) != 0) {
             datas[index].value.type      = NEU_TYPE_ERROR;
             datas[index].value.value.i32 = NEU_ERR_PLUGIN_READ_FAILURE;
             continue;