From 4bb23da006f050c6014156c9aab675f37dda906a Mon Sep 17 00:00:00 2001
From: guowenxue <guowenxue@gmail.com>
Date: Sat, 15 Nov 2025 01:48:22 +0800
Subject: [PATCH] update thingsboard and test okay
---
project/thingsboard/thingsboard.c | 244 +++++++++++++++++++-----------------------------
1 files changed, 99 insertions(+), 145 deletions(-)
diff --git a/project/mosquitto/main.c b/project/thingsboard/thingsboard.c
similarity index 62%
rename from project/mosquitto/main.c
rename to project/thingsboard/thingsboard.c
index 31d8fd7..bfba740 100644
--- a/project/mosquitto/main.c
+++ b/project/thingsboard/thingsboard.c
@@ -24,11 +24,12 @@
#include "logger.h"
#include "util_proc.h"
-#include "modules.h"
-#include "conf.h"
+#include "config.h"
+#include "sht20.h"
+#include "leds.h"
#define PROG_VERSION "v1.0.0"
-#define DAEMON_PIDFILE "/tmp/.mqtt.pid"
+#define DAEMON_PIDFILE "/tmp/.thingsboard.pid"
void *mqtt_sub_worker(void *args);
void *mqtt_pub_worker(void *args);
@@ -54,7 +55,7 @@
int daemon = 1;
pthread_t tid;
mqtt_ctx_t ctx;
- char *conf_file=NULL;
+ char *conf_file="/etc/mqttd.conf";
int debug = 0;
int opt;
char *progname=NULL;
@@ -69,7 +70,7 @@
progname = (char *)basename(argv[0]);
- /* Parser the command line parameters */
+ /* parser the command line parameters */
while ((opt = getopt_long(argc, argv, "c:dvh", long_options, NULL)) != -1)
{
switch (opt)
@@ -100,19 +101,24 @@
if( !conf_file )
debug = 1;
+ /* parser configure file */
if( mqttd_parser_conf(conf_file, &ctx, debug)<0 )
{
fprintf(stderr, "Parser mqtted configure file failure\n");
return -2;
}
+ /* install signal proc handler */
install_default_signal();
+ /* check program already running or not, if already running then exit, or set running as daemon */
if( check_set_program_running(daemon, DAEMON_PIDFILE) < 0 )
goto cleanup;
+ /* initial mosquitto library */
mosquitto_lib_init();
+ /* create MQTT subsciber thread */
if( thread_start(&tid, mqtt_sub_worker, &ctx ) < 0 )
{
log_error("Start MQTT subsciber worker thread failure\n");
@@ -120,6 +126,7 @@
}
log_info("Start MQTT subsciber worker thread ok\n");
+ /* create MQTT publisher thread */
if( thread_start(&tid, mqtt_pub_worker, &ctx) < 0 )
{
log_error("Start MQTT publisher worker thread failure\n");
@@ -157,38 +164,14 @@
log_info("Publisher connect to broker server[%s:%d] successfully\n", ctx->host, ctx->port);
log_debug("Publish topic '%s'\n", ctx->pubTopic);
- if( ctx->hwconf.ds18b20 )
- {
- memset(msg, 0, sizeof(msg));
-
- log_debug("DS18B20 temperature sensor enabled, start broadcast it\n");
-
- if( ds18b20_get_temperature(&temp) < 0 )
- snprintf(msg, sizeof(msg), "{ \"id\":%s, \"temp\":\"error\" }", ctx->devid);
- else
- snprintf(msg, sizeof(msg), "{ \"id\":%s, \"temp\":\"%.2f\" }", ctx->devid, temp);
-
- rv = mosquitto_publish(mosq, NULL, ctx->pubTopic, strlen(msg), msg, ctx->pubQos, retain);
- if( rv )
- {
- log_error("Publisher broadcast message '%s' failure: %d\n", msg, rv);
- }
- else
- {
- log_info("Publisher broadcast message '%s' ok\n", msg);
- }
- }
-
if( ctx->hwconf.sht2x )
{
memset(msg, 0, sizeof(msg));
log_debug("SHT2X temperature and humidity sensor enabled, start broadcast it\n");
- if( sht2x_get_temp_humidity(&temp, &rh) < 0 )
- snprintf(msg, sizeof(msg), "{ \"id\":%s, \"temp\":\"error\", \"RH\":\"error\" }", ctx->devid);
- else
- snprintf(msg, sizeof(msg), "{ \"id\":%s, \"temp\":\"%.2f\", \"RH\":\"%.2f\" }", ctx->devid, temp, rh);
+ if( 0 == sht2x_get_temp_humidity(&temp, &rh) )
+ snprintf(msg, sizeof(msg), "{\"temperature\":%.2f, \"humidity\":%.2f}", temp, rh);
rv = mosquitto_publish(mosq, NULL, ctx->pubTopic, strlen(msg), msg, ctx->pubQos, retain);
if( rv )
@@ -201,6 +184,7 @@
}
}
+ msleep(500);
log_info("Publisher broadcast over and disconnect broker now\n", ctx->host, ctx->port);
mosquitto_disconnect(mosq);
@@ -214,17 +198,24 @@
struct mosquitto *mosq;
bool session = true;
-
- mosq = mosquitto_new(NULL, session, ctx);
+ mosq = mosquitto_new(ctx->devid, session, ctx);
if( !mosq )
{
log_error("mosquitto_new failure\n");
return NULL;
}
- /* set connnect to broker username and password */
- if( strlen(ctx->uid)> 0 && strlen(ctx->pwd)> 0 )
+ /* connnect to broker by token or uid/pwd */
+ if( strlen(ctx->token ) > 0)
+ {
+ log_info("Using token authentication\n");
+ mosquitto_username_pw_set(mosq, ctx->token, NULL);
+ }
+ else if( strlen(ctx->uid)> 0 && strlen(ctx->pwd)> 0 )
+ {
+ log_info("Using username/password authentication\n");
mosquitto_username_pw_set(mosq, ctx->uid, ctx->pwd);
+ }
/* set callback functions */
mosquitto_connect_callback_set(mosq, pub_connect_callback);
@@ -271,101 +262,74 @@
log_warn("Subscriber disconnect to broker server[%s:%d], reason=%d\n", ctx->host, ctx->port, result);
}
-static inline void mqtt_turn_led(int which, char *cmd)
-{
- if( strcasestr(cmd, "on") )
- turn_led(which, ON);
- else if( strcasestr(cmd, "off") )
- turn_led(which, OFF);
-}
-
void proc_json_items(cJSON *root, mqtt_ctx_t *ctx)
{
- int i;
- char *value;
- cJSON *item;
- cJSON *array;
- hwconf_t *hwconf = &ctx->hwconf;
+ cJSON *item_method;
+ cJSON *item_params;
+ cJSON *item_device;
+ cJSON *item_status;
- if( !root )
+ char *method;
+ char *device;
+ int status;
+
+ /* 获取 method 字段 */
+ item_method = cJSON_GetObjectItem(root, "method");
+ if (item_method == NULL || !cJSON_IsString(item_method))
{
- log_error("Invalid input arguments $root\n");
- return ;
+ printf("JSON missing method\n");
+ return;
+ }
+ method = item_method->valuestring;
+
+ /* 只处理 setValue 命令 */
+ if (strcmp(method, "setValue") != 0)
+ {
+ printf("Ignore method: %s\n", method);
+ return;
}
- for( i=0; i<cJSON_GetArraySize(root); i++ )
+ /* 获取 params 对象 */
+ item_params = cJSON_GetObjectItem(root, "params");
+ if (item_params == NULL || !cJSON_IsObject(item_params))
{
- item = cJSON_GetArrayItem(root, i);
- if( !item )
- break;
-
- /* if item is cJSON_Object, then recursive call proc_json */
- if( cJSON_Object == item->type )
- {
- proc_json_items(item, ctx);
- }
- else if( cJSON_Array == item->type )
- {
- /* RGB colors led control: {"id":"RPi3B#01", "leds":[{"red":"on","green":"off","blue":"on"}]} */
- if( hwconf->led && !strcasecmp(item->string, "leds") )
- {
- array = cJSON_GetArrayItem(item, 0);
- if( NULL != array )
- {
- cJSON *led_item;
-
- if( NULL != (led_item=cJSON_GetObjectItem(array , "red")) )
- {
- log_info("turn red led '%s'\n", led_item->valuestring);
- mqtt_turn_led(LED_R, led_item->valuestring);
- }
-
- if( NULL != (led_item=cJSON_GetObjectItem(array , "green")) )
- {
- log_info("turn green led '%s'\n", led_item->valuestring);
- mqtt_turn_led(LED_G, led_item->valuestring);
- }
-
- if( NULL != (led_item=cJSON_GetObjectItem(array , "blue")) )
- {
- log_info("turn blue led '%s'\n", led_item->valuestring);
- mqtt_turn_led(LED_B, led_item->valuestring);
- }
- }
- }
- }
- else
- {
- value = cJSON_Print(item);
-
- /* light controled by relay: {"id":"RPi3B#01", "light":"on"} */
- if( hwconf->relay && !strcasecmp(item->string, "light") )
- {
- if( strcasestr(value, "on") )
- {
- log_info("Turn light on\n");
- turn_relay(RELAY1, ON);
- }
- else if( strcasestr(value, "off") )
- {
- log_info("Turn light off\n");
- turn_relay(RELAY1, OFF);
- }
- }
-
- /* buzzer controled : {"id":"RPi3B#01", "buzzer":"on"} */
- if( hwconf->beeper && !strcasecmp(item->string, "buzzer") )
- {
- if( strcasestr(value, "on") )
- {
- turn_beep(3);
- }
- }
-
- free(value); /* must free it, or it will result memory leak */
- }
+ printf("JSON missing params\n");
+ return;
}
+ /* 获取 params.device */
+ item_device = cJSON_GetObjectItem(item_params, "device");
+ if (item_device == NULL || !cJSON_IsString(item_device))
+ {
+ printf("JSON missing params.device\n");
+ return;
+ }
+ device = item_device->valuestring;
+
+ /* 获取 params.status (布尔值) */
+ item_status = cJSON_GetObjectItem(item_params, "status");
+ if (item_status == NULL || !(cJSON_IsBool(item_status)))
+ {
+ printf("JSON missing params.status\n");
+ return;
+ }
+ status = cJSON_IsTrue(item_status) ? ON : OFF;
+
+ /* 映射 LED 名称并调用控制函数 */
+ if (strcmp(device, "RedLed") == 0)
+ {
+ turn_led(LED_R, status);
+ }
+ else if (strcmp(device, "GreenLed") == 0)
+ {
+ turn_led(LED_G, status);
+ }
+ else if (strcmp(device, "BlueLed") == 0)
+ {
+ turn_led(LED_B, status);
+ }
+
+ return;
}
void sub_message_callback(struct mosquitto *mosq, void *userdata, const struct mosquitto_message *message)
@@ -373,8 +337,6 @@
mqtt_ctx_t *ctx = (mqtt_ctx_t *)userdata;
cJSON *root = NULL;
- cJSON *item;
- char *value;
if ( !message->payloadlen )
{
@@ -391,27 +353,11 @@
return ;
}
- item = cJSON_GetObjectItem(root, "id");
- if( !item )
- {
- log_error("cJSON_Parse get ID failure: %s\n", cJSON_GetErrorPtr());
- goto cleanup;
- }
-
- value = cJSON_PrintUnformatted(item);
- if( strcasecmp(value, ctx->devid) )
- {
- free(value);
- goto cleanup;
- }
-
- free(value);
- log_info("Subscriber receive message: '%s'\n", message->payload);
-
proc_json_items(root, ctx);
-cleanup:
- cJSON_Delete(root); /* must delete it, or it will result memory leak */
+ /* must delete it, or it will result memory leak */
+ cJSON_Delete(root);
+
return ;
}
@@ -422,16 +368,24 @@
struct mosquitto *mosq;
bool session = true;
- mosq = mosquitto_new(NULL, session, ctx);
+ mosq = mosquitto_new(ctx->devid, session, ctx);
if( !mosq )
{
log_error("mosquitto_new failure\n");
return NULL;
}
- /* set connnect to broker username and password */
- if( strlen(ctx->uid)> 0 && strlen(ctx->pwd)> 0 )
+ /* connnect to broker by token or uid/pwd */
+ if( strlen(ctx->token ) > 0)
+ {
+ log_info("Using token authentication\n");
+ mosquitto_username_pw_set(mosq, ctx->token, NULL);
+ }
+ else if( strlen(ctx->uid)> 0 && strlen(ctx->pwd)> 0 )
+ {
+ log_info("Using username/password authentication\n");
mosquitto_username_pw_set(mosq, ctx->uid, ctx->pwd);
+ }
/* set callback functions */
mosquitto_connect_callback_set(mosq, sub_connect_callback);
--
Gitblit v1.9.1