TD-1310

packaging/cfg/taos.cfg

@@ -122,11 +122,14 @@
 # number of replications, for cluster only 
 # replica               1
 
-# mqtt uri  
+# mqtt hostname  
-# mqttBrokerAddress     mqtt://username:password@hostname:1883/taos/
+# mqttHostName          test.mosquitto.org
 
-# mqtt client name 
+# mqtt port  
-# mqttBrokerClientId    taos_mqtt
+# mqttPort              1883
+
+# mqtt topic
+# mqttTopic             /weather/loop
 
 # the compressed rpc message, option:
 #  -1 (no compression)

src/common/inc/tglobal.h

@@ -104,8 +104,12 @@ extern int32_t  tsTelegrafUseFieldNum;
 
 // mqtt
 extern int32_t tsEnableMqttModule;
-extern char    tsMqttBrokerAddress[];
+extern char tsMqttHostName[];
-extern char    tsMqttBrokerClientId[];
+extern char tsMqttPort[];
+extern char tsMqttUser[];
+extern char tsMqttPass[];
+extern char tsMqttClientId[];
+extern char tsMqttTopic[];
 
 // monitor
 extern int32_t tsEnableMonitorModule;

src/common/src/tglobal.c

@@ -137,8 +137,12 @@ int32_t  tsTelegrafUseFieldNum = 0;
 
 // mqtt
 int32_t tsEnableMqttModule = 0;  // not finished yet, not started it by default
-char    tsMqttBrokerAddress[128] = {0};
+char    tsMqttHostName[TSDB_MQTT_HOSTNAME_LEN] = "test.mosquitto.org";
-char    tsMqttBrokerClientId[128] = {0};
+char    tsMqttPort[TSDB_MQTT_PORT_LEN] = "1883";
+char    tsMqttUser[TSDB_MQTT_USER_LEN] = {0};
+char    tsMqttPass[TSDB_MQTT_PASS_LEN] = {0};
+char    tsMqttClientId[TSDB_MQTT_CLIENT_ID_LEN] = "TDengineMqttSubscriber";
+char    tsMqttTopic[TSDB_MQTT_TOPIC_LEN] = "/weather/loop";
 
 // monitor
 int32_t tsEnableMonitorModule = 1;
@@ -767,23 +771,33 @@ static void doInitGlobalConfig(void) {
   cfg.unitType = TAOS_CFG_UTYPE_NONE;
   taosInitConfigOption(cfg);
 
-  cfg.option = "mqttBrokerAddress";
+  cfg.option = "mqttHostName";
-  cfg.ptr = tsMqttBrokerAddress;
+  cfg.ptr = tsMqttHostName;
   cfg.valType = TAOS_CFG_VTYPE_STRING;
   cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_CLIENT | TSDB_CFG_CTYPE_B_NOT_PRINT;
   cfg.minValue = 0;
   cfg.maxValue = 0;
-  cfg.ptrLength = 126;
+  cfg.ptrLength = TSDB_MQTT_HOSTNAME_LEN;
   cfg.unitType = TAOS_CFG_UTYPE_NONE;
   taosInitConfigOption(cfg);
 
-  cfg.option = "mqttBrokerClientId";
+  cfg.option = "mqttPort";
-  cfg.ptr = tsMqttBrokerClientId;
+  cfg.ptr = tsMqttPort;
   cfg.valType = TAOS_CFG_VTYPE_STRING;
   cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_CLIENT | TSDB_CFG_CTYPE_B_NOT_PRINT;
   cfg.minValue = 0;
   cfg.maxValue = 0;
-  cfg.ptrLength = 126;
+  cfg.ptrLength = TSDB_MQTT_PORT_LEN;
+  cfg.unitType = TAOS_CFG_UTYPE_NONE;
+  taosInitConfigOption(cfg);
+
+  cfg.option = "mqttTopic";
+  cfg.ptr = tsMqttTopic;
+  cfg.valType = TAOS_CFG_VTYPE_STRING;
+  cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_CLIENT | TSDB_CFG_CTYPE_B_NOT_PRINT;
+  cfg.minValue = 0;
+  cfg.maxValue = 0;
+  cfg.ptrLength = TSDB_MQTT_TOPIC_LEN;
   cfg.unitType = TAOS_CFG_UTYPE_NONE;
   taosInitConfigOption(cfg);
 

src/inc/taosdef.h

@@ -272,6 +272,13 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size);
 #define TSDB_SHOW_SQL_LEN         64
 #define TSDB_SLOW_QUERY_SQL_LEN   512
 
