Merge pull request #2193 from fengzeroz/v2.9

version: 2.9.3

.github/workflows/ci.yml

@@ -8,7 +8,7 @@ on:
       - published
 
 env:
-  VERSION: 2.9.2
+  VERSION: 2.9.3
   UI_VERSION: 2.6.1
   BRANCH: v2.9 
 

include/neuron/connection/mqtt_client.h

@@ -85,25 +85,26 @@ bool   neu_mqtt_client_is_open(neu_mqtt_client_t *client);
 bool   neu_mqtt_client_is_connected(neu_mqtt_client_t *client);
 size_t neu_mqtt_client_get_cached_msgs_num(neu_mqtt_client_t *client);
 
-int neu_mqtt_client_set_addr(neu_mqtt_client_t *client, const char *host,
-                             uint16_t port);
-int neu_mqtt_client_set_id(neu_mqtt_client_t *client, const char *id);
-int neu_mqtt_client_set_will_msg(neu_mqtt_client_t *client, const char *topic,
-                                 uint8_t *msg, uint32_t len, bool retain,
-                                 uint8_t qos);
-int neu_mqtt_client_set_user(neu_mqtt_client_t *client, const char *username,
-                             const char *password);
-int neu_mqtt_client_set_connect_cb(neu_mqtt_client_t *             client,
-                                   neu_mqtt_client_connection_cb_t cb,
-                                   void *                          data);
-int neu_mqtt_client_set_disconnect_cb(neu_mqtt_client_t *             client,
-                                      neu_mqtt_client_connection_cb_t cb,
-                                      void *                          data);
-int neu_mqtt_client_set_tls(neu_mqtt_client_t *client, bool enabled,
-                            const char *ca, const char *cert, const char *key,
-                            const char *keypass);
-int neu_mqtt_client_set_cache_size(neu_mqtt_client_t *client,
-                                   size_t mem_size_bytes, size_t db_size_bytes);
+int  neu_mqtt_client_set_addr(neu_mqtt_client_t *client, const char *host,
+                              uint16_t port);
+int  neu_mqtt_client_set_id(neu_mqtt_client_t *client, const char *id);
+int  neu_mqtt_client_set_will_msg(neu_mqtt_client_t *client, const char *topic,
+                                  uint8_t *msg, uint32_t len, bool retain,
+                                  uint8_t qos);
+int  neu_mqtt_client_set_user(neu_mqtt_client_t *client, const char *username,
+                              const char *password);
+int  neu_mqtt_client_set_connect_cb(neu_mqtt_client_t *             client,
+                                    neu_mqtt_client_connection_cb_t cb,
+                                    void *                          data);
+int  neu_mqtt_client_set_disconnect_cb(neu_mqtt_client_t *             client,
+                                       neu_mqtt_client_connection_cb_t cb,
+                                       void *                          data);
+int  neu_mqtt_client_set_tls(neu_mqtt_client_t *client, bool enabled,
+                             const char *ca, const char *cert, const char *key,
+                             const char *keypass);
+void neu_mqtt_client_remove_cache_db(neu_mqtt_client_t *client);
+int  neu_mqtt_client_set_cache_size(neu_mqtt_client_t *client,
+                                    size_t mem_size_bytes, size_t db_size_bytes);
 // default to NEU_MQTT_CACHE_SYNC_INTERVAL_DEFAULT if not set
 int neu_mqtt_client_set_cache_sync_interval(neu_mqtt_client_t *client,
                                             uint32_t           interval);

include/neuron/define.h

@@ -24,7 +24,7 @@
 
 #define NEU_VERSION_MAJOR 2
 #define NEU_VERSION_MINOR 9
-#define NEU_VERSION_FIX 2
+#define NEU_VERSION_FIX 3
 
 #define NEU_GET_VERSION_MAJOR(V) ((0xffff0000 & V) >> 16)
 #define NEU_GET_VERSION_MINOR(V) ((0x0000ff00 & V) >> 8)

include/neuron/type.h

@@ -474,6 +474,238 @@ static inline void neu_value16_set_bit(neu_value16_u *value, uint8_t index,
     }
 }
 
