mirror of
https://github.com/oopuuu/zTC1.git
synced 2026-03-23 20:49:55 +08:00
Compare commits
3 Commits
dev_udp_re
...
dev_file_o
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
eaf8f5bf60 | ||
|
|
5eecfefc74 | ||
|
|
0c675c6f03 |
@@ -34,8 +34,7 @@ $(NAME)_SOURCES := main.c\
|
||||
user_power.c\
|
||||
timed_task/timed_task.c\
|
||||
http_server/web_log.c\
|
||||
http_server/app_httpd.c\
|
||||
udp_server/udp_server.c
|
||||
http_server/app_httpd.c
|
||||
|
||||
$(NAME)_COMPONENTS := protocols/SNTP\
|
||||
protocols/mqtt\
|
||||
|
||||
@@ -11,7 +11,6 @@
|
||||
#include "time_server/user_rtc.h"
|
||||
#include "user_power.h"
|
||||
#include "http_server/app_httpd.h"
|
||||
#include "udp_server/udp_server.h"
|
||||
#include "timed_task/timed_task.h"
|
||||
|
||||
char rtc_init = 0; //sntp校时成功标志位
|
||||
@@ -190,10 +189,7 @@ int application_start(void) {
|
||||
require_noerr(err, exit);
|
||||
PowerInit();
|
||||
AppHttpdStart(); // start http server thread
|
||||
// udp_server_start();
|
||||
//if (!(MQTT_SERVER[0] < 0x20 || MQTT_SERVER[0] > 0x7f || MQTT_SERVER_PORT < 1)){
|
||||
// UserMqttInit();
|
||||
// }
|
||||
|
||||
UserLedSet(user_config->power_led_enabled);
|
||||
|
||||
err = mico_rtos_create_thread(NULL, MICO_APPLICATION_PRIORITY, "p_count",
|
||||
|
||||
@@ -36,18 +36,10 @@ typedef struct {
|
||||
uint32_t datalen;
|
||||
} mqtt_recv_msg_t, *p_mqtt_recv_msg_t, mqtt_send_msg_t, *p_mqtt_send_msg_t;
|
||||
|
||||
typedef struct {
|
||||
Client client;
|
||||
Network network;
|
||||
// 其他状态、配置参数……
|
||||
} mqtt_context_t;
|
||||
|
||||
static void MqttClientThread(mico_thread_arg_t arg);
|
||||
|
||||
static void MessageArrived(MessageData *md);
|
||||
|
||||
static void MqttClientThread2(mico_thread_arg_t arg);
|
||||
|
||||
static OSStatus
|
||||
MqttMsgPublish(Client *c, const char *topic, char qos, char retained, const unsigned char *msg,
|
||||
uint32_t msg_len);
|
||||
@@ -57,14 +49,13 @@ OSStatus UserRecvHandler(void *arg);
|
||||
void ProcessHaCmd(char *cmd);
|
||||
|
||||
bool isconnect = false;
|
||||
bool isconnect2 = false;
|
||||
mico_queue_t mqtt_msg_send_queue = NULL;
|
||||
mico_queue_t mqtt_msg_send_queue2 = NULL;
|
||||
|
||||
Client c; // mqtt client object
|
||||
Network n; // socket network for mqtt client
|
||||
volatile bool mqtt_thread_should_exit = false;
|
||||
|
||||
static mico_worker_thread_t mqtt_client_worker_thread; /* Worker thread to manage send/recv events */
|
||||
|
||||
//static mico_timed_event_t mqtt_client_send_event;
|
||||
|
||||
char topic_state[MAX_MQTT_TOPIC_SIZE];
|
||||
@@ -72,8 +63,6 @@ char topic_set[MAX_MQTT_TOPIC_SIZE];
|
||||
|
||||
mico_timer_t timer_handle;
|
||||
static char timer_status = 0;
|
||||
mico_timer_t timer_handle2;
|
||||
static char timer_status_2 = 0;
|
||||
|
||||
void UserMqttTimerFunc(void *arg) {
|
||||
LinkStatusTypeDef LinkStatus;
|
||||
@@ -110,41 +99,6 @@ void UserMqttTimerFunc(void *arg) {
|
||||
}
|
||||
}
|
||||
|
||||
void UserMqttTimerFunc2(void *arg) {
|
||||
LinkStatusTypeDef LinkStatus;
|
||||
micoWlanGetLinkStatus(&LinkStatus);
|
||||
if (LinkStatus.is_connected != 1) {
|
||||
mico_stop_timer(&timer_handle2);
|
||||
return;
|
||||
}
|
||||
if (mico_rtos_is_queue_empty(&mqtt_msg_send_queue2)) {
|
||||
|
||||
switch (timer_status_2) {
|
||||
case 0:
|
||||
UserMqttHassAutoLed();
|
||||
UserMqttHassAutoTotalSocket();
|
||||
UserMqttHassAutoChildLock();
|
||||
UserMqttHassAutoRebootButton();
|
||||
break;
|
||||
case 1:
|
||||
case 2:
|
||||
case 3:
|
||||
case 4:
|
||||
case 5:
|
||||
case 6:
|
||||
UserMqttHassAuto(timer_status_2);
|
||||
break;
|
||||
case 7:
|
||||
UserMqttHassAutoPower();
|
||||
break;
|
||||
default:
|
||||
mico_stop_timer(&timer_handle2);
|
||||
break;
|
||||
}
|
||||
timer_status_2++;
|
||||
}
|
||||
}
|
||||
|
||||
OSStatus UserMqttDeInit(void) {
|
||||
OSStatus err = kNoErr;
|
||||
|
||||
@@ -154,10 +108,23 @@ OSStatus UserMqttDeInit(void) {
|
||||
return err;
|
||||
}
|
||||
|
||||
void clear_mqtt_msg_send_queue(void) {
|
||||
if(mqtt_msg_send_queue == NULL){
|
||||
return;
|
||||
}
|
||||
void *msg = NULL;
|
||||
while (mico_rtos_is_queue_empty(&mqtt_msg_send_queue) == false) {
|
||||
if (mico_rtos_pop_from_queue(&mqtt_msg_send_queue, &msg, 0) == kNoErr) {
|
||||
if (msg) free(msg); // 释放消息内存,避免泄漏
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/* Application entrance */
|
||||
OSStatus UserMqttInit(void) {
|
||||
OSStatus err = kNoErr;
|
||||
|
||||
if(mqtt_msg_send_queue != NULL)
|
||||
return err;
|
||||
sprintf(topic_set, MQTT_CLIENT_SUB_TOPIC1);
|
||||
sprintf(topic_state, MQTT_CLIENT_PUB_TOPIC, str_mac);
|
||||
//TODO size:0x800
|
||||
@@ -174,28 +141,14 @@ OSStatus UserMqttInit(void) {
|
||||
require_noerr_action(err, exit, mqtt_log("ERROR: create mqtt msg send queue err=%d.", err));
|
||||
|
||||
/* start mqtt client */
|
||||
mqtt_context_t *ctx1 = malloc(sizeof(mqtt_context_t));
|
||||
memset(ctx1, 0, sizeof(mqtt_context_t));
|
||||
|
||||
err = mico_rtos_create_thread(NULL, MICO_APPLICATION_PRIORITY, "mqtt_client_1",
|
||||
(mico_thread_function_t)MqttClientThread,
|
||||
mqtt_thread_stack_size, ctx1);
|
||||
err = mico_rtos_create_thread(NULL, MICO_APPLICATION_PRIORITY, "mqtt_client",
|
||||
(mico_thread_function_t) MqttClientThread,
|
||||
mqtt_thread_stack_size, 0);
|
||||
require_noerr_string(err, exit, "ERROR: Unable to start the mqtt client thread.");
|
||||
|
||||
// 第二个客户端
|
||||
mqtt_context_t *ctx2 = malloc(sizeof(mqtt_context_t));
|
||||
memset(ctx2, 0, sizeof(mqtt_context_t));
|
||||
|
||||
err = mico_rtos_create_thread(NULL, MICO_APPLICATION_PRIORITY, "mqtt_client_2",
|
||||
(mico_thread_function_t)MqttClientThread2,
|
||||
mqtt_thread_stack_size, ctx2);
|
||||
|
||||
|
||||
require_noerr_string(err, exit, "ERROR: Unable to start the mqtt client thread2.");
|
||||
|
||||
/* Create a worker thread for user handling MQTT data event */
|
||||
err = mico_rtos_create_worker_thread(&mqtt_client_worker_thread, MICO_APPLICATION_PRIORITY,
|
||||
0x1000, 5);
|
||||
0x800, 5);
|
||||
require_noerr_string(err, exit, "ERROR: Unable to start the mqtt client worker thread.");
|
||||
|
||||
exit:
|
||||
@@ -206,13 +159,32 @@ OSStatus UserMqttInit(void) {
|
||||
static OSStatus UserMqttClientRelease(Client *c, Network *n) {
|
||||
OSStatus err = kNoErr;
|
||||
|
||||
if (c->isconnected) MQTTDisconnect(c);
|
||||
if (c == NULL || n == NULL) return kParamErr;
|
||||
|
||||
n->disconnect(n); // close connection
|
||||
if (c->isconnected) {
|
||||
MQTTDisconnect(c);
|
||||
c->isconnected = 0;
|
||||
}
|
||||
|
||||
if (MQTT_SUCCESS != MQTTClientDeinit(c)) { mqtt_log("MQTTClientDeinit failed!");
|
||||
if (c->buf) {
|
||||
free(c->buf);
|
||||
c->buf = NULL;
|
||||
}
|
||||
|
||||
if (c->readbuf) {
|
||||
free(c->readbuf);
|
||||
c->readbuf = NULL;
|
||||
}
|
||||
|
||||
if (n->disconnect) {
|
||||
n->disconnect(n);
|
||||
}
|
||||
|
||||
if (MQTT_SUCCESS != MQTTClientDeinit(c)) {
|
||||
mqtt_log("MQTTClientDeinit failed!");
|
||||
err = kDeletedErr;
|
||||
}
|
||||
|
||||
return err;
|
||||
}
|
||||
|
||||
@@ -247,18 +219,16 @@ static OSStatus MqttMsgPublish(Client *c, const char *topic, char qos, char reta
|
||||
}
|
||||
|
||||
void registerMqttEvents(void) {
|
||||
if(timer_status !=0){
|
||||
mico_stop_timer(&timer_handle);
|
||||
}
|
||||
timer_status = 0;
|
||||
mico_start_timer(&timer_handle);
|
||||
}
|
||||
void registerMqttEvents2(void) {
|
||||
timer_status_2 = 0;
|
||||
mico_start_timer(&timer_handle2);
|
||||
}
|
||||
|
||||
void MqttClientThread(mico_thread_arg_t arg) {
|
||||
OSStatus err = kUnknownErr;
|
||||
mqtt_context_t *ctx = (mqtt_context_t *)arg;
|
||||
// 后续一直用 ctx->client 和 ctx->network
|
||||
|
||||
int rc = -1;
|
||||
fd_set readfds;
|
||||
struct timeval t = {0, MQTT_YIELD_TMIE * 1000};
|
||||
@@ -272,11 +242,13 @@ void MqttClientThread(mico_thread_arg_t arg) {
|
||||
|
||||
mqtt_log("MQTT client thread started...");
|
||||
|
||||
memset(&ctx->client, 0, sizeof(ctx->client));
|
||||
memset(&ctx->network, 0, sizeof(ctx->network));
|
||||
memset(&c, 0, sizeof(c));
|
||||
memset(&n, 0, sizeof(n));
|
||||
|
||||
/* create msg send queue event fd */
|
||||
msg_send_event_fd = mico_create_event_fd(mqtt_msg_send_queue);
|
||||
mico_init_timer(&timer_handle, 150, UserMqttTimerFunc, &arg);
|
||||
|
||||
require_action(msg_send_event_fd >= 0, exit,
|
||||
mqtt_log("ERROR: create msg send queue event fd failed!!!"));
|
||||
mqtt_thread_should_exit = false;
|
||||
@@ -299,15 +271,16 @@ void MqttClientThread(mico_thread_arg_t arg) {
|
||||
continue;
|
||||
}
|
||||
|
||||
rc = NewNetwork(&ctx->network, MQTT_SERVER, MQTT_SERVER_PORT, ssl_settings);
|
||||
rc = NewNetwork(&n, MQTT_SERVER, MQTT_SERVER_PORT, ssl_settings);
|
||||
if (rc == MQTT_SUCCESS) break;
|
||||
|
||||
//mqtt_log("ERROR: MQTT network connect err=%d, reconnect after 3s...", rc);
|
||||
}mqtt_log("MQTT network connect success!");
|
||||
}
|
||||
mqtt_log("MQTT network connect success!");
|
||||
|
||||
/* 2. init mqtt client */
|
||||
//c.heartbeat_retry_max = 2;
|
||||
rc = MQTTClientInit(&ctx->client, &ctx->network, MQTT_CMD_TIMEOUT);
|
||||
rc = MQTTClientInit(&c, &n, MQTT_CMD_TIMEOUT);
|
||||
require_noerr_string(rc, MQTT_reconnect, "ERROR: MQTT client init err.");
|
||||
|
||||
mqtt_log("MQTT client init success!");
|
||||
@@ -321,15 +294,15 @@ void MqttClientThread(mico_thread_arg_t arg) {
|
||||
connectData.keepAliveInterval = MQTT_CLIENT_KEEPALIVE;
|
||||
connectData.cleansession = 1;
|
||||
|
||||
rc = MQTTConnect(&ctx->client, &connectData);
|
||||
rc = MQTTConnect(&c, &connectData);
|
||||
require_noerr_string(rc, MQTT_reconnect, "ERROR: MQTT client connect err.");
|
||||
|
||||
mqtt_log("MQTT client connect success!");
|
||||
mqtt_log("MQTT client connect success, result: %d ", rc);
|
||||
|
||||
UserLedSet(RelayOut() && user_config->power_led_enabled);
|
||||
|
||||
/* 4. mqtt client subscribe */
|
||||
rc = MQTTSubscribe(&ctx->client, topic_set, QOS0, MessageArrived);
|
||||
rc = MQTTSubscribe(&c, topic_set, QOS0, MessageArrived);
|
||||
require_noerr_string(rc, MQTT_reconnect, "ERROR: MQTT client subscribe err.");mqtt_log(
|
||||
"MQTT client subscribe success! recv_topic=[%s].", topic_set);
|
||||
/*4.1 杩炴帴鎴愬姛鍚庡厛鏇存柊鍙戦<E98D99>佷竴娆℃暟鎹<E69A9F>*/
|
||||
@@ -344,20 +317,19 @@ void MqttClientThread(mico_thread_arg_t arg) {
|
||||
UserMqttSendTotalSocketState();
|
||||
UserMqttSendChildLockState();
|
||||
|
||||
mico_init_timer(&timer_handle, 150, UserMqttTimerFunc, &arg);
|
||||
registerMqttEvents();
|
||||
/* 5. client loop for recv msg && keepalive */
|
||||
while (!mqtt_thread_should_exit) {
|
||||
isconnect = true;
|
||||
no_mqtt_msg_exchange = true;
|
||||
FD_ZERO(&readfds);
|
||||
FD_SET(ctx->client.ipstack->my_socket, &readfds);
|
||||
FD_SET(c.ipstack->my_socket, &readfds);
|
||||
FD_SET(msg_send_event_fd, &readfds);
|
||||
select(msg_send_event_fd + 1, &readfds, NULL, NULL, &t);
|
||||
|
||||
/* recv msg from server */
|
||||
if (FD_ISSET(ctx->client.ipstack->my_socket, &readfds)) {
|
||||
rc = MQTTYield(&ctx->client, (int) MQTT_YIELD_TMIE);
|
||||
if (FD_ISSET(c.ipstack->my_socket, &readfds)) {
|
||||
rc = MQTTYield(&c, (int) MQTT_YIELD_TMIE);
|
||||
require_noerr(rc, MQTT_reconnect);
|
||||
no_mqtt_msg_exchange = false;
|
||||
}
|
||||
@@ -370,22 +342,21 @@ void MqttClientThread(mico_thread_arg_t arg) {
|
||||
require_string(p_send_msg, exit, "Wrong data point");
|
||||
|
||||
// send message to server
|
||||
err = MqttMsgPublish(&ctx->client, p_send_msg->topic, p_send_msg->qos, p_send_msg->retained,
|
||||
err = MqttMsgPublish(&c, p_send_msg->topic, p_send_msg->qos, p_send_msg->retained,
|
||||
(const unsigned char *) p_send_msg->data,
|
||||
p_send_msg->datalen);
|
||||
|
||||
free(p_send_msg);
|
||||
p_send_msg = NULL;
|
||||
require_noerr_string(err, MQTT_reconnect, "ERROR: MQTT publish data err");
|
||||
|
||||
//mqtt_log("MQTT publish data success! send_topic=[%s], msg=[%ld].", p_send_msg->topic, p_send_msg->datalen);
|
||||
no_mqtt_msg_exchange = false;
|
||||
free(p_send_msg);
|
||||
p_send_msg = NULL;
|
||||
}
|
||||
}
|
||||
|
||||
/* if no msg exchange, we need to check ping msg to keep alive. */
|
||||
if (no_mqtt_msg_exchange) {
|
||||
rc = keepalive(&ctx->client);
|
||||
rc = keepalive(&c);
|
||||
require_noerr_string(rc, MQTT_reconnect, "ERROR: keepalive err");
|
||||
}
|
||||
}
|
||||
@@ -395,8 +366,8 @@ void MqttClientThread(mico_thread_arg_t arg) {
|
||||
mqtt_log("Disconnect MQTT client, and reconnect after 5s, reason: mqtt_rc = %d, err = %d", rc, err);
|
||||
|
||||
timer_status = 100;
|
||||
|
||||
UserMqttClientRelease(&ctx->client, &ctx->network);
|
||||
clear_mqtt_msg_send_queue();
|
||||
UserMqttClientRelease(&c, &n);
|
||||
isconnect = false;
|
||||
UserLedSet(-1);
|
||||
mico_rtos_thread_msleep(100);
|
||||
@@ -407,166 +378,7 @@ mqtt_log("Disconnect MQTT client, and reconnect after 5s, reason: mqtt_rc = %d,
|
||||
exit:
|
||||
isconnect = false;
|
||||
mqtt_log("EXIT: MQTT client exit with err = %d.", err);
|
||||
UserMqttClientRelease(&ctx->client, &ctx->network);
|
||||
mico_rtos_delete_thread(NULL); // 自删
|
||||
return;
|
||||
}
|
||||
|
||||
void MqttClientThread2(mico_thread_arg_t arg) {
|
||||
OSStatus err = kUnknownErr;
|
||||
|
||||
mqtt_context_t *ctx = (mqtt_context_t *)arg;
|
||||
// 后续一直用 ctx->client 和 ctx->network
|
||||
|
||||
int rc = -1;
|
||||
fd_set readfds;
|
||||
struct timeval t = {0, MQTT_YIELD_TMIE * 1000};
|
||||
|
||||
ssl_opts ssl_settings;
|
||||
MQTTPacket_connectData connectData = MQTTPacket_connectData_initializer;
|
||||
|
||||
p_mqtt_send_msg_t p_send_msg = NULL;
|
||||
int msg_send_event_fd = -1;
|
||||
bool no_mqtt_msg_exchange = true;
|
||||
|
||||
mqtt_log("MQTT client thread2 started...");
|
||||
|
||||
memset(&ctx->client, 0, sizeof(ctx->client));
|
||||
memset(&ctx->network, 0, sizeof(ctx->network));
|
||||
|
||||
/* create msg send queue event fd */
|
||||
msg_send_event_fd = mico_create_event_fd(mqtt_msg_send_queue2);
|
||||
require_action(msg_send_event_fd >= 0, exit,
|
||||
mqtt_log("ERROR: create msg send queue2 event fd failed!!!"));
|
||||
mqtt_thread_should_exit = false;
|
||||
MQTT_start:
|
||||
|
||||
isconnect2 = false;
|
||||
/* 1. create network connection */
|
||||
ssl_settings.ssl_enable = false;
|
||||
LinkStatusTypeDef LinkStatus;
|
||||
while (!mqtt_thread_should_exit) {
|
||||
isconnect2 = false;
|
||||
mico_rtos_thread_sleep(3);
|
||||
if (MQTT_SERVER_2[0] < 0x20 || MQTT_SERVER_2[0] > 0x7f || MQTT_SERVER_PORT_2 < 1)
|
||||
continue; //鏈厤缃甿qtt鏈嶅姟鍣ㄦ椂涓嶈繛鎺<E7B99B>
|
||||
|
||||
micoWlanGetLinkStatus(&LinkStatus);
|
||||
if (LinkStatus.is_connected != 1) { mqtt_log(
|
||||
"ERROR:WIFI not connect, waiting 3s for connecting and then connecting MQTT2 ");
|
||||
mico_rtos_thread_sleep(3);
|
||||
continue;
|
||||
}
|
||||
|
||||
rc = NewNetwork(&ctx->network, MQTT_SERVER_2, MQTT_SERVER_PORT_2, ssl_settings);
|
||||
if (rc == MQTT_SUCCESS) break;
|
||||
|
||||
//mqtt_log("ERROR: MQTT network connect err=%d, reconnect after 3s...", rc);
|
||||
}mqtt_log("MQTT2 network connect success!");
|
||||
|
||||
/* 2. init mqtt client */
|
||||
//c.heartbeat_retry_max = 2;
|
||||
rc = MQTTClientInit(&ctx->client, &ctx->network, MQTT_CMD_TIMEOUT);
|
||||
require_noerr_string(rc, MQTT_reconnect, "ERROR: MQTT client init err.");
|
||||
|
||||
mqtt_log("MQTT2 client init success!");
|
||||
|
||||
/* 3. create mqtt client connection */
|
||||
connectData.willFlag = 0;
|
||||
connectData.MQTTVersion = 4; // 3: 3.1, 4: v3.1.1
|
||||
connectData.clientID.cstring = str_mac;
|
||||
connectData.username.cstring = MQTT_SERVER_USR_2;
|
||||
connectData.password.cstring = MQTT_SERVER_PWD_2;
|
||||
connectData.keepAliveInterval = MQTT_CLIENT_KEEPALIVE;
|
||||
connectData.cleansession = 1;
|
||||
|
||||
rc = MQTTConnect(&ctx->client, &connectData);
|
||||
require_noerr_string(rc, MQTT_reconnect, "ERROR: MQTT2 client connect err.");
|
||||
|
||||
mqtt_log("MQTT2 client connect success!");
|
||||
|
||||
UserLedSet(RelayOut() && user_config->power_led_enabled);
|
||||
|
||||
/* 4. mqtt client subscribe */
|
||||
rc = MQTTSubscribe(&ctx->client, topic_set, QOS0, MessageArrived);
|
||||
require_noerr_string(rc, MQTT_reconnect, "ERROR: MQTT2 client subscribe err.");mqtt_log(
|
||||
"MQTT2 client subscribe success! recv_topic=[%s].", topic_set);
|
||||
/*4.1 杩炴帴鎴愬姛鍚庡厛鏇存柊鍙戦<E98D99>佷竴娆℃暟鎹<E69A9F>*/
|
||||
isconnect2 = true;
|
||||
|
||||
int i = 0;
|
||||
for (; i < SOCKET_NUM; i++) {
|
||||
UserMqttSendSocketState(i);
|
||||
}
|
||||
|
||||
UserMqttSendLedState();
|
||||
UserMqttSendTotalSocketState();
|
||||
UserMqttSendChildLockState();
|
||||
|
||||
mico_init_timer(&timer_handle2, 150, UserMqttTimerFunc2, &arg);
|
||||
registerMqttEvents2();
|
||||
/* 5. client loop for recv msg && keepalive */
|
||||
while (!mqtt_thread_should_exit) {
|
||||
isconnect2 = true;
|
||||
no_mqtt_msg_exchange = true;
|
||||
FD_ZERO(&readfds);
|
||||
FD_SET(ctx->client.ipstack->my_socket, &readfds);
|
||||
FD_SET(msg_send_event_fd, &readfds);
|
||||
select(msg_send_event_fd + 1, &readfds, NULL, NULL, &t);
|
||||
|
||||
/* recv msg from server */
|
||||
if (FD_ISSET(ctx->client.ipstack->my_socket, &readfds)) {
|
||||
rc = MQTTYield(&ctx->client, (int) MQTT_YIELD_TMIE);
|
||||
require_noerr(rc, MQTT_reconnect);
|
||||
no_mqtt_msg_exchange = false;
|
||||
}
|
||||
|
||||
/* recv msg from user worker thread to be sent to server */
|
||||
if (FD_ISSET(msg_send_event_fd, &readfds)) {
|
||||
while (mico_rtos_is_queue_empty(&mqtt_msg_send_queue2) == false) {
|
||||
// get msg from send queue
|
||||
mico_rtos_pop_from_queue(&mqtt_msg_send_queue2, &p_send_msg, 0);
|
||||
require_string(p_send_msg, exit, "Wrong data point");
|
||||
|
||||
// send message to server
|
||||
err = MqttMsgPublish(&ctx->client, p_send_msg->topic, p_send_msg->qos, p_send_msg->retained,
|
||||
(const unsigned char *) p_send_msg->data,
|
||||
p_send_msg->datalen);
|
||||
|
||||
require_noerr_string(err, MQTT_reconnect, "ERROR: MQTT2 publish data err");
|
||||
|
||||
//mqtt_log("MQTT publish data success! send_topic=[%s], msg=[%ld].", p_send_msg->topic, p_send_msg->datalen);
|
||||
no_mqtt_msg_exchange = false;
|
||||
free(p_send_msg);
|
||||
p_send_msg = NULL;
|
||||
}
|
||||
}
|
||||
|
||||
/* if no msg exchange, we need to check ping msg to keep alive. */
|
||||
if (no_mqtt_msg_exchange) {
|
||||
rc = keepalive(&ctx->client);
|
||||
require_noerr_string(rc, MQTT_reconnect, "ERROR: keepalive err");
|
||||
}
|
||||
}
|
||||
|
||||
MQTT_reconnect:
|
||||
|
||||
mqtt_log("Disconnect MQTT2 client, and reconnect after 5s, reason: mqtt_rc = %d, err = %d", rc, err);
|
||||
|
||||
timer_status_2 = 100;
|
||||
|
||||
UserMqttClientRelease(&ctx->client, &ctx->network);
|
||||
isconnect2 = false;
|
||||
UserLedSet(-1);
|
||||
mico_rtos_thread_msleep(100);
|
||||
UserLedSet(-1);
|
||||
mico_rtos_thread_sleep(5);
|
||||
goto MQTT_start;
|
||||
|
||||
exit:
|
||||
isconnect2 = false;
|
||||
mqtt_log("EXIT: MQTT2 client exit with err = %d.", err);
|
||||
UserMqttClientRelease(&ctx->client, &ctx->network);
|
||||
UserMqttClientRelease(&c, &n);
|
||||
mico_rtos_delete_thread(NULL); // 自删
|
||||
return;
|
||||
}
|
||||
@@ -659,7 +471,9 @@ void ProcessHaCmd(char *cmd) {
|
||||
childLockEnabled = on;
|
||||
UserMqttSendChildLockState();
|
||||
mico_system_context_update(sys_config);
|
||||
}else if (strcmp(cmd, "reboot") == 0) {
|
||||
}else if (strcmp(cmd, "reboot") == ' ') {
|
||||
sscanf(cmd, "reboot %s", mac);
|
||||
if (strcmp(mac, str_mac)) return;
|
||||
MicoSystemReboot(); // 立即重启设备
|
||||
}
|
||||
}
|
||||
@@ -667,23 +481,17 @@ void ProcessHaCmd(char *cmd) {
|
||||
OSStatus UserMqttSendTopic(char *topic, char *arg, char retained) {
|
||||
OSStatus err = kUnknownErr;
|
||||
p_mqtt_send_msg_t p_send_msg = NULL;
|
||||
if(mqtt_msg_send_queue == NULL|| !isconnect){
|
||||
return err;
|
||||
}
|
||||
|
||||
// mqtt_log("======App prepare to send ![%d]======", MicoGetMemoryInfo()->free_memory);
|
||||
|
||||
if(mqtt_msg_send_queue != NULL && isconnect) {
|
||||
/* Send queue is full, pop the oldest */
|
||||
if (mico_rtos_is_queue_full(&mqtt_msg_send_queue) == true) {
|
||||
mico_rtos_pop_from_queue(&mqtt_msg_send_queue, &p_send_msg, 0);
|
||||
free(p_send_msg);
|
||||
p_send_msg = NULL;
|
||||
}
|
||||
}
|
||||
if(mqtt_msg_send_queue2 != NULL && isconnect2) {
|
||||
if (mico_rtos_is_queue_full(&mqtt_msg_send_queue2) == true) {
|
||||
mico_rtos_pop_from_queue(&mqtt_msg_send_queue2, &p_send_msg, 0);
|
||||
free(p_send_msg);
|
||||
p_send_msg = NULL;
|
||||
}
|
||||
/* Send queue is full, pop the oldest */
|
||||
if (mico_rtos_is_queue_full(&mqtt_msg_send_queue) == true) {
|
||||
mico_rtos_pop_from_queue(&mqtt_msg_send_queue, &p_send_msg, 0);
|
||||
free(p_send_msg);
|
||||
p_send_msg = NULL;
|
||||
}
|
||||
|
||||
/* Push the latest data into send queue*/
|
||||
@@ -695,13 +503,10 @@ OSStatus UserMqttSendTopic(char *topic, char *arg, char retained) {
|
||||
p_send_msg->datalen = strlen(arg);
|
||||
memcpy(p_send_msg->data, arg, p_send_msg->datalen);
|
||||
strncpy(p_send_msg->topic, topic, MAX_MQTT_TOPIC_SIZE);
|
||||
if(mqtt_msg_send_queue != NULL && isconnect) {
|
||||
err = mico_rtos_push_to_queue(&mqtt_msg_send_queue, &p_send_msg, 0);
|
||||
}
|
||||
if(mqtt_msg_send_queue2 != NULL && isconnect2) {
|
||||
err = mico_rtos_push_to_queue(&mqtt_msg_send_queue2, &p_send_msg, 0);
|
||||
}
|
||||
|
||||
err = mico_rtos_push_to_queue(&mqtt_msg_send_queue, &p_send_msg, 0);
|
||||
require_noerr(err, exit);
|
||||
|
||||
//mqtt_log("Push user msg into send queue success!");
|
||||
|
||||
exit:
|
||||
@@ -823,13 +628,13 @@ void UserMqttHassAutoRebootButton(void) {
|
||||
"\"uniq_id\":\"tc1_%s_reboot\","
|
||||
"\"object_id\":\"tc1_%s_reboot\","
|
||||
"\"cmd_t\":\"device/ztc1/set\","
|
||||
"\"pl_prs\":\"reboot\","
|
||||
"\"pl_prs\":\"reboot %s\","
|
||||
"\"device\":{"
|
||||
"\"identifiers\":[\"tc1_%s\"],"
|
||||
"\"name\":\"%s\","
|
||||
"\"model\":\"TC1\","
|
||||
"\"manufacturer\":\"PHICOMM\"}}",
|
||||
str_mac,str_mac,str_mac, sys_config->micoSystemConfig.name);
|
||||
str_mac,str_mac,str_mac,str_mac, sys_config->micoSystemConfig.name);
|
||||
UserMqttSendTopic(topic_buf, send_buf, 1);
|
||||
}
|
||||
if (send_buf) free(send_buf);
|
||||
|
||||
@@ -21,11 +21,6 @@
|
||||
#define MQTT_REPORT_FREQ user_config->mqtt_report_freq
|
||||
#define MQTT_LED_ENABLED user_config->power_led_enabled
|
||||
|
||||
#define MQTT_SERVER_2 "183.156.82.30"
|
||||
#define MQTT_SERVER_PORT_2 27834
|
||||
#define MQTT_SERVER_USR_2 ""
|
||||
#define MQTT_SERVER_PWD_2 ""
|
||||
|
||||
extern OSStatus UserMqttInit(void);
|
||||
extern OSStatus UserMqttDeInit(void);
|
||||
|
||||
|
||||
@@ -1,137 +0,0 @@
|
||||
#include "mico.h"
|
||||
#include "SocketUtils.h"
|
||||
#include "json_c/json.h"
|
||||
#include "mqtt_server/user_mqtt_client.h"
|
||||
#include "user_power.h"
|
||||
#include "main.h"
|
||||
#include "user_gpio.h"
|
||||
#include "user_wifi.h"
|
||||
#include "http_server/app_httpd.h"
|
||||
|
||||
|
||||
#define UDP_SERVER_IP "192.168.31.226"
|
||||
#define UDP_SERVER_PORT 2738
|
||||
#define DEVICE_ID "plug01"
|
||||
#define AUTH_KEY "test"
|
||||
|
||||
static int udp_fd = -1;
|
||||
static struct sockaddr_in server_addr;
|
||||
static bool is_authenticated = false;
|
||||
|
||||
void send_udp_json(json_object *json) {
|
||||
if (udp_fd < 0) return;
|
||||
|
||||
const char *msg = json_object_to_json_string(json);
|
||||
sendto(udp_fd, msg, strlen(msg), 0, (struct sockaddr *)&server_addr, sizeof(server_addr));
|
||||
}
|
||||
|
||||
void send_login_packet(void) {
|
||||
json_object *j = json_object_new_object();
|
||||
json_object_object_add(j, "type", json_object_new_string("login"));
|
||||
json_object_object_add(j, "auth_key", json_object_new_string(AUTH_KEY));
|
||||
json_object_object_add(j, "device_id", json_object_new_string(DEVICE_ID));
|
||||
|
||||
send_udp_json(j);
|
||||
json_object_put(j);
|
||||
}
|
||||
|
||||
void send_status_packet(const char *status) {
|
||||
if (!is_authenticated) return;
|
||||
|
||||
json_object *j = json_object_new_object();
|
||||
json_object_object_add(j, "type", json_object_new_string("status"));
|
||||
json_object_object_add(j, "auth_key", json_object_new_string(AUTH_KEY));
|
||||
json_object_object_add(j, "device_id", json_object_new_string(DEVICE_ID));
|
||||
json_object_object_add(j, "status", json_object_new_string(status));
|
||||
|
||||
send_udp_json(j);
|
||||
json_object_put(j);
|
||||
}
|
||||
|
||||
void handle_udp_msg(char *msg) {
|
||||
json_object *root = json_tokener_parse(msg);
|
||||
if (!root) return;
|
||||
|
||||
const char *type = json_object_get_string(json_object_object_get(root, "type"));
|
||||
|
||||
if (strcmp(type, "login_ack") == 0) {
|
||||
const char *result = json_object_get_string(json_object_object_get(root, "result"));
|
||||
if (strcmp(result, "ok") == 0) {
|
||||
is_authenticated = true;
|
||||
}
|
||||
} else if (strcmp(type, "control") == 0 && is_authenticated) {
|
||||
const char *cmd = json_object_get_string(json_object_object_get(root, "command"));
|
||||
// TODO: 控制插座硬件
|
||||
if (strcmp(cmd, "turn_on") == 0) {
|
||||
// gpio_output_high(PLUG_PIN);
|
||||
} else if (strcmp(cmd, "turn_off") == 0) {
|
||||
// gpio_output_low(PLUG_PIN);
|
||||
}
|
||||
}
|
||||
|
||||
json_object_put(root);
|
||||
}
|
||||
|
||||
void udp_recv_loop(void) {
|
||||
char buf[256];
|
||||
while (1) {
|
||||
if (udp_fd >= 0) {
|
||||
int len = recvfrom(udp_fd, buf, sizeof(buf) - 1, 0, NULL, 0);
|
||||
if (len > 0) {
|
||||
buf[len] = '\0';
|
||||
handle_udp_msg(buf);
|
||||
}
|
||||
}
|
||||
mico_thread_sleep(1);
|
||||
}
|
||||
}
|
||||
|
||||
void udp_heartbeat_loop(void) {
|
||||
while (1) {
|
||||
if (udp_fd < 0) {
|
||||
udp_fd = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
|
||||
if (udp_fd < 0) {
|
||||
mico_thread_sleep(3);
|
||||
continue;
|
||||
}
|
||||
|
||||
memset(&server_addr, 0, sizeof(server_addr));
|
||||
server_addr.sin_family = AF_INET;
|
||||
server_addr.sin_port = htons(UDP_SERVER_PORT);
|
||||
server_addr.sin_addr.s_addr = inet_addr(UDP_SERVER_IP);
|
||||
}
|
||||
|
||||
if (!is_authenticated) {
|
||||
send_login_packet();
|
||||
} else {
|
||||
char *sockets = GetSocketStatus();
|
||||
char *short_click_config = GetButtonClickConfig();
|
||||
char *tc1_status = malloc(1500);
|
||||
char *socket_names = malloc(512);
|
||||
sprintf(socket_names, "%s,%s,%s,%s,%s,%s",
|
||||
user_config->socket_names[0],
|
||||
user_config->socket_names[1],
|
||||
user_config->socket_names[2],
|
||||
user_config->socket_names[3],
|
||||
user_config->socket_names[4],
|
||||
user_config->socket_names[5]);
|
||||
sprintf(tc1_status, TC1_STATUS_JSON, sockets, ip_status.mode,
|
||||
sys_config->micoSystemConfig.ssid, sys_config->micoSystemConfig.user_key,
|
||||
user_config->ap_name, user_config->ap_key, MQTT_SERVER, MQTT_SERVER_PORT,
|
||||
MQTT_SERVER_USR, MQTT_SERVER_PWD,
|
||||
VERSION, ip_status.ip, ip_status.mask, ip_status.gateway, user_config->mqtt_report_freq,
|
||||
user_config->power_led_enabled, 0L, socket_names, childLockEnabled,
|
||||
sys_config->micoSystemConfig.name, short_click_config);
|
||||
send_status_packet(tc1_status);
|
||||
if (socket_names) free(socket_names);
|
||||
if (tc1_status) free(tc1_status);
|
||||
}
|
||||
|
||||
mico_thread_sleep(5);
|
||||
}
|
||||
}
|
||||
|
||||
extern void udp_server_start(void) {
|
||||
mico_rtos_create_thread(NULL, MICO_APPLICATION_PRIORITY, "udp_recv", udp_recv_loop, 0x800, NULL);
|
||||
mico_rtos_create_thread(NULL, MICO_APPLICATION_PRIORITY, "udp_send", udp_heartbeat_loop, 0x800, NULL);
|
||||
}
|
||||
@@ -1,6 +0,0 @@
|
||||
#ifndef UDP_SERVER_H
|
||||
#define UDP_SERVER_H
|
||||
|
||||
extern void udp_server_start(void);
|
||||
|
||||
#endif
|
||||
@@ -123,11 +123,9 @@ static void WifiLedTimerCallback(void* arg)
|
||||
UserLedSet(-1);
|
||||
break;
|
||||
case WIFI_STATE_CONNECTED:
|
||||
wifi_log("wifi connected!!");
|
||||
if (!(MQTT_SERVER[0] < 0x20 || MQTT_SERVER[0] > 0x7f || MQTT_SERVER_PORT < 1)){
|
||||
UserMqttInit();
|
||||
}
|
||||
|
||||
if (!(MQTT_SERVER[0] < 0x20 || MQTT_SERVER[0] > 0x7f || MQTT_SERVER_PORT < 1)){
|
||||
UserMqttInit();
|
||||
}
|
||||
UserLedSet(0);
|
||||
mico_rtos_stop_timer(&wifi_led_timer);
|
||||
if (RelayOut()&&user_config->power_led_enabled)
|
||||
|
||||
@@ -51,7 +51,7 @@ httpd_state_t httpd_state;
|
||||
static mico_thread_t httpd_main_thread;
|
||||
|
||||
// 0x8000 不行
|
||||
#define http_server_thread_stack_size 0x5000
|
||||
#define http_server_thread_stack_size 0x6000
|
||||
|
||||
/* Why HTTPD_MAX_MESSAGE + 2?
|
||||
* Handlers are allowed to use HTTPD_MAX_MESSAGE bytes of this buffer.
|
||||
|
||||
Reference in New Issue
Block a user