+#define TSDB_MQTT_HOSTNAME_LEN    64
+#define TSDB_MQTT_PORT_LEN        8
+#define TSDB_MQTT_USER_LEN        24
+#define TSDB_MQTT_PASS_LEN        24
+#define TSDB_MQTT_TOPIC_LEN       64
+#define TSDB_MQTT_CLIENT_ID_LEN   32
+
 #define TSDB_METER_STATE_OFFLINE  0
 #define TSDB_METER_STATE_ONLLINE  1
 

src/plugins/mqtt/inc/mqttInit.h

@@ -37,12 +37,6 @@ extern "C" {
  * \ref mqttReconnectClient is called, this instance will be passed.
  */
 struct reconnect_state_t {
-  char*    hostname;
-  char*    port;
-  char*    topic;
-  char*    client_id;
-  char*    user_name;
-  char*    password;
   uint8_t* sendbuf;
   size_t   sendbufsz;
   uint8_t* recvbuf;

src/plugins/mqtt/src/mqttSystem.c

@@ -30,123 +30,58 @@
 #include "tsocket.h"
 #include "ttimer.h"
 #include "mqttSystem.h"
-struct mqtt_client       mqttClient = {0};
-pthread_t                clientDaemonThread = {0};
-void*                    mqttConnect=NULL;
-struct reconnect_state_t recntStatus = {0};
-char*                    topicPath=NULL;
-int                      mttIsRuning = 1;
 
-int32_t mqttInitSystem() {
+#define MQTT_SEND_BUF_SIZE 102400
-  int   rc = 0;
+#define MQTT_RECV_BUF_SIZE 102400
-#if 0
-  uint8_t sendbuf[2048];
-  uint8_t recvbuf[1024];
-  recntStatus.sendbuf = sendbuf;
-  recntStatus.sendbufsz = sizeof(sendbuf);
-  recntStatus.recvbuf = recvbuf;
-  recntStatus.recvbufsz = sizeof(recvbuf);
-  char* url = tsMqttBrokerAddress;
-  recntStatus.user_name = strstr(url, "@") != NULL ? strbetween(url, "//", ":") : NULL;
 
-  char * passStr = strstr(url, recntStatus.user_name);
+struct mqtt_client       tsMqttClient = {0};
-  if (passStr != NULL) {
+struct reconnect_state_t tsMqttStatus = {0};
-    recntStatus.password = strstr(url, "@") != NULL ? strbetween(passStr, ":", "@") : NULL;
+static pthread_t         tsMqttClientDaemonThread = {0};
-  }
+static void*             tsMqttConnect = NULL;
+static bool              mqttIsRuning = false;
 
-  if (strlen(url) == 0) {
+void  mqttPublishCallback(void** unused, struct mqtt_response_publish* published);
-    mqttDebug("mqtt module not init, url is null");
+void  mqttCleanupRes(int status, int sockfd, pthread_t* client_daemon);
-    return rc;
+void* mqttClientRefresher(void* client);
-  }
 
-  if (strstr(url, "@") != NULL) {
+int32_t mqttInitSystem() { return 0; }
-    recntStatus.hostname = strbetween(url, "@", ":");
-  } else if (strstr(strstr(url, "://") + 3, ":") != NULL) {
-    recntStatus.hostname = strbetween(url, "//", ":");
-
-  } else {
-    recntStatus.hostname = strbetween(url, "//", "/");
-  }
-
-  char* _begin_hostname = strstr(url, recntStatus.hostname);
-  if (_begin_hostname != NULL && strstr(_begin_hostname, ":") != NULL) {
-    recntStatus.port = strbetween(_begin_hostname, ":", "/");
-  } else {
-    recntStatus.port = strbetween("'1883'", "'", "'");
-  }
-
-  char* portStr = recntStatus.hostname;
-  if (_begin_hostname != NULL) {
-    char* colonStr = strstr(_begin_hostname, ":");
-    if (colonStr != NULL) {
-      portStr = recntStatus.port;
-    }
-  }
-
-  char* topicStr = strstr(url, portStr);
-  if (topicStr != NULL) {
-    topicPath = strbetween(topicStr, "/", "/");
-    char* _topic = "+/+/+/";
-    int   _tpsize = strlen(topicPath) + strlen(_topic) + 1;
-    recntStatus.topic = calloc(1, _tpsize);
-    sprintf(recntStatus.topic, "/%s/%s", topicPath, _topic);
-    recntStatus.client_id = strlen(tsMqttBrokerClientId) < 3 ? tsMqttBrokerClientId : "taos_mqtt";
-    mqttConnect = NULL;
-  } else {
-    topicPath = NULL;
-  }
-  
-#endif  
-  return rc;
-}
 
 int32_t mqttStartSystem() {
-  int rc = 0;
+  tsMqttStatus.sendbufsz = MQTT_SEND_BUF_SIZE;
-#if 0  
+  tsMqttStatus.recvbufsz = MQTT_RECV_BUF_SIZE;
-  if (recntStatus.user_name != NULL && recntStatus.password != NULL) {
+  tsMqttStatus.sendbuf = malloc(MQTT_SEND_BUF_SIZE);
-    mqttInfo("connecting to  mqtt://%s:%s@%s:%s/%s/", recntStatus.user_name, recntStatus.password,
+  tsMqttStatus.recvbuf = malloc(MQTT_RECV_BUF_SIZE);
-              recntStatus.hostname, recntStatus.port, topicPath);
+  mqttIsRuning = true;
-  } else if (recntStatus.user_name != NULL && recntStatus.password == NULL) {
+
-    mqttInfo("connecting to  mqtt://%s@%s:%s/%s/", recntStatus.user_name, recntStatus.hostname, recntStatus.port,
+  mqtt_init_reconnect(&tsMqttClient, mqttReconnectClient, &tsMqttStatus, mqttPublishCallback);
-              topicPath);
+  if (pthread_create(&tsMqttClientDaemonThread, NULL, mqttClientRefresher, &tsMqttClient)) {
+    mqttError("mqtt client failed to start daemon.");
+    mqttCleanupRes(EXIT_FAILURE, -1, NULL);
+    return -1;
   }
 
-  mqtt_init_reconnect(&mqttClient, mqttReconnectClient, &recntStatus, mqtt_PublishCallback);
+  mqttInfo("mqtt client listening for %s messages", tsMqttTopic);
-  if (pthread_create(&clientDaemonThread, NULL, mqttClientRefresher, &mqttClient)) {
+  return 0;
-    mqttError("Failed to start client daemon.");
-    mqttCleanup(EXIT_FAILURE, -1, NULL);
-    rc = -1;
-  } else {
-    mqttInfo("listening for '%s' messages.", recntStatus.topic);
-  }
-#endif
-  return rc;
 }
 
 void mqttStopSystem() {
-#if 0  
+  if (mqttIsRuning) {
-  mqttClient.error = MQTT_ERROR_SOCKET_ERROR;
+    mqttIsRuning = false;
-  mttIsRuning = 0;
+    tsMqttClient.error = MQTT_ERROR_SOCKET_ERROR;
-  usleep(300000U);
+
-  mqttCleanup(EXIT_SUCCESS, mqttClient.socketfd, &clientDaemonThread);
+    taosMsleep(300);
-  mqttInfo("mqtt is stoped");
+    mqttCleanupRes(EXIT_SUCCESS, tsMqttClient.socketfd, &tsMqttClientDaemonThread);
-#endif  
+
+    mqttInfo("mqtt is stopped");
+  }
 }
 
 void mqttCleanUpSystem() {
-#if 0
+  mqttStopSystem();
-  mqttInfo("starting to cleanup mqtt");
-  free(recntStatus.user_name);
-  free(recntStatus.password);
-  free(recntStatus.hostname);
-  free(recntStatus.port);
-  free(recntStatus.topic);
-  free(topicPath);
   mqttInfo("mqtt is cleaned up");
-#endif  
 }
 
-void mqtt_PublishCallback(void** unused, struct mqtt_response_publish* published) {
+void mqttPublishCallback(void** unused, struct mqtt_response_publish* published) {
   /* note that published->topic_name is NOT null-terminated (here we'll change it to a c-string) */
   char* topic_name = (char*)malloc(published->topic_name_size + 1);
   memcpy(topic_name, published->topic_name, published->topic_name_size);
@@ -155,28 +90,29 @@ void mqtt_PublishCallback(void** unused, struct mqtt_response_publish* published
   char _token[128] = {0};
   char _dbname[128] = {0};
   char _tablename[128] = {0};
-  if (mqttConnect == NULL) {
+  if (tsMqttConnect == NULL) {
     mqttInfo("connect database");
-    taos_connect_a(NULL, "_root", tsInternalPass, "", 0, mqttInitConnCb, &mqttClient, &mqttConnect);
+    taos_connect_a(NULL, "_root", tsInternalPass, "", 0, mqttInitConnCb, &tsMqttClient, &tsMqttConnect);
   }
-  if (topic_name[1]=='/' &&  strncmp((char*)&topic_name[1], topicPath, strlen(topicPath)) == 0) {
+  if (topic_name[1] == '/' && strncmp((char*)&topic_name[1], tsMqttTopic, strlen(tsMqttTopic)) == 0) {
     char* p_p_cmd_part[5] = {0};
     char  copystr[1024] = {0};
     strncpy(copystr, topic_name, MIN(1024, published->topic_name_size));
     char part_index = split(copystr, "/", p_p_cmd_part, 10);
     if (part_index < 4) {
-      mqttError("The topic %s is't format '/path/token/dbname/table name/'. for expmle: '/taos/token/db/t'", topic_name);
+      mqttError("The topic %s is't format '/path/token/dbname/table name/'. for expmle: '/taos/token/db/t'",
+                topic_name);
     } else {
       strncpy(_token, p_p_cmd_part[1], 127);
       strncpy(_dbname, p_p_cmd_part[2], 127);
       strncpy(_tablename, p_p_cmd_part[3], 127);
       mqttInfo("part count=%d,access token:%s,database name:%s, table name:%s", part_index, _token, _dbname,
-                _tablename);
+               _tablename);
 
-      if (mqttConnect != NULL) {
+      if (tsMqttConnect != NULL) {
         char* _sql = converJsonToSql((char*)published->application_message, _dbname, _tablename);
         mqttInfo("query:%s", _sql);
-        taos_query_a(mqttConnect, _sql, mqttQueryInsertCallback, &mqttClient);
+        taos_query_a(tsMqttConnect, _sql, mqttQueryInsertCallback, &tsMqttClient);
         mqttInfo("free sql:%s", _sql);
         free(_sql);
       }
@@ -186,27 +122,31 @@ void mqtt_PublishCallback(void** unused, struct mqtt_response_publish* published
 }
 
 void* mqttClientRefresher(void* client) {
-  while (mttIsRuning) {
+  while (mqttIsRuning) {
     mqtt_sync((struct mqtt_client*)client);
     taosMsleep(100);
   }
-  mqttDebug("quit refresher");
+
+  mqttDebug("mqtt client quit refresher");
   return NULL;
 }
 
-void mqttCleanup(int status, int sockfd, pthread_t* client_daemon) {
+void mqttCleanupRes(int status, int sockfd, pthread_t* client_daemon) {
-#if 0 
   mqttInfo("clean up mqtt module");
-  if (sockfd != -1) close(sockfd);
+  if (sockfd != -1) {
-  if (client_daemon != NULL) pthread_cancel(*client_daemon);
+    close(sockfd);
-#endif
+  }
+
+  if (client_daemon != NULL) {
+    pthread_cancel(*client_daemon);
+  }
 }
 
 void mqttInitConnCb(void* param, TAOS_RES* result, int32_t code) {
   if (code < 0) {
     mqttError("mqtt:%d, connect to database failed, reason:%s", code, tstrerror(code));
-    taos_close(mqttConnect);
+    taos_close(tsMqttConnect);
-    mqttConnect = NULL;
+    tsMqttConnect = NULL;
     return;
   }
   mqttDebug("mqtt:%d, connect to database success, reason:%s", code, tstrerror(code));
@@ -222,36 +162,28 @@ void mqttQueryInsertCallback(void* param, TAOS_RES* result, int32_t code) {
   }
 }
 
-void mqttReconnectClient(struct mqtt_client* client, void** reconnect_state_vptr) {
+void mqttReconnectClient(struct mqtt_client* client, void** unused) {
-  mqttInfo("reconnect client");
+  mqttInfo("mqtt client tries to connect to the server");
-  struct reconnect_state_t* reconnect_state = *((struct reconnect_state_t**)reconnect_state_vptr);
 
-  /* Close the clients socket if this isn't the initial reconnect call */
   if (client->error != MQTT_ERROR_INITIAL_RECONNECT) {
     close(client->socketfd);
   }
 
-  /* Perform error handling here. */
   if (client->error != MQTT_ERROR_INITIAL_RECONNECT) {
-    mqttError("mqttReconnectClient: called while client was in error state \"%s\"", mqtt_error_str(client->error));
+    mqttError("mqtt client was in error state %s", mqtt_error_str(client->error));
   }
 
-  /* Open a new socket. */
+  int sockfd = open_nb_socket("test.mosquitto.org", "1883");
-  int sockfd = open_nb_socket(reconnect_state->hostname, reconnect_state->port);
+  if (sockfd < 0) {
-  if (sockfd == -1) {
+    mqttError("mqtt client failed to open socket %s:%s", tsMqttHostName, tsMqttPort);
-    mqttError("failed to open socket: ");
+    mqttCleanupRes(EXIT_FAILURE, sockfd, NULL);
-    mqttCleanup(EXIT_FAILURE, sockfd, NULL);
   }
 
-  /* Reinitialize the client. */
+  // mqtt_reinit(client, sockfd, tsMqttStatus.sendbuf, tsMqttStatus.sendbufsz, tsMqttStatus.recvbuf, tsMqttStatus.recvbufsz);
-  mqtt_reinit(client, sockfd, reconnect_state->sendbuf, reconnect_state->sendbufsz, reconnect_state->recvbuf,
+  // mqtt_connect(client, tsMqttClientId, NULL, NULL, 0, tsMqttUser, tsMqttPass, MQTT_CONNECT_CLEAN_SESSION, 400);
-              reconnect_state->recvbufsz);
+  // mqtt_subscribe(client, tsMqttTopic, 0);
 
-  /* Ensure we have a clean session */
+  mqtt_reinit(client, sockfd, tsMqttStatus.sendbuf, tsMqttStatus.sendbufsz, tsMqttStatus.recvbuf, tsMqttStatus.recvbufsz);
-  uint8_t connect_flags = MQTT_CONNECT_CLEAN_SESSION;
+  mqtt_connect(client, tsMqttClientId, NULL, NULL, 0, NULL, NULL, MQTT_CONNECT_CLEAN_SESSION, 400);
-  /* Send connection request to the broker. */
+  mqtt_subscribe(client, "datetime", 0);
-  mqtt_connect(client, reconnect_state->client_id, NULL, NULL, 0, reconnect_state->user_name, reconnect_state->password,connect_flags, 400);
-
-  /* Subscribe to the topic. */
-  mqtt_subscribe(client, reconnect_state->topic, 0);
 }

src/wal/CMakeLists.txt

@@ -7,6 +7,5 @@ AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR}/src SRC)
 IF (TD_LINUX)
   ADD_LIBRARY(twal ${SRC})
   TARGET_LINK_LIBRARIES(twal tutil common)
-
   ADD_SUBDIRECTORY(test)
 ENDIF ()  

tests/script/general/connection/mqtt.sim

@@ -0,0 +1,11 @@
+system sh/stop_dnodes.sh
+
+system sh/deploy.sh -n dnode1 -i 1
+system sh/cfg.sh -n dnode1 -c walLevel -v 2
+system sh/cfg.sh -n dnode1 -c numOfMnodes -v 1
+system sh/cfg.sh -n dnode1 -c mnodeEqualVnodeNum -v 4
+system sh/cfg.sh -n dnode1 -c maxTablesPerVnode -v 100000
+system sh/cfg.sh -n dnode1 -c http -v 1
+system sh/cfg.sh -n dnode1 -c http -v 1
+system sh/cfg.sh -n dnode1 -c mqttBrokerAddress  -v mqtt://test.mosquitto.org:1883/#
+system sh/cfg.sh -n dnode1 -c mqttBrokerClientId -v taosmqtt