+static inline uint16_t neu_get_u16(uint8_t *bytes)
+{
+    uint16_t ret = 0;
+    uint8_t *t   = NULL;
+
+    t = (uint8_t *) &ret;
+
+    t[0] = bytes[0];
+    t[1] = bytes[1];
+
+    return ret;
+}
+
+static inline void neu_set_u16(uint8_t *bytes, uint16_t val)
+{
+    uint8_t *t = NULL;
+
+    t = (uint8_t *) &val;
+
+    bytes[0] = t[0];
+    bytes[1] = t[1];
+}
+
+static inline int16_t neu_get_i16(uint8_t *bytes)
+{
+    int16_t  ret = 0;
+    uint8_t *t   = NULL;
+
+    t = (uint8_t *) &ret;
+
+    t[0] = bytes[0];
+    t[1] = bytes[1];
+
+    return ret;
+}
+
+static inline void neu_set_i16(uint8_t *bytes, int16_t val)
+{
+    uint8_t *t = NULL;
+
+    t = (uint8_t *) &val;
+
+    bytes[0] = t[0];
+    bytes[1] = t[1];
+}
+
+static inline uint32_t neu_get_u32(uint8_t *bytes)
+{
+    uint32_t ret = 0;
+    uint8_t *t   = NULL;
+
+    t = (uint8_t *) &ret;
+
+    t[0] = bytes[0];
+    t[1] = bytes[1];
+    t[2] = bytes[2];
+    t[3] = bytes[3];
+
+    return ret;
+}
+
+static inline void neu_set_u32(uint8_t *bytes, uint32_t val)
+{
+    uint8_t *t = NULL;
+
+    t = (uint8_t *) &val;
+
+    bytes[0] = t[0];
+    bytes[1] = t[1];
+    bytes[2] = t[2];
+    bytes[3] = t[3];
+}
+
+static inline int32_t neu_get_i32(uint8_t *bytes)
+{
+    int32_t  ret = 0;
+    uint8_t *t   = NULL;
+
+    t = (uint8_t *) &ret;
+
+    t[0] = bytes[0];
+    t[1] = bytes[1];
+    t[2] = bytes[2];
+    t[3] = bytes[3];
+
+    return ret;
+}
+
+static inline void neu_set_i32(uint8_t *bytes, int32_t val)
+{
+    uint8_t *t = NULL;
+
+    t = (uint8_t *) &val;
+
+    bytes[0] = t[0];
+    bytes[1] = t[1];
+    bytes[2] = t[2];
+    bytes[3] = t[3];
+}
+
+static inline float neu_get_f32(uint8_t *bytes)
+{
+    float    ret = 0;
+    uint8_t *t   = NULL;
+
+    t = (uint8_t *) &ret;
+
+    t[0] = bytes[0];
+    t[1] = bytes[1];
+    t[2] = bytes[2];
+    t[3] = bytes[3];
+
+    return ret;
+}
+
+static inline void neu_set_f32(uint8_t *bytes, float val)
+{
+    uint8_t *t = NULL;
+
+    t = (uint8_t *) &val;
+
+    bytes[0] = t[0];
+    bytes[1] = t[1];
+    bytes[2] = t[2];
+    bytes[3] = t[3];
+}
+
+static inline uint64_t neu_get_u64(uint8_t *bytes)
+{
+    uint64_t ret = 0;
+    uint8_t *t   = NULL;
+
+    t = (uint8_t *) &ret;
+
+    t[0] = bytes[0];
+    t[1] = bytes[1];
+    t[2] = bytes[2];
+    t[3] = bytes[3];
+    t[4] = bytes[4];
+    t[5] = bytes[5];
+    t[6] = bytes[6];
+    t[7] = bytes[7];
+
+    return ret;
+}
+
+static inline void neu_set_u64(uint8_t *bytes, uint64_t val)
+{
+    uint8_t *t = NULL;
+
+    t = (uint8_t *) &val;
+
+    bytes[0] = t[0];
+    bytes[1] = t[1];
+    bytes[2] = t[2];
+    bytes[3] = t[3];
+    bytes[4] = t[4];
+    bytes[5] = t[5];
+    bytes[6] = t[6];
+    bytes[7] = t[7];
+}
+
+static inline int64_t neu_get_i64(uint8_t *bytes)
+{
+    int64_t  ret = 0;
+    uint8_t *t   = NULL;
+
+    t = (uint8_t *) &ret;
+
+    t[0] = bytes[0];
+    t[1] = bytes[1];
+    t[2] = bytes[2];
+    t[3] = bytes[3];
+    t[4] = bytes[4];
+    t[5] = bytes[5];
+    t[6] = bytes[6];
+    t[7] = bytes[7];
+
+    return ret;
+}
+
+static inline void neu_set_i64(uint8_t *bytes, int64_t val)
+{
+    uint8_t *t = NULL;
+
+    t = (uint8_t *) &val;
+
+    bytes[0] = t[0];
+    bytes[1] = t[1];
+    bytes[2] = t[2];
+    bytes[3] = t[3];
+    bytes[4] = t[4];
+    bytes[5] = t[5];
+    bytes[6] = t[6];
+    bytes[7] = t[7];
+}
+
+static inline double neu_get_f64(uint8_t *bytes)
+{
+    double   ret = 0;
+    uint8_t *t   = NULL;
+
+    t = (uint8_t *) &ret;
+
+    t[0] = bytes[0];
+    t[1] = bytes[1];
+    t[2] = bytes[2];
+    t[3] = bytes[3];
+    t[4] = bytes[4];
+    t[5] = bytes[5];
+    t[6] = bytes[6];
+    t[7] = bytes[7];
+
+    return ret;
+}
+
+static inline void neu_set_f64(uint8_t *bytes, double val)
+{
+    uint8_t *t = NULL;
+
+    t = (uint8_t *) &val;
+
+    bytes[0] = t[0];
+    bytes[1] = t[1];
+    bytes[2] = t[2];
+    bytes[3] = t[3];
+    bytes[4] = t[4];
+    bytes[5] = t[5];
+    bytes[6] = t[6];
+    bytes[7] = t[7];
+}
+
 static inline void neu_ntohs24_p(uint8_t *value)
 {
     uint8_t t = 0;
@@ -513,7 +745,8 @@ static inline uint64_t neu_ntohll(uint64_t u)
 
 static inline void neu_ntonll_p(uint64_t *pu)
 {
-    *pu = neu_htonll(*pu);
+    uint64_t ret = neu_htonll(neu_get_u64((uint8_t *) pu));
+    neu_set_u64((uint8_t *) pu, ret);
 }
 
 static inline void neu_htonll_p(uint64_t *pu)
@@ -523,7 +756,8 @@ static inline void neu_htonll_p(uint64_t *pu)
 
 static inline void neu_ntohl_p(uint32_t *pu)
 {
-    *pu = ntohl(*pu);
+    uint32_t ret = ntohl(neu_get_u32((uint8_t *) pu));
+    neu_set_u32((uint8_t *) pu, ret);
 }
 
 static inline void neu_htonl_p(uint32_t *pu)
@@ -533,7 +767,8 @@ static inline void neu_htonl_p(uint32_t *pu)
 
 static inline void neu_ntohs_p(uint16_t *pu)
 {
-    *pu = ntohs(*pu);
+    uint16_t ret = ntohs(neu_get_u16((uint8_t *) pu));
+    neu_set_u16((uint8_t *) pu, ret);
 }
 
 static inline void neu_htons_p(uint16_t *pu)
@@ -541,134 +776,6 @@ static inline void neu_htons_p(uint16_t *pu)
     neu_ntohs_p(pu);
 }
 
