|
@@ -32,8 +32,8 @@
|
|
|
#define LOGW printf
|
|
#define LOGW printf
|
|
|
#endif
|
|
#endif
|
|
|
|
|
|
|
|
-
|
|
|
|
|
-#define CONN_PERIOD 5000 //ms
|
|
|
|
|
|
|
+#define KEEPLIVE 30 //s
|
|
|
|
|
+#define CONN_PERIOD 6000 //ms
|
|
|
#define SEND_PERIOD 1000 //ms
|
|
#define SEND_PERIOD 1000 //ms
|
|
|
#define SEND_PERIOD2 5000 //ms
|
|
#define SEND_PERIOD2 5000 //ms
|
|
|
|
|
|
|
@@ -59,6 +59,7 @@ typedef struct {
|
|
|
int quit;
|
|
int quit;
|
|
|
|
|
|
|
|
void *h;
|
|
void *h;
|
|
|
|
|
+ time_t last_recv_time;
|
|
|
}mqtt_conn_t;
|
|
}mqtt_conn_t;
|
|
|
typedef struct {
|
|
typedef struct {
|
|
|
int inited;
|
|
int inited;
|
|
@@ -156,7 +157,12 @@ static int get_account(mqtt_account_t *host, user_account_t *user)
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
-
|
|
|
|
|
|
|
+static int get_status(mqtt_conn_t *conn)
|
|
|
|
|
+{
|
|
|
|
|
+ mqtt_handle_t *h=(mqtt_handle_t*)conn->h;
|
|
|
|
|
+ int id=conn->ser.id;
|
|
|
|
|
+ return h->dm->mqttInfo.ser[id].status;
|
|
|
|
|
+}
|
|
|
static void set_status(mqtt_conn_t *conn, int flag)
|
|
static void set_status(mqtt_conn_t *conn, int flag)
|
|
|
{
|
|
{
|
|
|
mqtt_handle_t *h=(mqtt_handle_t*)conn->h;
|
|
mqtt_handle_t *h=(mqtt_handle_t*)conn->h;
|
|
@@ -230,7 +236,7 @@ static int pub_one(mg_conn_t *c, char *topic, char *data)
|
|
|
|
|
|
|
|
static int my_pub(mqtt_conn_t *conn, char *topic, char *data)
|
|
static int my_pub(mqtt_conn_t *conn, char *topic, char *data)
|
|
|
{
|
|
{
|
|
|
- if(conn->c) {
|
|
|
|
|
|
|
+ if(conn->c && get_status(conn)) {
|
|
|
pub_one(conn->c, topic, data);
|
|
pub_one(conn->c, topic, data);
|
|
|
}
|
|
}
|
|
|
return 0;
|
|
return 0;
|
|
@@ -387,12 +393,15 @@ static int conn_one(mqtt_conn_t *conn)
|
|
|
conn->c = mg_mqtt_connect(&conn->mgr, url, &opts, mqtt_fn, conn);
|
|
conn->c = mg_mqtt_connect(&conn->mgr, url, &opts, mqtt_fn, conn);
|
|
|
if(conn->c) {
|
|
if(conn->c) {
|
|
|
//mg_mqtt_bind(conn->c);
|
|
//mg_mqtt_bind(conn->c);
|
|
|
-
|
|
|
|
|
- sprintf(conn->ser.cid, "smartPDU_%s\n", h->prod_id);
|
|
|
|
|
- my_sub(conn->c, h->prod_id);
|
|
|
|
|
|
|
+ log_i("Trying connect MQTT Broker server ID: %d : %s...",conn->ser.id, conn->ser.server);
|
|
|
|
|
+ //my_sub(conn->c, h->prod_id);
|
|
|
r = 0;
|
|
r = 0;
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
+ else
|
|
|
|
|
+ {
|
|
|
|
|
+ if(get_status(conn)) mg_mqtt_ping(conn->c);
|
|
|
|
|
+ }
|
|
|
|
|
|
|
|
return r;
|
|
return r;
|
|
|
}
|
|
}
|
|
@@ -506,26 +515,49 @@ static void mqtt_fn(mg_conn_t *c, int ev, void *ev_data)
|
|
|
|
|
|
|
|
case MG_EV_MQTT_OPEN:
|
|
case MG_EV_MQTT_OPEN:
|
|
|
{
|
|
{
|
|
|
|
|
+ conn->c = c;
|
|
|
send_once(conn);
|
|
send_once(conn);
|
|
|
set_status(conn, 1);
|
|
set_status(conn, 1);
|
|
|
|
|
+ log_i("MQTT Broker server ID: %d : %s connected successful...",conn->ser.id, conn->ser.server);
|
|
|
|
|
+ conn->last_recv_time = time(NULL); // 初始化接收时间
|
|
|
}
|
|
}
|
|
|
break;
|
|
break;
|
|
|
|
|
|
|
|
case MG_EV_POLL:
|
|
case MG_EV_POLL:
|
|
|
{
|
|
{
|
|
|
my_send(conn);
|
|
my_send(conn);
|
|
|
|
|
+
|
|
|
|
|
+ // 检查连接状态,如果超过KEEPLIVE时间没有收到PINGRESP响应则断开连接.
|
|
|
|
|
+ //当前配置为每6s发送一次心跳包,总共5次30s超时未收到心跳包则断开连接。
|
|
|
|
|
+ time_t now = time(NULL);
|
|
|
|
|
+ if (now - conn->last_recv_time > (time_t)(KEEPLIVE) && get_status(conn)) {
|
|
|
|
|
+ log_e("MQTT Broker server ID: %d : %s PINGRESP timeout! Disconnecting...",conn->ser.id, conn->ser.server);
|
|
|
|
|
+ conn->c = NULL;
|
|
|
|
|
+ set_status(conn, 0);
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|
|
|
break;
|
|
break;
|
|
|
|
|
|
|
|
case MG_EV_MQTT_MSG:
|
|
case MG_EV_MQTT_MSG:
|
|
|
{
|
|
{
|
|
|
- struct mg_mqtt_message *mm=(struct mg_mqtt_message*)ev_data;
|
|
|
|
|
- if(mm && mm->topic.buf && mm->data.buf) {
|
|
|
|
|
- my_recv(mm->topic.buf, mm->data.buf);
|
|
|
|
|
|
|
+ struct mg_mqtt_message *msg=(struct mg_mqtt_message*)ev_data;
|
|
|
|
|
+ if(msg && msg->topic.buf && msg->data.buf) {
|
|
|
|
|
+ my_recv(msg->topic.buf, msg->data.buf);
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
break;
|
|
break;
|
|
|
|
|
|
|
|
|
|
+ case MG_EV_MQTT_CMD:
|
|
|
|
|
+ {
|
|
|
|
|
+ struct mg_mqtt_message *msg = (struct mg_mqtt_message *)ev_data;
|
|
|
|
|
+ if (msg && msg->cmd == MQTT_CMD_PINGRESP)
|
|
|
|
|
+ {
|
|
|
|
|
+ conn->last_recv_time = time(NULL); // 更新接收时间
|
|
|
|
|
+ log_i("Received PINGRESP from MQTT Broker server %d : %s !", conn->ser.id, conn->ser.server);
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ break;
|
|
|
|
|
+
|
|
|
case MG_EV_ERROR:
|
|
case MG_EV_ERROR:
|
|
|
{
|
|
{
|
|
|
//MG_ERROR(("___MG_EV_ERROR, %p %s", c->fd, (char *) ev_data));
|
|
//MG_ERROR(("___MG_EV_ERROR, %p %s", c->fd, (char *) ev_data));
|
|
@@ -535,8 +567,12 @@ static void mqtt_fn(mg_conn_t *c, int ev, void *ev_data)
|
|
|
|
|
|
|
|
case MG_EV_CLOSE:
|
|
case MG_EV_CLOSE:
|
|
|
{
|
|
{
|
|
|
- conn->c = NULL;
|
|
|
|
|
- set_status(conn, 0);
|
|
|
|
|
|
|
+ if (conn->c && conn->c == c)
|
|
|
|
|
+ {
|
|
|
|
|
+ conn->c = NULL;
|
|
|
|
|
+ set_status(conn, 0);
|
|
|
|
|
+ log_e("MQTT Broker server ID: %d : %s connected failed...", conn->ser.id, conn->ser.server);
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|
|
|
break;
|
|
break;
|
|
|
}
|
|
}
|
|
@@ -640,7 +676,7 @@ static int send_stat(mqtt_conn_t *conn, int type)
|
|
|
char buf[20],date[40],time[40];
|
|
char buf[20],date[40],time[40];
|
|
|
mqtt_handle_t *h=(mqtt_handle_t*)conn->h;
|
|
mqtt_handle_t *h=(mqtt_handle_t*)conn->h;
|
|
|
|
|
|
|
|
- if((!conn->c) || (type<0) || (type>=MQTT_PUB_MAX)) {
|
|
|
|
|
|
|
+ if((!conn->c)|| !get_status(conn) || (type<0) || (type>=MQTT_PUB_MAX)) {
|
|
|
return -1;
|
|
return -1;
|
|
|
}
|
|
}
|
|
|
|
|
|