feedc0de/esp-mqtt
1
0
Fork 0
forked from espressif/esp-mqtt
Code Pull Requests Activity

utilizing pingreq to check if server is still reachable

Browse Source
This commit is contained in:
Tijn Kooijmans
2017-07-26 13:46:34 +02:00
parent b5761d5312
commit f49ecd1b92
1 changed files with 24 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
mqtt.c
View File

@@ -259,7 +259,7 @@ static bool mqtt_connect(mqtt_client *client)
write_len = client->settings->write_cb(client, write_len = client->settings->write_cb(client,
client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->data,
client->mqtt_state.outbound_message->length, 0); client->mqtt_state.outbound_message->length, 0);
if(write_len < 0) { if (write_len <= 0) {
mqtt_error("Writing failed: %d", errno); mqtt_error("Writing failed: %d", errno);
return false; return false;
} }
@@ -304,7 +304,7 @@ void mqtt_sending_task(void *pvParameters)
{ {
mqtt_client *client = (mqtt_client *)pvParameters; mqtt_client *client = (mqtt_client *)pvParameters;
uint32_t msg_len; uint32_t msg_len;
int send_len; int send_len, read_len;
mqtt_info("mqtt_sending_task"); mqtt_info("mqtt_sending_task");
while (1) { while (1) {
@@ -319,8 +319,8 @@ void mqtt_sending_task(void *pvParameters)
rb_read(&client->send_rb, client->mqtt_state.out_buffer, send_len); rb_read(&client->send_rb, client->mqtt_state.out_buffer, send_len);
client->mqtt_state.pending_msg_type = mqtt_get_type(client->mqtt_state.out_buffer); client->mqtt_state.pending_msg_type = mqtt_get_type(client->mqtt_state.out_buffer);
client->mqtt_state.pending_msg_id = mqtt_get_id(client->mqtt_state.out_buffer, send_len); client->mqtt_state.pending_msg_id = mqtt_get_id(client->mqtt_state.out_buffer, send_len);
send_len = client->settings->write_cb(client, client->mqtt_state.out_buffer, send_len, 0); send_len = client->settings->write_cb(client, client->mqtt_state.out_buffer, send_len, 10 * 1000);
if(send_len < 0) { if (send_len <= 0) {
mqtt_info("Write error: %d", errno); mqtt_info("Write error: %d", errno);
break; // TODO is this right handling? break; // TODO is this right handling?
} }
@@ -328,10 +328,7 @@ void mqtt_sending_task(void *pvParameters)
//TODO: Check sending type, to callback publish message //TODO: Check sending type, to callback publish message
msg_len -= send_len; msg_len -= send_len;
} }
//invalidate keepalive timer
client->keepalive_tick = client->settings->keepalive / 2;
} }
else {
if (client->keepalive_tick > 0) client->keepalive_tick --; if (client->keepalive_tick > 0) client->keepalive_tick --;
else { else {
client->keepalive_tick = client->settings->keepalive / 2; client->keepalive_tick = client->settings->keepalive / 2;
@@ -343,9 +340,15 @@ void mqtt_sending_task(void *pvParameters)
client->settings->write_cb(client, client->settings->write_cb(client,
client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->data,
client->mqtt_state.outbound_message->length, 0); client->mqtt_state.outbound_message->length, 0);
read_len = client->settings->read_cb(client, client->mqtt_state.in_buffer, CONFIG_MQTT_BUFFER_SIZE_BYTE, 10 * 1000);
if (read_len < 0) {
mqtt_error("Error network response");
break;
} }
} }
} }
xMqttSendingTask = NULL;
vTaskDelete(NULL); vTaskDelete(NULL);
} }
@@ -399,6 +402,7 @@ void mqtt_start_receive_schedule(mqtt_client *client)
while (1) { while (1) {
if (terminate_mqtt) break; if (terminate_mqtt) break;
if (xMqttSendingTask == NULL) break;
read_len = client->settings->read_cb(client, client->mqtt_state.in_buffer, CONFIG_MQTT_BUFFER_SIZE_BYTE, 0); read_len = client->settings->read_cb(client, client->mqtt_state.in_buffer, CONFIG_MQTT_BUFFER_SIZE_BYTE, 0);