-static inline uint16_t neu_get_u16(uint8_t *bytes)
-{
-    uint16_t ret = 0;
-    uint8_t *t   = NULL;
-
-    t = (uint8_t *) &ret;
-
-    t[0] = bytes[0];
-    t[1] = bytes[1];
-
-    return ret;
-}
-
-static inline int16_t neu_get_i16(uint8_t *bytes)
-{
-    int16_t  ret = 0;
-    uint8_t *t   = NULL;
-
-    t = (uint8_t *) &ret;
-
-    t[0] = bytes[0];
-    t[1] = bytes[1];
-
-    return ret;
-}
-
-static inline uint32_t neu_get_u32(uint8_t *bytes)
-{
-    uint32_t ret = 0;
-    uint8_t *t   = NULL;
-
-    t = (uint8_t *) &ret;
-
-    t[0] = bytes[0];
-    t[1] = bytes[1];
-    t[2] = bytes[2];
-    t[3] = bytes[3];
-
-    return ret;
-}
-
-static inline int32_t neu_get_i32(uint8_t *bytes)
-{
-    int32_t  ret = 0;
-    uint8_t *t   = NULL;
-
-    t = (uint8_t *) &ret;
-
-    t[0] = bytes[0];
-    t[1] = bytes[1];
-    t[2] = bytes[2];
-    t[3] = bytes[3];
-
-    return ret;
-}
-
-static inline float neu_get_f32(uint8_t *bytes)
-{
-    float    ret = 0;
-    uint8_t *t   = NULL;
-
-    t = (uint8_t *) &ret;
-
-    t[0] = bytes[0];
-    t[1] = bytes[1];
-    t[2] = bytes[2];
-    t[3] = bytes[3];
-
-    return ret;
-}
-
-static inline uint64_t neu_get_u64(uint8_t *bytes)
-{
-    uint64_t ret = 0;
-    uint8_t *t   = NULL;
-
-    t = (uint8_t *) &ret;
-
-    t[0] = bytes[0];
-    t[1] = bytes[1];
-    t[2] = bytes[2];
-    t[3] = bytes[3];
-    t[4] = bytes[4];
-    t[5] = bytes[5];
-    t[6] = bytes[6];
-    t[7] = bytes[7];
-
-    return ret;
-}
-
-static inline int64_t neu_get_i64(uint8_t *bytes)
-{
-    int64_t  ret = 0;
-    uint8_t *t   = NULL;
-
-    t = (uint8_t *) &ret;
-
-    t[0] = bytes[0];
-    t[1] = bytes[1];
-    t[2] = bytes[2];
-    t[3] = bytes[3];
-    t[4] = bytes[4];
-    t[5] = bytes[5];
-    t[6] = bytes[6];
-    t[7] = bytes[7];
-
-    return ret;
-}
-
-static inline double neu_get_f64(uint8_t *bytes)
-{
-    double   ret = 0;
-    uint8_t *t   = NULL;
-
-    t = (uint8_t *) &ret;
-
-    t[0] = bytes[0];
-    t[1] = bytes[1];
-    t[2] = bytes[2];
-    t[3] = bytes[3];
-    t[4] = bytes[4];
-    t[5] = bytes[5];
-    t[6] = bytes[6];
-    t[7] = bytes[7];
-
-    return ret;
-}
-
 #ifdef __cplusplus
 }
 #endif

