diff options
Diffstat (limited to 'libs/libcurl/src/mqtt.c')
| -rw-r--r-- | libs/libcurl/src/mqtt.c | 314 |
1 files changed, 229 insertions, 85 deletions
diff --git a/libs/libcurl/src/mqtt.c b/libs/libcurl/src/mqtt.c index 05fd45feb9..568ca4ceab 100644 --- a/libs/libcurl/src/mqtt.c +++ b/libs/libcurl/src/mqtt.c @@ -37,7 +37,7 @@ #include "strdup.h"
#include "url.h"
#include "escape.h"
-#include "warnless.h"
+#include "curlx/warnless.h"
#include "curl_printf.h"
#include "curl_memory.h"
#include "multiif.h"
@@ -46,17 +46,59 @@ /* The last #include file should be: */
#include "memdebug.h"
-#define MQTT_MSG_CONNECT 0x10
-#define MQTT_MSG_CONNACK 0x20
-#define MQTT_MSG_PUBLISH 0x30
-#define MQTT_MSG_SUBSCRIBE 0x82
-#define MQTT_MSG_SUBACK 0x90
+/* first byte is command.
+ second byte is for flags. */
+#define MQTT_MSG_CONNECT 0x10
+/* #define MQTT_MSG_CONNACK 0x20 */
+#define MQTT_MSG_PUBLISH 0x30
+#define MQTT_MSG_SUBSCRIBE 0x82
+#define MQTT_MSG_SUBACK 0x90
#define MQTT_MSG_DISCONNECT 0xe0
+#define MQTT_MSG_PINGREQ 0xC0
+#define MQTT_MSG_PINGRESP 0xD0
#define MQTT_CONNACK_LEN 2
#define MQTT_SUBACK_LEN 3
#define MQTT_CLIENTID_LEN 12 /* "curl0123abcd" */
+/* meta key for storing protocol meta at easy handle */
+#define CURL_META_MQTT_EASY "meta:proto:mqtt:easy"
+/* meta key for storing protocol meta at connection */
+#define CURL_META_MQTT_CONN "meta:proto:mqtt:conn"
+
+enum mqttstate {
+ MQTT_FIRST, /* 0 */
+ MQTT_REMAINING_LENGTH, /* 1 */
+ MQTT_CONNACK, /* 2 */
+ MQTT_SUBACK, /* 3 */
+ MQTT_SUBACK_COMING, /* 4 - the SUBACK remainder */
+ MQTT_PUBWAIT, /* 5 - wait for publish */
+ MQTT_PUB_REMAIN, /* 6 - wait for the remainder of the publish */
+
+ MQTT_NOSTATE /* 7 - never used an actual state */
+};
+
+struct mqtt_conn {
+ enum mqttstate state;
+ enum mqttstate nextstate; /* switch to this after remaining length is
+ done */
+ unsigned int packetid;
+};
+
+/* protocol-specific transfer-related data */
+struct MQTT {
+ struct dynbuf sendbuf;
+ /* when receiving */
+ struct dynbuf recvbuf;
+ size_t npacket; /* byte counter */
+ size_t remaining_length;
+ unsigned char pkt_hd[4]; /* for decoding the arriving packet length */
+ struct curltime lastTime; /* last time we sent or received data */
+ unsigned char firstbyte;
+ BIT(pingsent); /* 1 while we wait for ping response */
+};
+
+
/*
* Forward declarations.
*/
@@ -92,51 +134,79 @@ const struct Curl_handler Curl_handler_mqtt = { ZERO_NULL, /* write_resp_hd */
ZERO_NULL, /* connection_check */
ZERO_NULL, /* attach connection */
+ ZERO_NULL, /* follow */
PORT_MQTT, /* defport */
CURLPROTO_MQTT, /* protocol */
CURLPROTO_MQTT, /* family */
PROTOPT_NONE /* flags */
};
+static void mqtt_easy_dtor(void *key, size_t klen, void *entry)
+{
+ struct MQTT *mq = entry;
+ (void)key;
+ (void)klen;
+ curlx_dyn_free(&mq->sendbuf);
+ curlx_dyn_free(&mq->recvbuf);
+ free(mq);
+}
+
+static void mqtt_conn_dtor(void *key, size_t klen, void *entry)
+{
+ (void)key;
+ (void)klen;
+ free(entry);
+}
+
static CURLcode mqtt_setup_conn(struct Curl_easy *data,
struct connectdata *conn)
{
- /* allocate the HTTP-specific struct for the Curl_easy, only to survive
- during this request */
+ /* setup MQTT specific meta data at easy handle and connection */
+ struct mqtt_conn *mqtt;
struct MQTT *mq;
- (void)conn;
- DEBUGASSERT(data->req.p.mqtt == NULL);
+
+ mqtt = calloc(1, sizeof(*mqtt));
+ if(!mqtt ||
+ Curl_conn_meta_set(conn, CURL_META_MQTT_CONN, mqtt, mqtt_conn_dtor))
+ return CURLE_OUT_OF_MEMORY;
mq = calloc(1, sizeof(struct MQTT));
if(!mq)
return CURLE_OUT_OF_MEMORY;
- Curl_dyn_init(&mq->recvbuf, DYN_MQTT_RECV);
- data->req.p.mqtt = mq;
+ curlx_dyn_init(&mq->recvbuf, DYN_MQTT_RECV);
+ curlx_dyn_init(&mq->sendbuf, DYN_MQTT_SEND);
+ if(Curl_meta_set(data, CURL_META_MQTT_EASY, mq, mqtt_easy_dtor))
+ return CURLE_OUT_OF_MEMORY;
return CURLE_OK;
}
static CURLcode mqtt_send(struct Curl_easy *data,
- char *buf, size_t len)
+ const char *buf, size_t len)
{
- CURLcode result = CURLE_OK;
- struct MQTT *mq = data->req.p.mqtt;
size_t n;
+ CURLcode result;
+ struct MQTT *mq = Curl_meta_get(data, CURL_META_MQTT_EASY);
+
+ if(!mq)
+ return CURLE_FAILED_INIT;
+
result = Curl_xfer_send(data, buf, len, FALSE, &n);
if(result)
return result;
+ mq->lastTime = curlx_now();
Curl_debug(data, CURLINFO_HEADER_OUT, buf, (size_t)n);
if(len != n) {
size_t nsend = len - n;
- char *sendleftovers = Curl_memdup(&buf[n], nsend);
- if(!sendleftovers)
- return CURLE_OUT_OF_MEMORY;
- mq->sendleftovers = sendleftovers;
- mq->nsend = nsend;
- }
- else {
- mq->sendleftovers = NULL;
- mq->nsend = 0;
+ if(curlx_dyn_len(&mq->sendbuf)) {
+ DEBUGASSERT(curlx_dyn_len(&mq->sendbuf) >= nsend);
+ result = curlx_dyn_tail(&mq->sendbuf, nsend); /* keep this much */
+ }
+ else {
+ result = curlx_dyn_addn(&mq->sendbuf, &buf[n], nsend);
+ }
}
+ else
+ curlx_dyn_reset(&mq->sendbuf);
return result;
}
@@ -256,7 +326,6 @@ static CURLcode mqtt_connect(struct Curl_easy *data) int remain_pos = 0;
char remain[4] = {0};
size_t packetlen = 0;
- size_t payloadlen = 0;
size_t start_user = 0;
size_t start_pwd = 0;
char client_id[MQTT_CLIENTID_LEN + 1] = "curl";
@@ -271,14 +340,11 @@ static CURLcode mqtt_connect(struct Curl_easy *data) const char *passwd = data->state.aptr.passwd ?
data->state.aptr.passwd : "";
const size_t plen = strlen(passwd);
-
- payloadlen = ulen + plen + MQTT_CLIENTID_LEN + 2;
- /* The plus 2 are for the MSB and LSB describing the length of the string to
- * be added on the payload. Refer to spec 1.5.2 and 1.5.4 */
- if(ulen)
- payloadlen += 2;
- if(plen)
- payloadlen += 2;
+ const size_t payloadlen = ulen + plen + MQTT_CLIENTID_LEN + 2 +
+ /* The plus 2s below are for the MSB and LSB describing the length of the
+ string to be added on the payload. Refer to spec 1.5.2 and 1.5.4 */
+ (ulen ? 2 : 0) +
+ (plen ? 2 : 0);
/* getting how much occupy the remain length */
remain_pos = mqtt_encode_len(remain, payloadlen + 10);
@@ -287,12 +353,11 @@ static CURLcode mqtt_connect(struct Curl_easy *data) packetlen = payloadlen + 10 + remain_pos + 1;
/* allocating packet */
- if(packetlen > 268435455)
+ if(packetlen > 0xFFFFFFF)
return CURLE_WEIRD_SERVER_REPLY;
- packet = malloc(packetlen);
+ packet = calloc(1, packetlen);
if(!packet)
return CURLE_OUT_OF_MEMORY;
- memset(packet, 0, packetlen);
/* set initial values for the CONNECT packet */
pos = init_connpack(packet, remain, remain_pos);
@@ -308,9 +373,9 @@ static CURLcode mqtt_connect(struct Curl_easy *data) }
infof(data, "Using client id '%s'", client_id);
- /* position where starts the user payload */
+ /* position where the user payload starts */
start_user = pos + 3 + MQTT_CLIENTID_LEN;
- /* position where starts the password payload */
+ /* position where the password payload starts */
start_pwd = start_user + ulen;
/* if username was provided, add it to the packet */
if(ulen) {
@@ -319,7 +384,7 @@ static CURLcode mqtt_connect(struct Curl_easy *data) rc = add_user(username, ulen,
(unsigned char *)packet, start_user, remain_pos);
if(rc) {
- failf(data, "Username is too large: [%zu]", ulen);
+ failf(data, "Username too long: [%zu]", ulen);
result = CURLE_WEIRD_SERVER_REPLY;
goto end;
}
@@ -329,7 +394,7 @@ static CURLcode mqtt_connect(struct Curl_easy *data) if(plen) {
rc = add_passwd(passwd, plen, packet, start_pwd, remain_pos);
if(rc) {
- failf(data, "Password is too large: [%zu]", plen);
+ failf(data, "Password too long: [%zu]", plen);
result = CURLE_WEIRD_SERVER_REPLY;
goto end;
}
@@ -348,20 +413,19 @@ end: static CURLcode mqtt_disconnect(struct Curl_easy *data)
{
- CURLcode result = CURLE_OK;
- struct MQTT *mq = data->req.p.mqtt;
- result = mqtt_send(data, (char *)"\xe0\x00", 2);
- Curl_safefree(mq->sendleftovers);
- Curl_dyn_free(&mq->recvbuf);
- return result;
+ return mqtt_send(data, "\xe0\x00", 2);
}
static CURLcode mqtt_recv_atleast(struct Curl_easy *data, size_t nbytes)
{
- struct MQTT *mq = data->req.p.mqtt;
- size_t rlen = Curl_dyn_len(&mq->recvbuf);
+ struct MQTT *mq = Curl_meta_get(data, CURL_META_MQTT_EASY);
+ size_t rlen;
CURLcode result;
+ if(!mq)
+ return CURLE_FAILED_INIT;
+ rlen = curlx_dyn_len(&mq->recvbuf);
+
if(rlen < nbytes) {
unsigned char readbuf[1024];
ssize_t nread;
@@ -371,42 +435,49 @@ static CURLcode mqtt_recv_atleast(struct Curl_easy *data, size_t nbytes) if(result)
return result;
DEBUGASSERT(nread >= 0);
- if(Curl_dyn_addn(&mq->recvbuf, readbuf, (size_t)nread))
+ if(curlx_dyn_addn(&mq->recvbuf, readbuf, (size_t)nread))
return CURLE_OUT_OF_MEMORY;
- rlen = Curl_dyn_len(&mq->recvbuf);
+ rlen = curlx_dyn_len(&mq->recvbuf);
}
return (rlen >= nbytes) ? CURLE_OK : CURLE_AGAIN;
}
static void mqtt_recv_consume(struct Curl_easy *data, size_t nbytes)
{
- struct MQTT *mq = data->req.p.mqtt;
- size_t rlen = Curl_dyn_len(&mq->recvbuf);
- if(rlen <= nbytes)
- Curl_dyn_reset(&mq->recvbuf);
- else
- Curl_dyn_tail(&mq->recvbuf, rlen - nbytes);
+ struct MQTT *mq = Curl_meta_get(data, CURL_META_MQTT_EASY);
+ DEBUGASSERT(mq);
+ if(mq) {
+ size_t rlen = curlx_dyn_len(&mq->recvbuf);
+ if(rlen <= nbytes)
+ curlx_dyn_reset(&mq->recvbuf);
+ else
+ curlx_dyn_tail(&mq->recvbuf, rlen - nbytes);
+ }
}
static CURLcode mqtt_verify_connack(struct Curl_easy *data)
{
- struct MQTT *mq = data->req.p.mqtt;
+ struct MQTT *mq = Curl_meta_get(data, CURL_META_MQTT_EASY);
CURLcode result;
char *ptr;
+ DEBUGASSERT(mq);
+ if(!mq)
+ return CURLE_FAILED_INIT;
+
result = mqtt_recv_atleast(data, MQTT_CONNACK_LEN);
if(result)
goto fail;
/* verify CONNACK */
- DEBUGASSERT(Curl_dyn_len(&mq->recvbuf) >= MQTT_CONNACK_LEN);
- ptr = Curl_dyn_ptr(&mq->recvbuf);
+ DEBUGASSERT(curlx_dyn_len(&mq->recvbuf) >= MQTT_CONNACK_LEN);
+ ptr = curlx_dyn_ptr(&mq->recvbuf);
Curl_debug(data, CURLINFO_HEADER_IN, ptr, MQTT_CONNACK_LEN);
if(ptr[0] != 0x00 || ptr[1] != 0x00) {
failf(data, "Expected %02x%02x but got %02x%02x",
0x00, 0x00, ptr[0], ptr[1]);
- Curl_dyn_reset(&mq->recvbuf);
+ curlx_dyn_reset(&mq->recvbuf);
result = CURLE_WEIRD_SERVER_REPLY;
goto fail;
}
@@ -443,12 +514,16 @@ static CURLcode mqtt_subscribe(struct Curl_easy *data) char encodedsize[4];
size_t n;
struct connectdata *conn = data->conn;
+ struct mqtt_conn *mqtt = Curl_conn_meta_get(conn, CURL_META_MQTT_CONN);
+
+ if(!mqtt)
+ return CURLE_FAILED_INIT;
result = mqtt_get_topic(data, &topic, &topiclen);
if(result)
goto fail;
- conn->proto.mqtt.packetid++;
+ mqtt->packetid++;
packetlen = topiclen + 5; /* packetid + topic (has a two byte length field)
+ 2 bytes topic length + QoS byte */
@@ -463,14 +538,14 @@ static CURLcode mqtt_subscribe(struct Curl_easy *data) packet[0] = MQTT_MSG_SUBSCRIBE;
memcpy(&packet[1], encodedsize, n);
- packet[1 + n] = (conn->proto.mqtt.packetid >> 8) & 0xff;
- packet[2 + n] = conn->proto.mqtt.packetid & 0xff;
+ packet[1 + n] = (mqtt->packetid >> 8) & 0xff;
+ packet[2 + n] = mqtt->packetid & 0xff;
packet[3 + n] = (topiclen >> 8) & 0xff;
packet[4 + n ] = topiclen & 0xff;
memcpy(&packet[5 + n], topic, topiclen);
packet[5 + n + topiclen] = 0; /* QoS zero */
- result = mqtt_send(data, (char *)packet, packetlen);
+ result = mqtt_send(data, (const char *)packet, packetlen);
fail:
free(topic);
@@ -483,25 +558,28 @@ fail: */
static CURLcode mqtt_verify_suback(struct Curl_easy *data)
{
- struct MQTT *mq = data->req.p.mqtt;
+ struct MQTT *mq = Curl_meta_get(data, CURL_META_MQTT_EASY);
struct connectdata *conn = data->conn;
- struct mqtt_conn *mqtt = &conn->proto.mqtt;
+ struct mqtt_conn *mqtt = Curl_conn_meta_get(conn, CURL_META_MQTT_CONN);
CURLcode result;
char *ptr;
+ if(!mqtt || !mq)
+ return CURLE_FAILED_INIT;
+
result = mqtt_recv_atleast(data, MQTT_SUBACK_LEN);
if(result)
goto fail;
/* verify SUBACK */
- DEBUGASSERT(Curl_dyn_len(&mq->recvbuf) >= MQTT_SUBACK_LEN);
- ptr = Curl_dyn_ptr(&mq->recvbuf);
+ DEBUGASSERT(curlx_dyn_len(&mq->recvbuf) >= MQTT_SUBACK_LEN);
+ ptr = curlx_dyn_ptr(&mq->recvbuf);
Curl_debug(data, CURLINFO_HEADER_IN, ptr, MQTT_SUBACK_LEN);
if(((unsigned char)ptr[0]) != ((mqtt->packetid >> 8) & 0xff) ||
((unsigned char)ptr[1]) != (mqtt->packetid & 0xff) ||
ptr[2] != 0x00) {
- Curl_dyn_reset(&mq->recvbuf);
+ curlx_dyn_reset(&mq->recvbuf);
result = CURLE_WEIRD_SERVER_REPLY;
goto fail;
}
@@ -557,7 +635,7 @@ static CURLcode mqtt_publish(struct Curl_easy *data) i += topiclen;
memcpy(&pkt[i], payload, payloadlen);
i += payloadlen;
- result = mqtt_send(data, (char *)pkt, i);
+ result = mqtt_send(data, (const char *)pkt, i);
fail:
free(pkt);
@@ -605,7 +683,10 @@ static void mqstate(struct Curl_easy *data, enum mqttstate nextstate) /* used if state == FIRST */
{
struct connectdata *conn = data->conn;
- struct mqtt_conn *mqtt = &conn->proto.mqtt;
+ struct mqtt_conn *mqtt = Curl_conn_meta_get(conn, CURL_META_MQTT_CONN);
+ DEBUGASSERT(mqtt);
+ if(!mqtt)
+ return;
#ifdef DEBUGBUILD
infof(data, "%s (from %s) (next is %s)",
statenames[state],
@@ -624,10 +705,14 @@ static CURLcode mqtt_read_publish(struct Curl_easy *data, bool *done) struct connectdata *conn = data->conn;
ssize_t nread;
size_t remlen;
- struct mqtt_conn *mqtt = &conn->proto.mqtt;
- struct MQTT *mq = data->req.p.mqtt;
+ struct mqtt_conn *mqtt = Curl_conn_meta_get(conn, CURL_META_MQTT_CONN);
+ struct MQTT *mq = Curl_meta_get(data, CURL_META_MQTT_EASY);
unsigned char packet;
+ DEBUGASSERT(mqtt);
+ if(!mqtt || !mq)
+ return CURLE_FAILED_INIT;
+
switch(mqtt->state) {
MQTT_SUBACK_COMING:
case MQTT_SUBACK_COMING:
@@ -691,6 +776,9 @@ MQTT_SUBACK_COMING: goto end;
}
+ /* we received something */
+ mq->lastTime = curlx_now();
+
/* if QoS is set, message contains packet id */
result = Curl_client_write(data, CLIENTWRITE_BODY, buffer, nread);
if(result)
@@ -713,9 +801,15 @@ end: static CURLcode mqtt_do(struct Curl_easy *data, bool *done)
{
+ struct MQTT *mq = Curl_meta_get(data, CURL_META_MQTT_EASY);
CURLcode result = CURLE_OK;
*done = FALSE; /* unconditionally */
+ if(!mq)
+ return CURLE_FAILED_INIT;
+ mq->lastTime = curlx_now();
+ mq->pingsent = FALSE;
+
result = mqtt_connect(data);
if(result) {
failf(data, "Error %d sending MQTT CONNECT request", result);
@@ -728,34 +822,73 @@ static CURLcode mqtt_do(struct Curl_easy *data, bool *done) static CURLcode mqtt_done(struct Curl_easy *data,
CURLcode status, bool premature)
{
- struct MQTT *mq = data->req.p.mqtt;
+ struct MQTT *mq = Curl_meta_get(data, CURL_META_MQTT_EASY);
(void)status;
(void)premature;
- Curl_safefree(mq->sendleftovers);
- Curl_dyn_free(&mq->recvbuf);
+ if(mq) {
+ curlx_dyn_free(&mq->sendbuf);
+ curlx_dyn_free(&mq->recvbuf);
+ }
return CURLE_OK;
}
-static CURLcode mqtt_doing(struct Curl_easy *data, bool *done)
+/* we ping regularly to avoid being disconnected by the server */
+static CURLcode mqtt_ping(struct Curl_easy *data)
{
+ struct MQTT *mq = Curl_meta_get(data, CURL_META_MQTT_EASY);
CURLcode result = CURLE_OK;
struct connectdata *conn = data->conn;
- struct mqtt_conn *mqtt = &conn->proto.mqtt;
- struct MQTT *mq = data->req.p.mqtt;
+ struct mqtt_conn *mqtt = Curl_conn_meta_get(conn, CURL_META_MQTT_CONN);
+
+ if(!mqtt || !mq)
+ return CURLE_FAILED_INIT;
+
+ if(mqtt->state == MQTT_FIRST &&
+ !mq->pingsent &&
+ data->set.upkeep_interval_ms > 0) {
+ struct curltime t = curlx_now();
+ timediff_t diff = curlx_timediff(t, mq->lastTime);
+
+ if(diff > data->set.upkeep_interval_ms) {
+ /* 0xC0 is PINGREQ, and 0x00 is remaining length */
+ unsigned char packet[2] = { 0xC0, 0x00 };
+ size_t packetlen = sizeof(packet);
+
+ result = mqtt_send(data, (char *)packet, packetlen);
+ if(!result) {
+ mq->pingsent = TRUE;
+ }
+ infof(data, "mqtt_ping: sent ping request.");
+ }
+ }
+ return result;
+}
+
+static CURLcode mqtt_doing(struct Curl_easy *data, bool *done)
+{
+ struct MQTT *mq = Curl_meta_get(data, CURL_META_MQTT_EASY);
+ CURLcode result = CURLE_OK;
ssize_t nread;
unsigned char recvbyte;
+ struct mqtt_conn *mqtt = Curl_conn_meta_get(data->conn, CURL_META_MQTT_CONN);
+
+ if(!mqtt || !mq)
+ return CURLE_FAILED_INIT;
*done = FALSE;
- if(mq->nsend) {
+ if(curlx_dyn_len(&mq->sendbuf)) {
/* send the remainder of an outgoing packet */
- char *ptr = mq->sendleftovers;
- result = mqtt_send(data, mq->sendleftovers, mq->nsend);
- free(ptr);
+ result = mqtt_send(data, curlx_dyn_ptr(&mq->sendbuf),
+ curlx_dyn_len(&mq->sendbuf));
if(result)
return result;
}
+ result = mqtt_ping(data);
+ if(result)
+ return result;
+
infof(data, "mqtt_doing: state [%d]", (int) mqtt->state);
switch(mqtt->state) {
case MQTT_FIRST:
@@ -769,7 +902,11 @@ static CURLcode mqtt_doing(struct Curl_easy *data, bool *done) result = CURLE_RECV_ERROR;
break;
}
- Curl_debug(data, CURLINFO_HEADER_IN, (char *)&mq->firstbyte, 1);
+ Curl_debug(data, CURLINFO_HEADER_IN, (const char *)&mq->firstbyte, 1);
+
+ /* we received something */
+ mq->lastTime = curlx_now();
+
/* remember the first byte */
mq->npacket = 0;
mqstate(data, MQTT_REMAINING_LENGTH, MQTT_NOSTATE);
@@ -779,7 +916,7 @@ static CURLcode mqtt_doing(struct Curl_easy *data, bool *done) result = Curl_xfer_recv(data, (char *)&recvbyte, 1, &nread);
if(result || !nread)
break;
- Curl_debug(data, CURLINFO_HEADER_IN, (char *)&recvbyte, 1);
+ Curl_debug(data, CURLINFO_HEADER_IN, (const char *)&recvbyte, 1);
mq->pkt_hd[mq->npacket++] = recvbyte;
} while((recvbyte & 0x80) && (mq->npacket < 4));
if(!result && nread && (recvbyte & 0x80))
@@ -800,6 +937,13 @@ static CURLcode mqtt_doing(struct Curl_easy *data, bool *done) infof(data, "Got DISCONNECT");
*done = TRUE;
}
+
+ /* ping response */
+ if(mq->firstbyte == MQTT_MSG_PINGRESP) {
+ infof(data, "Received ping response.");
+ mq->pingsent = FALSE;
+ mqstate(data, MQTT_FIRST, MQTT_PUBWAIT);
+ }
break;
case MQTT_CONNACK:
result = mqtt_verify_connack(data);
|