espressif/esp-mqtt
1
0
Fork 1
mirror of https://github.com/espressif/esp-mqtt.git synced 2025-11-11 12:32:46 +01:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'espnode/master' of https://github.com/jnwatts/espmqtt into jnwatts-espnode/master

Browse Source
This commit is contained in:
jarek
2017-05-14 22:38:36 +02:00
parent 629dba3c23 8d1e6ec202
commit 7da9e28a86
3 changed files with 118 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
mqtt.c
View File

@@ -13,6 +13,9 @@
#include "lwip/sockets.h"
#include "lwip/dns.h"
#include "lwip/netdb.h"
#if defined(CONFIG_MQTT_SECURITY_ON)
#include "openssl/ssl.h"
#endif
#include "ringbuf.h"
#include "mqtt.h"
@@ -176,6 +179,65 @@ void closeclient(mqtt_client *client)
#endif
}
int mqtt_read(mqtt_client *client, void *buffer, int len, int timeout_ms)
{
int result;
struct timeval tv;
if (timeout_ms > 0) {
tv.tv_sec = 0;
tv.tv_usec = timeout_ms * 1000;
while (tv.tv_usec > 1000 * 1000) {
tv.tv_usec -= 1000 * 1000;
tv.tv_sec++;
}
setsockopt(client->socket, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
}
#if defined(CONFIG_MQTT_SECURITY_ON)
result = SSL_read(client->ssl, buffer, len);
#else
result = read(client->socket, buffer, len);
#endif
if (timeout_ms > 0) {
tv.tv_sec = 0;
tv.tv_usec = 0;
setsockopt(client->socket, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
}
return result;
}
int mqtt_write(mqtt_client *client, const void *buffer, int len, int timeout_ms)
{
int result;
struct timeval tv;
if (timeout_ms > 0) {
tv.tv_sec = 0;
tv.tv_usec = timeout_ms * 1000;
while (tv.tv_usec > 1000 * 1000) {
tv.tv_usec -= 1000 * 1000;
tv.tv_sec++;
}
setsockopt(client->socket, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
}
#if defined(CONFIG_MQTT_SECURITY_ON)
result = SSL_write(client->ssl, buffer, len)
#else
result = write(client->socket, buffer, len);
#endif
if (timeout_ms > 0) {
tv.tv_sec = 0;
tv.tv_usec = 0;
setsockopt(client->socket, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
}
return result;
}
/*
* mqtt_connect
* input - client
@@ -184,12 +246,7 @@ void closeclient(mqtt_client *client)
static bool mqtt_connect(mqtt_client *client)
{
int write_len, read_len, connect_rsp_code;
struct timeval tv;
tv.tv_sec = 10; /* 30 Secs Timeout */
tv.tv_usec = 0; // Not init'ing this can cause strange errors
setsockopt(client->socket, SOL_SOCKET, SO_RCVTIMEO, (char *)&tv, sizeof(struct timeval));
mqtt_msg_init(&client->mqtt_state.mqtt_connection,
client->mqtt_state.out_buffer,
@@ -203,16 +260,13 @@ static bool mqtt_connect(mqtt_client *client)
client->mqtt_state.pending_msg_type,
client->mqtt_state.pending_msg_id);
write_len = ClientWrite(
write_len = client->settings->write_cb(client,
client->mqtt_state.outbound_message->data,
client->mqtt_state.outbound_message->length);
client->mqtt_state.outbound_message->length, 0);
mqtt_info("Reading MQTT CONNECT response message");
read_len = ClientRead(client->mqtt_state.in_buffer, CONFIG_MQTT_BUFFER_SIZE_BYTE);
tv.tv_sec = 0; /* No timeout */
setsockopt(client->socket, SOL_SOCKET, SO_RCVTIMEO, (char *)&tv, sizeof(struct timeval));
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");
@@ -258,7 +312,7 @@ void mqtt_sending_task(void *pvParameters)
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_id = mqtt_get_id(client->mqtt_state.out_buffer, send_len);
ClientWrite(client->mqtt_state.out_buffer, send_len);
client->settings->write_cb(client, client->mqtt_state.out_buffer, send_len, 0);
//TODO: Check sending type, to callback publish message
msg_len -= send_len;
@@ -275,9 +329,9 @@ void mqtt_sending_task(void *pvParameters)
client->mqtt_state.pending_msg_id = mqtt_get_id(client->mqtt_state.outbound_message->data,
client->mqtt_state.outbound_message->length);
mqtt_info("Sending pingreq");
ClientWrite(
client->settings->write_cb(client,
client->mqtt_state.outbound_message->data,
client->mqtt_state.outbound_message->length);
client->mqtt_state.outbound_message->length, 0);
}
}
}
@@ -315,7 +369,7 @@ void deliver_publish(mqtt_client *client, uint8_t *message, int length)
if (client->mqtt_state.message_length_read >= client->mqtt_state.message_length)
break;
len_read = ClientRead(client->mqtt_state.in_buffer, CONFIG_MQTT_BUFFER_SIZE_BYTE);
len_read = client->settings->read_cb(client, client->mqtt_state.in_buffer, CONFIG_MQTT_BUFFER_SIZE_BYTE, 0);
client->mqtt_state.message_length_read += len_read;
} while (1);
@@ -329,7 +383,7 @@ void mqtt_start_receive_schedule(mqtt_client *client)
while (1) {
read_len = ClientRead(client->mqtt_state.in_buffer, CONFIG_MQTT_BUFFER_SIZE_BYTE);
read_len = client->settings->read_cb(client, client->mqtt_state.in_buffer, CONFIG_MQTT_BUFFER_SIZE_BYTE, 0);
mqtt_info("Read len %d", read_len);
if (read_len == 0)
@@ -419,11 +473,11 @@ void mqtt_task(void *pvParameters)
mqtt_client *client = (mqtt_client *)pvParameters;
while (1) {
client_connect(client);
client->settings->connect_cb(client);
mqtt_info("Connected to server %s:%d", client->settings->host, client->settings->port);
if (!mqtt_connect(client)) {
closeclient(client);
client->settings->disconnect_cb(client);
continue;
//return;
}
@@ -436,7 +490,7 @@ void mqtt_task(void *pvParameters)
mqtt_info("mqtt_start_receive_schedule");
mqtt_start_receive_schedule(client);
closeclient(client);
client->settings->disconnect_cb(client);
vTaskDelete(xMqttSendingTask);
vTaskDelay(1000 / portTICK_RATE_MS);
@@ -484,6 +538,15 @@ mqtt_client *mqtt_start(mqtt_settings *settings)
client->socket = -1;
if (!client->settings->connect_cb)
client->settings->connect_cb = client_connect;
if (!client->settings->disconnect_cb)
client->settings->disconnect_cb = closeclient;
if (!client->settings->read_cb)
client->settings->read_cb = mqtt_read;
if (!client->settings->write_cb)
client->settings->write_cb = mqtt_write;
#if defined(CONFIG_MQTT_SECURITY_ON) // ENABLE MQTT OVER SSL
client->ctx = NULL;
client->ssl = NULL;