plugins/mqtt/mqtt_plugin_intf.c

@@ -402,9 +402,17 @@ int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
     case NEU_REQ_UPDATE_NODE:
         error = handle_update_driver(plugin, data);
         break;
-    case NEU_REQRESP_NODE_DELETED:
-        error = handle_del_driver(plugin, data);
+    case NEU_REQRESP_NODE_DELETED: {
+        neu_reqresp_node_deleted_t *req = data;
+        if (0 != strcmp(plugin->common.name, req->node)) {
+            error = handle_del_driver(plugin, data);
+        } else {
+            if (plugin->client) {
+                neu_mqtt_client_remove_cache_db(plugin->client);
+            }
+        }
         break;
+    }
     default:
         error = NEU_ERR_MQTT_FAILURE;
         break;

src/connection/mqtt_client.c

@@ -123,6 +123,7 @@ struct neu_mqtt_client_s {
     neu_mqtt_client_connection_cb_t disconnect_cb;
     void *                          disconnect_cb_data;
     nng_mqtt_sqlite_option *        sqlite_cfg;
+    char *                          db;
     bool                            receiving;
     nng_aio *                       recv_aio;
     subscription_t *                subscriptions;
@@ -541,8 +542,10 @@ static void connect_cb(nng_pipe p, nng_pipe_ev ev, void *arg)
     neu_mqtt_client_t *             client = arg;
     neu_mqtt_client_connection_cb_t cb     = NULL;
     void *                          data   = NULL;
+    int                             reason = 0;
 
-    log(notice, "mqtt client connected");
+    nng_pipe_get_int(p, NNG_OPT_MQTT_CONNECT_REASON, &reason);
+    log(notice, "mqtt client connected, reason: %d", reason);
 
     nng_mtx_lock(client->mtx);
     // start receiving
@@ -567,8 +570,11 @@ static void disconnect_cb(nng_pipe p, nng_pipe_ev ev, void *arg)
     subscription_t *                sub    = NULL;
     neu_mqtt_client_connection_cb_t cb     = NULL;
     void *                          data   = NULL;
+    int                             reason = 0;
+
+    nng_pipe_get_int(p, NNG_OPT_MQTT_DISCONNECT_REASON, &reason);
+    log(notice, "mqtt client disconnected, reason: %d", reason);
 
-    log(notice, "mqtt client disconnected");
     nng_mtx_lock(client->mtx);
     client->connected = false;
     cb                = client->disconnect_cb;
@@ -901,26 +907,10 @@ static inline nng_mqtt_sqlite_option *
 alloc_sqlite_config(neu_mqtt_client_t *client)
 {
     int                     rv;
-    char *                  db  = NULL;
     nng_mqtt_sqlite_option *cfg = NULL;
-    const mqtt_buf          client_id =
-        nng_mqtt_msg_get_connect_client_id(client->conn_msg);
-
-    if (NULL == client_id.buf || 0 == client_id.length) {
-        log(error, "nng_mqtt_msg_get_connect_client_id fail");
-        return NULL;
-    }
-
-    neu_asprintf(&db, "neuron-mqtt-client-%.*s.db", (unsigned) client_id.length,
-                 (char *) client_id.buf);
-    if (NULL == db) {
-        log(error, "neu_asprintf nano-mqtt cilent db fail");
-        return NULL;
-    }
 
     if ((rv = nng_mqtt_alloc_sqlite_opt(&cfg)) != 0) {
         log(error, "nng_mqtt_alloc_sqlite_opt fail: %s", nng_strerror(rv));
-        free(db);
         return NULL;
     }
 
@@ -928,9 +918,7 @@ alloc_sqlite_config(neu_mqtt_client_t *client)
     nng_mqtt_set_sqlite_flush_threshold(cfg, 256);
     nng_mqtt_set_sqlite_max_rows(cfg, 256);
     nng_mqtt_set_sqlite_db_dir(cfg, "persistence/");
-    nng_mqtt_sqlite_db_init(cfg, db, client->version);
 
-    free(db);
     return cfg;
 }
 
@@ -989,6 +977,7 @@ void neu_mqtt_client_free(neu_mqtt_client_t *client)
         subscriptions_free(client->subscriptions);
         tasks_free(client->task_free_list);
         nng_msg_free(client->conn_msg);
+        free(client->db);
         free(client->url);
         free(client->host);
         nng_mtx_free(client->mtx);
@@ -1213,6 +1202,76 @@ end:
     return rv;
 }
 
