Merge pull request #2008 from maikebing/feature/mqtt

Add mqtt plugin to subscribe to telemetry data

.gitignore

@@ -41,3 +41,27 @@ pysim/
 
 # Doxygen Generated files
 html/
+/.vs
+/CMakeFiles/3.10.2
+/CMakeCache.txt
+/Makefile
+/*.cmake
+/deps
+/src/cq/test/CMakeFiles/cqtest.dir/*.cmake
+*.cmake
+/src/cq/test/CMakeFiles/cqtest.dir/*.make
+*.make
+link.txt
+*.internal
+*.includecache
+*.marks
+Makefile
+CMakeError.log
+*.log
+/CMakeFiles/CMakeRuleHashes.txt
+/CMakeFiles/Makefile2
+/CMakeFiles/TargetDirectories.txt
+/CMakeFiles/cmake.check_cache
+/out/isenseconfig/WSL-Clang-Debug
+/out/isenseconfig/WSL-GCC-Debug
+/test/cfg

CMakeSettings.json

+{
+  "configurations": [
+    {
+      "name": "WSL-GCC-Debug",
+      "generator": "Unix Makefiles",
+      "configurationType": "Debug",
+      "buildRoot": "${projectDir}\\build\\",
+      "installRoot": "${projectDir}\\out\\install\\${name}",
+      "cmakeExecutable": "/usr/bin/cmake",
+      "cmakeCommandArgs": "",
+      "buildCommandArgs": "",
+      "ctestCommandArgs": "",
+      "inheritEnvironments": [ "linux_x64" ],
+      "wslPath": "${defaultWSLPath}",
+      "addressSanitizerRuntimeFlags": "detect_leaks=0",
+      "variables": [
+        {
+          "name": "CMAKE_INSTALL_PREFIX",
+          "value": "/mnt/d/TDengine/TDengine/build",
+          "type": "PATH"
+        }
+      ]
+    }
+  ]
+}
\ No newline at end of file

src/common/inc/tglobal.h

@@ -103,6 +103,7 @@ extern int32_t tsOfflineThreshold;
 extern int32_t tsMgmtEqualVnodeNum;
 
 extern int32_t tsEnableHttpModule;
+extern int32_t tsEnableMqttModule;
 extern int32_t tsEnableMonitorModule;
 
 extern int32_t tsRestRowLimit;
@@ -147,6 +148,7 @@ extern int32_t jniDebugFlag;
 extern int32_t tmrDebugFlag;
 extern int32_t sdbDebugFlag;
 extern int32_t httpDebugFlag;
+extern int32_t mqttDebugFlag;
 extern int32_t monitorDebugFlag;
 extern int32_t uDebugFlag;
 extern int32_t rpcDebugFlag;

src/common/src/tglobal.c

@@ -120,6 +120,7 @@ int32_t tsOfflineThreshold = 86400*100;   // seconds 10days
 int32_t tsMgmtEqualVnodeNum = 4;
 
 int32_t tsEnableHttpModule = 1;
+int32_t tsEnableMqttModule = 1;
 int32_t tsEnableMonitorModule = 0;
 
 int32_t tsRestRowLimit = 10240;
@@ -134,6 +135,7 @@ int32_t cDebugFlag = 135;
 int32_t jniDebugFlag = 131;
 int32_t odbcDebugFlag = 131;
 int32_t httpDebugFlag = 131;
+int32_t mqttDebugFlag = 131;
 int32_t monitorDebugFlag = 131;
 int32_t qDebugFlag = 131;
 int32_t rpcDebugFlag = 135;
@@ -212,6 +214,7 @@ void taosSetAllDebugFlag() {
     jniDebugFlag = debugFlag;
     odbcDebugFlag = debugFlag;
     httpDebugFlag = debugFlag;
+    mqttDebugFlag = debugFlag;
     monitorDebugFlag = debugFlag;
     rpcDebugFlag = debugFlag;
     uDebugFlag = debugFlag;
@@ -890,6 +893,17 @@ static void doInitGlobalConfig() {
   cfg.unitType = TAOS_CFG_UTYPE_NONE;
   taosInitConfigOption(cfg);
 
+  
+  cfg.option = "mqtt";
+  cfg.ptr = &tsEnableMqttModule;
+  cfg.valType = TAOS_CFG_VTYPE_INT32;
+  cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW;
+  cfg.minValue = 0;
+  cfg.maxValue = 1;
+  cfg.ptrLength = 1;
+  cfg.unitType = TAOS_CFG_UTYPE_NONE;
+  taosInitConfigOption(cfg);
+
   cfg.option = "monitor";
   cfg.ptr = &tsEnableMonitorModule;
   cfg.valType = TAOS_CFG_VTYPE_INT32;
@@ -1112,6 +1126,17 @@ static void doInitGlobalConfig() {
   cfg.unitType = TAOS_CFG_UTYPE_NONE;
   taosInitConfigOption(cfg);
 
+  cfg.option = "mqttDebugFlag";
+  cfg.ptr = &mqttDebugFlag;
+  cfg.valType = TAOS_CFG_VTYPE_INT32;
+  cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_LOG;
+  cfg.minValue = 0;
+  cfg.maxValue = 255;
+  cfg.ptrLength = 0;
+  cfg.unitType = TAOS_CFG_UTYPE_NONE;
+  taosInitConfigOption(cfg);
+
+
   cfg.option = "monitorDebugFlag";
   cfg.ptr = &monitorDebugFlag;
   cfg.valType = TAOS_CFG_VTYPE_INT32;

src/dnode/CMakeLists.txt

@@ -16,7 +16,7 @@ IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
   AUX_SOURCE_DIRECTORY(src SRC)
 
   ADD_EXECUTABLE(taosd ${SRC})
-  TARGET_LINK_LIBRARIES(taosd mnode taos_static monitor http tsdb twal vnode cJson lz4)
+  TARGET_LINK_LIBRARIES(taosd mnode taos_static monitor http mqtt tsdb twal vnode cJson lz4)
 
   IF (TD_ACCOUNT)
     TARGET_LINK_LIBRARIES(taosd account)

src/dnode/src/dnodeModule.c

@@ -20,6 +20,7 @@
 #include "trpc.h"
 #include "mnode.h"
 #include "http.h"
+#include "mqtt.h"
 #include "monitor.h"
 #include "dnodeInt.h"
 #include "dnodeModule.h"
@@ -62,6 +63,16 @@ static void dnodeAllocModules() {
     dnodeSetModuleStatus(TSDB_MOD_HTTP);
   }
 
+  tsModule[TSDB_MOD_MQTT].enable = (tsEnableMqttModule == 1);
+  tsModule[TSDB_MOD_MQTT].name = "mqtt";
+  tsModule[TSDB_MOD_MQTT].initFp = mqttInitSystem;
+  tsModule[TSDB_MOD_MQTT].cleanUpFp = mqttCleanUpSystem;
+  tsModule[TSDB_MOD_MQTT].startFp = mqttStartSystem;
+  tsModule[TSDB_MOD_MQTT].stopFp = mqttStopSystem;
+  if (tsEnableMqttModule) {
+    dnodeSetModuleStatus(TSDB_MOD_MQTT);
+  }
+
   tsModule[TSDB_MOD_MONITOR].enable    = (tsEnableMonitorModule == 1);
   tsModule[TSDB_MOD_MONITOR].name      = "monitor";
   tsModule[TSDB_MOD_MONITOR].initFp    = monitorInitSystem;

src/inc/mqtt.h

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef TDENGINE_MQTT_H
+#define TDENGINE_MQTT_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include <stdint.h>
+
+int32_t mqttGetReqCount();
+int32_t mqttInitSystem();
+int32_t mqttStartSystem();
+void mqttStopSystem();
+void mqttCleanUpSystem();
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

src/inc/taosdef.h

@@ -376,6 +376,7 @@ typedef enum {
   TSDB_MOD_MGMT,
   TSDB_MOD_HTTP,
   TSDB_MOD_MONITOR,
+  TSDB_MOD_MQTT,
   TSDB_MOD_MAX
 } EModuleType;
 

src/plugins/CMakeLists.txt

@@ -3,3 +3,4 @@ PROJECT(TDengine)
 
 ADD_SUBDIRECTORY(monitor)
 ADD_SUBDIRECTORY(http)
+ADD_SUBDIRECTORY(mqtt)

src/plugins/mqtt/CMakeLists.txt

+CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
+PROJECT(TDengine)
+
+IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
+  INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/inc)
+  INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/client/inc)
+  INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/util/inc)
+  INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/query/inc)
+  INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/common/inc)
+  INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/zlib-1.2.11/inc)
+  INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/cJson/inc)
+  INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/lz4/inc)
+  INCLUDE_DIRECTORIES(${TD_OS_DIR}/inc)
+  INCLUDE_DIRECTORIES(inc)
+  AUX_SOURCE_DIRECTORY(src SRC)
+  ADD_LIBRARY(mqtt ${SRC})
+  TARGET_LINK_LIBRARIES(mqtt taos_static z)
+
+  IF (TD_ADMIN)
+    TARGET_LINK_LIBRARIES(mqtt admin)
+  ENDIF ()
+ENDIF ()

src/plugins/mqtt/inc/mqttLog.h

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef TDENGINE_MQTT_LOG_H
+#define TDENGINE_MQTT_LOG_H
+
+#include "tlog.h"
+
+extern int32_t mqttDebugFlag;
+
+#define mqttError(...)                       \
+  if (mqttDebugFlag & DEBUG_ERROR) {         \
+    taosPrintLog("ERROR MQT ", 255, __VA_ARGS__); \
+  }
+#define mqttWarn(...)                                  \
+  if ( mqttDebugFlag & DEBUG_WARN) {                    \
+    taosPrintLog("WARN  MQT ",  mqttDebugFlag, __VA_ARGS__); \
+  }
+#define  mqttTrace(...)                           \
+  if ( mqttDebugFlag & DEBUG_TRACE) {             \
+    taosPrintLog("MQT ",  mqttDebugFlag, __VA_ARGS__); \
+  }
+#define  mqttDump(...)                                        \
+  if ( mqttDebugFlag & DEBUG_TRACE) {                         \
+    taosPrintLongString("MQT ",  mqttDebugFlag, __VA_ARGS__); \
+  }
+#define  mqttPrint(...) \
+  { taosPrintLog("MQT ", 255, __VA_ARGS__); }
+
+#endif

src/plugins/mqtt/inc/mqttSystem.h

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef TDENGINE_MQTT_SYSTEM_H
+#define TDENGINE_MQTT_SYSTEM_H
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include <stdint.h>
+
+int32_t mqttGetReqCount();
+int32_t mqttInitSystem();
+int32_t mqttStartSystem();
+void    mqttStopSystem();
+void    mqttCleanUpSystem();
+ 
+#ifdef __cplusplus
+}
+#endif
+
+#endif

src/plugins/mqtt/src/mqttSystem.c

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#define _DEFAULT_SOURCE
+#include "mqttSystem.h"
+#include "mqtt.h"
+#include "mqttLog.h"
+#include "os.h"
+#include "taos.h"
+#include "tglobal.h"
+#include "tsocket.h"
+#include "ttimer.h"
+
+int32_t mqttGetReqCount() { return 0; }
+int     mqttInitSystem() {
+  mqttPrint("mqttInitSystem");
+  return 0;
+}
+
+int mqttStartSystem() {
+  mqttPrint("mqttStartSystem");
+  return 0;
+}
+
+void mqttStopSystem() {
+	mqttPrint("mqttStopSystem");
+}
+
+void mqttCleanUpSystem() { 
+	mqttPrint("mqttCleanUpSystem");
+}