3  * MQTT client
5  * @defgroup mqtt MQTT client
80  * MQTT client connection states
166  * @param client MQTT client
170 msg_generate_packet_id(mqtt_client_t *client)  in msg_generate_packet_id()  argument
172   client->pkt_id_seq++;  in msg_generate_packet_id()
173   if (client->pkt_id_seq == 0) {  in msg_generate_packet_id()
174     client->pkt_id_seq++;  in msg_generate_packet_id()
176   return client->pkt_id_seq;  in msg_generate_packet_id()
533  * @param client MQTT client
537 mqtt_close(mqtt_client_t *client, mqtt_connection_status_t reason)  in mqtt_close()  argument
539   LWIP_ASSERT("mqtt_close: client != NULL", client != NULL);  in mqtt_close()
542   if (client->conn != NULL) {  in mqtt_close()
544     altcp_recv(client->conn, NULL);  in mqtt_close()
545     altcp_err(client->conn,  NULL);  in mqtt_close()
546     altcp_sent(client->conn, NULL);  in mqtt_close()
547     res = altcp_close(client->conn);  in mqtt_close()
549       altcp_abort(client->conn);  in mqtt_close()
552     client->conn = NULL;  in mqtt_close()
556   mqtt_clear_requests(&client->pend_req_queue);  in mqtt_close()
558   sys_untimeout(mqtt_cyclic_timer, client);  in mqtt_close()
561   if (client->conn_state != TCP_DISCONNECTED) {  in mqtt_close()
563     client->conn_state = TCP_DISCONNECTED;  in mqtt_close()
564     if (client->connect_cb != NULL) {  in mqtt_close()
565       client->connect_cb(client, client->connect_arg, reason);  in mqtt_close()
573  * @param arg MQTT client
579   mqtt_client_t *client = (mqtt_client_t *)arg;  in mqtt_cyclic_timer()  local
580   LWIP_ASSERT("mqtt_cyclic_timer: client != NULL", client != NULL);  in mqtt_cyclic_timer()
582   if (client->conn_state == MQTT_CONNECTING) {  in mqtt_cyclic_timer()
583     client->cyclic_tick++;  in mqtt_cyclic_timer()
584     if ((client->cyclic_tick * MQTT_CYCLIC_TIMER_INTERVAL) >= MQTT_CONNECT_TIMOUT) {  in mqtt_cyclic_timer()
587       mqtt_close(client, MQTT_CONNECT_TIMEOUT);  in mqtt_cyclic_timer()
590   } else if (client->conn_state == MQTT_CONNECTED) {  in mqtt_cyclic_timer()
592     mqtt_request_time_elapsed(&client->pend_req_queue, MQTT_CYCLIC_TIMER_INTERVAL);  in mqtt_cyclic_timer()
595     if (client->keep_alive > 0) {  in mqtt_cyclic_timer()
597       client->server_watchdog++;  in mqtt_cyclic_timer()
599 …if ((client->server_watchdog * MQTT_CYCLIC_TIMER_INTERVAL) > (client->keep_alive + client->keep_al…  in mqtt_cyclic_timer()
601         mqtt_close(client, MQTT_CONNECT_TIMEOUT);  in mqtt_cyclic_timer()
606       if ((client->cyclic_tick * MQTT_CYCLIC_TIMER_INTERVAL) >= client->keep_alive) {  in mqtt_cyclic_timer()
608         if (mqtt_output_check_space(&client->output, 0) != 0) {  in mqtt_cyclic_timer()
609           mqtt_output_append_fixed_header(&client->output, MQTT_MSG_TYPE_PINGREQ, 0, 0, 0, 0);  in mqtt_cyclic_timer()
610           client->cyclic_tick = 0;  in mqtt_cyclic_timer()
613         client->cyclic_tick++;  in mqtt_cyclic_timer()
617 …_DEBUG_WARN, ("mqtt_cyclic_timer: Timer should not be running in state %d\n", client->conn_state));  in mqtt_cyclic_timer()
628  * @param client MQTT client
635 pub_ack_rec_rel_response(mqtt_client_t *client, u8_t msg, u16_t pkt_id, u8_t qos)  in pub_ack_rec_rel_response()  argument
638   if (mqtt_output_check_space(&client->output, 2)) {  in pub_ack_rec_rel_response()
639     mqtt_output_append_fixed_header(&client->output, msg, 0, qos, 0, 2);  in pub_ack_rec_rel_response()
640     mqtt_output_append_u16(&client->output, pkt_id);  in pub_ack_rec_rel_response()
641     mqtt_output_send(&client->output, client->conn);  in pub_ack_rec_rel_response()
666  * @param client MQTT client
672 mqtt_message_received(mqtt_client_t *client, u8_t fixed_hdr_idx, u16_t length, u32_t remaining_leng…  in mqtt_message_received()  argument
676   u8_t *var_hdr_payload = client->rx_buffer + fixed_hdr_idx;  in mqtt_message_received()
677   size_t var_hdr_payload_bufsize = sizeof(client->rx_buffer) - fixed_hdr_idx;  in mqtt_message_received()
680   u8_t pkt_type = MQTT_CTL_PACKET_TYPE(client->rx_buffer[0]);  in mqtt_message_received()
683 …LWIP_ASSERT("client->msg_idx < MQTT_VAR_HEADER_BUFFER_LEN", client->msg_idx < MQTT_VAR_HEADER_BUFF…  in mqtt_message_received()
684   LWIP_ASSERT("fixed_hdr_idx <= client->msg_idx", fixed_hdr_idx <= client->msg_idx);  in mqtt_message_received()
689     if (client->conn_state == MQTT_CONNECTING) {  in mqtt_message_received()
699         client->cyclic_tick = 0;  in mqtt_message_received()
700         client->conn_state = MQTT_CONNECTED;  in mqtt_message_received()
702         if (client->connect_cb != 0) {  in mqtt_message_received()
703           client->connect_cb(client, client->connect_arg, res);  in mqtt_message_received()
715     u8_t qos = MQTT_CTL_PACKET_QOS(client->rx_buffer[0]);  in mqtt_message_received()
717     if (client->msg_idx <= MQTT_VAR_HEADER_BUFFER_LEN) {  in mqtt_message_received()
750 …client->inpub_pkt_id = ((u16_t)var_hdr_payload[after_topic] << 8) + (u16_t)var_hdr_payload[after_t…  in mqtt_message_received()
753         client->inpub_pkt_id = 0;  in mqtt_message_received()
765       if (client->pub_cb != NULL) {  in mqtt_message_received()
766         client->pub_cb(client->inpub_arg, (const char *)topic, remaining_length + payload_length);  in mqtt_message_received()
776 …client->data_cb(client->inpub_arg, var_hdr_payload + payload_offset, payload_length, remaining_len…  in mqtt_message_received()
782                                        mqtt_msg_type_to_str(resp_msg), client->inpub_pkt_id));  in mqtt_message_received()
783         pub_ack_rec_rel_response(client, resp_msg, client->inpub_pkt_id, 0);  in mqtt_message_received()
796       pub_ack_rec_rel_response(client, MQTT_MSG_TYPE_PUBREL, pkt_id, 1);  in mqtt_message_received()
800       pub_ack_rec_rel_response(client, MQTT_MSG_TYPE_PUBCOMP, pkt_id, 0);  in mqtt_message_received()
804       struct mqtt_request_t *r = mqtt_take_request(&client->pend_req_queue, pkt_id);  in mqtt_message_received()
834  * @param client MQTT client
839 mqtt_parse_incoming(mqtt_client_t *client, struct pbuf *p)  in mqtt_parse_incoming()  argument
849        client->rx_buffer. client->msg_idx keeps track of this. */  in mqtt_parse_incoming()
852       if (fixed_hdr_idx < client->msg_idx) {  in mqtt_parse_incoming()
853         /* parse header from old pbuf (buffered in client->rx_buffer) */  in mqtt_parse_incoming()
854         b = client->rx_buffer[fixed_hdr_idx];  in mqtt_parse_incoming()
856         /* parse header from this pbuf and save it in client->rx_buffer in case  in mqtt_parse_incoming()
859         client->rx_buffer[client->msg_idx++] = b;  in mqtt_parse_incoming()
873             mqtt_message_received(client, fixed_hdr_idx, 0, 0);  in mqtt_parse_incoming()
874             client->msg_idx = 0;  in mqtt_parse_incoming()
879             msg_rem_len = (msg_rem_len + fixed_hdr_idx) - client->msg_idx;  in mqtt_parse_incoming()
887 …cpy_start = (client->msg_idx - fixed_hdr_idx) % (MQTT_VAR_HEADER_BUFFER_LEN - fixed_hdr_idx) + fix…  in mqtt_parse_incoming()
897       pbuf_copy_partial(p, client->rx_buffer + cpy_start, cpy_len, in_offset);  in mqtt_parse_incoming()
900       client->msg_idx += cpy_len;  in mqtt_parse_incoming()
904 …e_incoming: msg_idx: %"U32_F", cpy_len: %"U16_F", remaining %"U32_F"\n", client->msg_idx, cpy_len,…  in mqtt_parse_incoming()
907 …mqtt_connection_status_t res = mqtt_message_received(client, fixed_hdr_idx, (cpy_start + cpy_len) …  in mqtt_parse_incoming()
913           client->msg_idx = 0;  in mqtt_parse_incoming()
926  * @param arg MQTT client
934   mqtt_client_t *client = (mqtt_client_t *)arg;  in mqtt_tcp_recv_cb()  local
935   LWIP_ASSERT("mqtt_tcp_recv_cb: client != NULL", client != NULL);  in mqtt_tcp_recv_cb()
936   LWIP_ASSERT("mqtt_tcp_recv_cb: client->conn == pcb", client->conn == pcb);  in mqtt_tcp_recv_cb()
940     mqtt_close(client, MQTT_CONNECT_DISCONNECTED);  in mqtt_tcp_recv_cb()
951     res = mqtt_parse_incoming(client, p);  in mqtt_tcp_recv_cb()
955       mqtt_close(client, res);  in mqtt_tcp_recv_cb()
958     if (client->keep_alive != 0) {  in mqtt_tcp_recv_cb()
960       client->server_watchdog = 0;  in mqtt_tcp_recv_cb()
970  * @param arg MQTT client
978   mqtt_client_t *client = (mqtt_client_t *)arg;  in mqtt_tcp_sent_cb()  local
983   if (client->conn_state == MQTT_CONNECTED) {  in mqtt_tcp_sent_cb()
987     client->cyclic_tick = 0;  in mqtt_tcp_sent_cb()
988     client->server_watchdog = 0;  in mqtt_tcp_sent_cb()
990     while ((r = mqtt_take_request(&client->pend_req_queue, 0)) != NULL) {  in mqtt_tcp_sent_cb()
998     mqtt_output_send(&client->output, client->conn);  in mqtt_tcp_sent_cb()
1005  * @param arg MQTT client
1011   mqtt_client_t *client = (mqtt_client_t *)arg;  in mqtt_tcp_err_cb()  local
1014   LWIP_ASSERT("mqtt_tcp_err_cb: client != NULL", client != NULL);  in mqtt_tcp_err_cb()
1016   client->conn = 0;  in mqtt_tcp_err_cb()
1017   mqtt_close(client, MQTT_CONNECT_DISCONNECTED);  in mqtt_tcp_err_cb()
1022  * @param arg MQTT client
1029   mqtt_client_t *client = (mqtt_client_t *)arg;  in mqtt_tcp_poll_cb()  local
1030   if (client->conn_state == MQTT_CONNECTED) {  in mqtt_tcp_poll_cb()
1032     mqtt_output_send(&client->output, tpcb);  in mqtt_tcp_poll_cb()
1039  * @param arg MQTT client
1046   mqtt_client_t *client = (mqtt_client_t *)arg;  in mqtt_tcp_connect_cb()  local
1054   client->msg_idx = 0;  in mqtt_tcp_connect_cb()
1063   client->conn_state = MQTT_CONNECTING;  in mqtt_tcp_connect_cb()
1066   sys_timeout(MQTT_CYCLIC_TIMER_INTERVAL * 1000, mqtt_cyclic_timer, client);  in mqtt_tcp_connect_cb()
1067   client->cyclic_tick = 0;  in mqtt_tcp_connect_cb()
1070   mqtt_output_send(&client->output, client->conn);  in mqtt_tcp_connect_cb()
1084  * @param client MQTT client
1093  *         ERR_CONN if client is disconnected
1097 mqtt_publish(mqtt_client_t *client, const char *topic, const void *payload, u16_t payload_length, u…  in mqtt_publish()  argument
1108   LWIP_ASSERT("mqtt_publish: client != NULL", client);  in mqtt_publish()
1110 …LWIP_ERROR("mqtt_publish: TCP disconnected", (client->conn_state != TCP_DISCONNECTED), return ERR_…  in mqtt_publish()
1120     pkt_id = msg_generate_packet_id(client);  in mqtt_publish()
1130   r = mqtt_create_request(client->req_list, LWIP_ARRAYSIZE(client->req_list), pkt_id, cb, arg);  in mqtt_publish()
1135   if (mqtt_output_check_space(&client->output, remaining_length) == 0) {  in mqtt_publish()
1140 …mqtt_output_append_fixed_header(&client->output, MQTT_MSG_TYPE_PUBLISH, 0, qos, retain, remaining_…  in mqtt_publish()
1143   mqtt_output_append_string(&client->output, topic, topic_len);  in mqtt_publish()
1147     mqtt_output_append_u16(&client->output, pkt_id);  in mqtt_publish()
1152     mqtt_output_append_buf(&client->output, payload, payload_length);  in mqtt_publish()
1155   mqtt_append_request(&client->pend_req_queue, r);  in mqtt_publish()
1156   mqtt_output_send(&client->output, client->conn);  in mqtt_publish()
1164  * @param client MQTT client
1173 mqtt_sub_unsub(mqtt_client_t *client, const char *topic, u8_t qos, mqtt_request_cb_t cb, void *arg,…  in mqtt_sub_unsub()  argument
1183   LWIP_ASSERT("mqtt_sub_unsub: client != NULL", client);  in mqtt_sub_unsub()
1195   if (client->conn_state == TCP_DISCONNECTED) {  in mqtt_sub_unsub()
1200   pkt_id = msg_generate_packet_id(client);  in mqtt_sub_unsub()
1201   r = mqtt_create_request(client->req_list, LWIP_ARRAYSIZE(client->req_list), pkt_id, cb, arg);  in mqtt_sub_unsub()
1206   if (mqtt_output_check_space(&client->output, remaining_length) == 0) {  in mqtt_sub_unsub()
1211 …LWIP_DEBUGF(MQTT_DEBUG_TRACE, ("mqtt_sub_unsub: Client (un)subscribe to topic \"%s\", id: %d\n", t…  in mqtt_sub_unsub()
1213 …mqtt_output_append_fixed_header(&client->output, sub ? MQTT_MSG_TYPE_SUBSCRIBE : MQTT_MSG_TYPE_UNS…  in mqtt_sub_unsub()
1215   mqtt_output_append_u16(&client->output, pkt_id);  in mqtt_sub_unsub()
1217   mqtt_output_append_string(&client->output, topic, topic_len);  in mqtt_sub_unsub()
1220     mqtt_output_append_u8(&client->output, LWIP_MIN(qos, 2));  in mqtt_sub_unsub()
1223   mqtt_append_request(&client->pend_req_queue, r);  in mqtt_sub_unsub()
1224   mqtt_output_send(&client->output, client->conn);  in mqtt_sub_unsub()
1232  * @param client MQTT client
1238 mqtt_set_inpub_callback(mqtt_client_t *client, mqtt_incoming_publish_cb_t pub_cb,  in mqtt_set_inpub_callback()  argument
1242   LWIP_ASSERT("mqtt_set_inpub_callback: client != NULL", client != NULL);  in mqtt_set_inpub_callback()
1243   client->data_cb = data_cb;  in mqtt_set_inpub_callback()
1244   client->pub_cb = pub_cb;  in mqtt_set_inpub_callback()
1245   client->inpub_arg = arg;  in mqtt_set_inpub_callback()
1250  * Create a new MQTT client instance
1262  * Free MQTT client instance
1263  * @param client Pointer to instance to be freed
1266 mqtt_client_free(mqtt_client_t *client)  in mqtt_client_free()  argument
1268   mem_free(client);  in mqtt_client_free()
1274  * @param client MQTT client
1279  * @param client_info Client identification and connection options
1283 mqtt_client_connect(mqtt_client_t *client, const ip_addr_t *ip_addr, u16_t port, mqtt_connection_cb…  in mqtt_client_connect()  argument
1295   LWIP_ASSERT("mqtt_client_connect: client != NULL", client != NULL);  in mqtt_client_connect()
1300   if (client->conn_state != TCP_DISCONNECTED) {  in mqtt_client_connect()
1306   memset(client, 0, sizeof(mqtt_client_t));  in mqtt_client_connect()
1307   client->connect_arg = arg;  in mqtt_client_connect()
1308   client->connect_cb = cb;  in mqtt_client_connect()
1309   client->keep_alive = client_info->keep_alive;  in mqtt_client_connect()
1310   mqtt_init_requests(client->req_list, LWIP_ARRAYSIZE(client->req_list));  in mqtt_client_connect()
1361   if (mqtt_output_check_space(&client->output, remaining_length) == 0) {  in mqtt_client_connect()
1367     client->conn = altcp_tls_new(client_info->tls_config, IP_GET_TYPE(ip_addr));  in mqtt_client_connect()
1371     client->conn = altcp_tcp_new_ip_type(IP_GET_TYPE(ip_addr));  in mqtt_client_connect()
1373   if (client->conn == NULL) {  in mqtt_client_connect()
1378   altcp_arg(client->conn, client);  in mqtt_client_connect()
1380   err = altcp_bind(client->conn, IP_ADDR_ANY, 0);  in mqtt_client_connect()
1388   err = altcp_connect(client->conn, ip_addr, port, mqtt_tcp_connect_cb);  in mqtt_client_connect()
1394   altcp_err(client->conn, mqtt_tcp_err_cb);  in mqtt_client_connect()
1395   client->conn_state = TCP_CONNECTING;  in mqtt_client_connect()
1398 …mqtt_output_append_fixed_header(&client->output, MQTT_MSG_TYPE_CONNECT, 0, 0, 0, remaining_length);  in mqtt_client_connect()
1400   mqtt_output_append_string(&client->output, "MQTT", 4);  in mqtt_client_connect()
1402   mqtt_output_append_u8(&client->output, 4);  in mqtt_client_connect()
1404   mqtt_output_append_u8(&client->output, flags);  in mqtt_client_connect()
1406   mqtt_output_append_u16(&client->output, client_info->keep_alive);  in mqtt_client_connect()
1407   /* Append client id */  in mqtt_client_connect()
1408   mqtt_output_append_string(&client->output, client_info->client_id, client_id_length);  in mqtt_client_connect()
1411     mqtt_output_append_string(&client->output, client_info->will_topic, will_topic_len);  in mqtt_client_connect()
1412     mqtt_output_append_string(&client->output, client_info->will_msg, will_msg_len);  in mqtt_client_connect()
1416     mqtt_output_append_string(&client->output, client_info->client_user, client_user_len);  in mqtt_client_connect()
1420     mqtt_output_append_string(&client->output, client_info->client_pass, client_pass_len);  in mqtt_client_connect()
1425   altcp_abort(client->conn);  in mqtt_client_connect()
1426   client->conn = NULL;  in mqtt_client_connect()
1434  * @param client MQTT client
1437 mqtt_disconnect(mqtt_client_t *client)  in mqtt_disconnect()  argument
1440   LWIP_ASSERT("mqtt_disconnect: client != NULL", client);  in mqtt_disconnect()
1442   if (client->conn_state != TCP_DISCONNECTED) {  in mqtt_disconnect()
1444     client->conn_state = TCP_DISCONNECTED;  in mqtt_disconnect()
1445     mqtt_close(client, (mqtt_connection_status_t)0);  in mqtt_disconnect()
1452  * @param client MQTT client
1456 mqtt_client_is_connected(mqtt_client_t *client)  in mqtt_client_is_connected()  argument
1459   LWIP_ASSERT("mqtt_client_is_connected: client != NULL", client);  in mqtt_client_is_connected()
1460   return client->conn_state == MQTT_CONNECTED;  in mqtt_client_is_connected()