+static void rand_str(char *dst, size_t len)
+{
+    char alphabet[] = "0123456789"
+                      "abcdefghijklmnopqrstuvwxyz"
+                      "ABCDEFGHIJKLMNOPQRSTUVWXYZ";
+
+    while (len-- > 1) {
+        size_t i = (double) rand() / RAND_MAX * ((sizeof alphabet) - 1);
+        *dst++   = alphabet[i];
+    }
+    *dst = '\0';
+}
+
+static int neu_mqtt_client_set_cache_db(neu_mqtt_client_t *client)
+{
+    char str[8] = { 0 };
+    rand_str(str, sizeof(str));
+
+    char *db = NULL;
+    neu_asprintf(&db, "neuron-mqtt-client-%s.db", str);
+    if (NULL == db) {
+        log(error, "neu_asprintf nano-mqtt cilent db fail");
+        return NEU_ERR_EINTERNAL;
+    }
+
+    if (client->db && 0 == strcmp(client->db, db)) {
+        free(db);
+        return 0;
+    }
+
+    neu_mqtt_client_remove_cache_db(client);
+
+    free(client->db);
+    client->db = db;
+
+    if (client->sqlite_cfg) {
+        nng_mqtt_sqlite_db_init(client->sqlite_cfg, db, client->version);
+    }
+
+    log(notice, "set cache db %s", db);
+
+    return 0;
+}
+
+void neu_mqtt_client_remove_cache_db(neu_mqtt_client_t *client)
+{
+    if (NULL == client->db) {
+        return;
+    }
+
+    const char *dir       = "persistence";
+    int         path_size = strlen(dir) + strlen(client->db) + 6;
+    char *      path      = calloc(path_size, 1);
+    if (NULL == path) {
+        return;
+    }
+
+    snprintf(path, path_size, "%s/%s", dir, client->db);
+    log(notice, "rm cache db %s", path);
+    remove(path);
+    snprintf(path, path_size, "%s/%s-shm", dir, client->db);
+    log(notice, "rm cache db %s", path);
+    remove(path);
+    snprintf(path, path_size, "%s/%s-wal", dir, client->db);
+    log(notice, "rm cache db %s", path);
+    remove(path);
+
+    free(path);
+}
+
 int neu_mqtt_client_set_cache_size(neu_mqtt_client_t *client,
                                    size_t mem_size_bytes, size_t db_size_bytes)
 {
@@ -1227,6 +1286,7 @@ int neu_mqtt_client_set_cache_size(neu_mqtt_client_t *client,
         if (client->sqlite_cfg) {
             nng_mqtt_free_sqlite_opt(client->sqlite_cfg);
             client->sqlite_cfg = NULL;
+            neu_mqtt_client_remove_cache_db(client);
         }
         goto end;
     }
@@ -1240,6 +1300,10 @@ int neu_mqtt_client_set_cache_size(neu_mqtt_client_t *client,
         }
     }
 
+    if (0 != (rv = neu_mqtt_client_set_cache_db(client))) {
+        return rv;
+    }
+
     // FIXME: until NanoSDK cache size limit feature out
     // Assume that,
     // 1. every upload message follows the below template

src/core/manager.c

@@ -801,10 +801,22 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
             }
             utarray_free(subscriptions);
         }
+
         neu_reqresp_node_deleted_t resp = { 0 };
         strcpy(resp.node, header->receiver);
+        // notify MQTT about node removal
+        if (0 == strcmp(adapter->module->module_name, "MQTT")) {
+            msg = neu_msg_new(NEU_REQRESP_NODE_DELETED, NULL, &resp);
+            if (NULL != msg) {
+                neu_reqresp_head_t *hd = neu_msg_get_header(msg);
+                strcpy(hd->receiver, cmd->node);
+                strcpy(hd->sender, "manager");
+                reply(manager, hd, &resp);
+            }
+        }
         header->type = NEU_REQ_NODE_UNINIT;
         forward_msg(manager, header, header->receiver);
+
         if (neu_adapter_get_type(adapter) == NEU_NA_TYPE_DRIVER) {
             UT_array *apps = neu_subscribe_manager_find_by_driver(
                 manager->subscribe_manager, resp.node);

src/core/manager_internal.c

@@ -243,9 +243,11 @@ int neu_manager_subscribe(neu_manager_t *manager, const char *app,
     // guard against empty mqtt topic parameter
     // this is not an elegant solution due to the current architecture
     if (params && 0 == strcmp(adapter->module->module_name, "MQTT")) {
-        neu_json_elem_t elem = { .name = "topic", .t = NEU_JSON_STR };
-        neu_parse_param(params, NULL, 1, &elem);
-        if (elem.v.val_str && 0 == strlen(elem.v.val_str)) {
+        neu_json_elem_t elem = { .name      = "topic",
+                                 .t         = NEU_JSON_STR,
+                                 .v.val_str = NULL };
+        int             ret  = neu_parse_param(params, NULL, 1, &elem);
+        if (ret != 0 || (elem.v.val_str && 0 == strlen(elem.v.val_str))) {
             free(elem.v.val_str);
             return NEU_ERR_MQTT_SUBSCRIBE_FAILURE;
         }

tests/ft/neuron/api.py

@@ -247,12 +247,12 @@ def get_subscribe_group(app, driver=None, group=None):
 
 
 @gen_check
-def subscribe_group(app, driver, group, params={}):
+def subscribe_group(app, driver, group, params={"topic": "/neuron/mqtt/test"}):
     return requests.post(url=config.BASE_URL + "/api/v2/subscribe", headers={"Authorization": config.default_jwt}, json={"app": app, "driver": driver, "group": group, "params": params})
 
 
 @gen_check
-def subscribe_group_update(app, driver, group, params={}):
+def subscribe_group_update(app, driver, group, params={"topic": "/neuron/mqtt/test"}):
     return requests.put(url=config.BASE_URL + "/api/v2/subscribe", headers={"Authorization": config.default_jwt}, json={"app": app, "driver": driver, "group": group, "params": params})
 
 

version

@@ -1 +1 @@
-2.9.2
+2.9.3