diff --git a/app/modules/mqtt.c b/app/modules/mqtt.c index f01e9aae..9b120155 100644 --- a/app/modules/mqtt.c +++ b/app/modules/mqtt.c @@ -55,7 +55,6 @@ typedef struct mqtt_state_t typedef struct lmqtt_userdata { - lua_State *L; struct espconn *pesp_conn; int self_ref; int cb_connect_ref; @@ -92,11 +91,13 @@ static void mqtt_socket_disconnected(void *arg) // tcp only os_timer_disarm(&mud->mqttTimer); + lua_State *L = lua_getstate(); + if(mud->connected){ // call back only called when socket is from connection to disconnection. mud->connected = false; - if((mud->L != NULL) && (mud->cb_disconnect_ref != LUA_NOREF) && (mud->self_ref != LUA_NOREF)) { - lua_rawgeti(mud->L, LUA_REGISTRYINDEX, mud->cb_disconnect_ref); - lua_rawgeti(mud->L, LUA_REGISTRYINDEX, mud->self_ref); // pass the userdata(client) to callback func in lua + if((mud->cb_disconnect_ref != LUA_NOREF) && (mud->self_ref != LUA_NOREF)) { + lua_rawgeti(L, LUA_REGISTRYINDEX, mud->cb_disconnect_ref); + lua_rawgeti(L, LUA_REGISTRYINDEX, mud->self_ref); // pass the userdata(client) to callback func in lua call_back = true; } } @@ -121,18 +122,14 @@ static void mqtt_socket_disconnected(void *arg) // tcp only mud->pesp_conn = NULL; } - if(mud->L == NULL) - return; - lua_gc(mud->L, LUA_GCSTOP, 0); if(mud->self_ref != LUA_NOREF){ // TODO: should we unref the client and delete it? - luaL_unref(mud->L, LUA_REGISTRYINDEX, mud->self_ref); + luaL_unref(L, LUA_REGISTRYINDEX, mud->self_ref); mud->self_ref = LUA_NOREF; // unref this, and the mqtt.socket userdata will delete it self } - lua_gc(mud->L, LUA_GCRESTART, 0); } - if((mud->L != NULL) && call_back){ - lua_call(mud->L, 1, 0); + if(call_back){ + lua_call(L, 1, 0); } NODE_DBG("leave mqtt_socket_disconnected.\n"); @@ -178,25 +175,58 @@ static void deliver_publish(lmqtt_userdata * mud, uint8_t* message, int length) return; if(mud->self_ref == LUA_NOREF) return; - if(mud->L == NULL) - return; + lua_State *L = lua_getstate(); if(event_data.topic && (event_data.topic_length > 0)){ - lua_rawgeti(mud->L, LUA_REGISTRYINDEX, mud->cb_message_ref); - lua_rawgeti(mud->L, LUA_REGISTRYINDEX, mud->self_ref); // pass the userdata to callback func in lua - lua_pushlstring(mud->L, event_data.topic, event_data.topic_length); + lua_rawgeti(L, LUA_REGISTRYINDEX, mud->cb_message_ref); + lua_rawgeti(L, LUA_REGISTRYINDEX, mud->self_ref); // pass the userdata to callback func in lua + lua_pushlstring(L, event_data.topic, event_data.topic_length); } else { NODE_DBG("get wrong packet.\n"); return; } if(event_data.data && (event_data.data_length > 0)){ - lua_pushlstring(mud->L, event_data.data, event_data.data_length); - lua_call(mud->L, 3, 0); + lua_pushlstring(L, event_data.data, event_data.data_length); + lua_call(L, 3, 0); } else { - lua_call(mud->L, 2, 0); + lua_call(L, 2, 0); } NODE_DBG("leave deliver_publish.\n"); } +static sint8 mqtt_send_if_possible(struct espconn *pesp_conn) +{ + if(pesp_conn == NULL) + return ESPCONN_OK; + lmqtt_userdata *mud = (lmqtt_userdata *)pesp_conn->reverse; + if(mud == NULL) + return ESPCONN_OK; + + sint8 espconn_status = ESPCONN_OK; + + // This indicates if we have sent something and are waiting for something to + // happen + if (mud->event_timeout == 0) { + msg_queue_t *pending_msg = msg_peek(&(mud->mqtt_state.pending_msg_q)); + if (pending_msg) { + mud->event_timeout = MQTT_SEND_TIMEOUT; + NODE_DBG("Sent: %d\n", pending_msg->msg.length); +#ifdef CLIENT_SSL_ENABLE + if( mud->secure ) + { + espconn_status = espconn_secure_send( pesp_conn, pending_msg->msg.data, pending_msg->msg.length ); + } + else +#endif + { + espconn_status = espconn_send( pesp_conn, pending_msg->msg.data, pending_msg->msg.length ); + } + mud->keep_alive_tick = 0; + } + } + NODE_DBG("send_if_poss, queue size: %d\n", msg_size(&(mud->mqtt_state.pending_msg_q))); + return espconn_status; +} + static void mqtt_socket_received(void *arg, char *pdata, unsigned short len) { NODE_DBG("enter mqtt_socket_received.\n"); @@ -204,7 +234,6 @@ static void mqtt_socket_received(void *arg, char *pdata, unsigned short len) uint8_t msg_type; uint8_t msg_qos; uint16_t msg_id; - msg_queue_t *node = NULL; int length = (int)len; // uint8_t in_buffer[MQTT_BUF_SIZE]; uint8_t *in_buffer = (uint8_t *)pdata; @@ -224,6 +253,7 @@ READPACKET: uint8_t temp_buffer[MQTT_BUF_SIZE]; mqtt_msg_init(&mud->mqtt_state.mqtt_connection, temp_buffer, MQTT_BUF_SIZE); mqtt_message_t *temp_msg = NULL; + lua_State *L = lua_getstate(); switch(mud->connState){ case MQTT_CONNECT_SENDING: case MQTT_CONNECT_SENT: @@ -247,11 +277,9 @@ READPACKET: break; if(mud->self_ref == LUA_NOREF) break; - if(mud->L == NULL) - break; - lua_rawgeti(mud->L, LUA_REGISTRYINDEX, mud->cb_connect_ref); - lua_rawgeti(mud->L, LUA_REGISTRYINDEX, mud->self_ref); // pass the userdata(client) to callback func in lua - lua_call(mud->L, 1, 0); + lua_rawgeti(L, LUA_REGISTRYINDEX, mud->cb_connect_ref); + lua_rawgeti(L, LUA_REGISTRYINDEX, mud->self_ref); // pass the userdata(client) to callback func in lua + lua_call(L, 1, 0); break; } break; @@ -280,11 +308,9 @@ READPACKET: break; if (mud->self_ref == LUA_NOREF) break; - if(mud->L == NULL) - break; - lua_rawgeti(mud->L, LUA_REGISTRYINDEX, mud->cb_suback_ref); - lua_rawgeti(mud->L, LUA_REGISTRYINDEX, mud->self_ref); - lua_call(mud->L, 1, 0); + lua_rawgeti(L, LUA_REGISTRYINDEX, mud->cb_suback_ref); + lua_rawgeti(L, LUA_REGISTRYINDEX, mud->self_ref); + lua_call(L, 1, 0); } break; case MQTT_MSG_TYPE_UNSUBACK: @@ -296,12 +322,12 @@ READPACKET: case MQTT_MSG_TYPE_PUBLISH: if(msg_qos == 1){ temp_msg = mqtt_msg_puback(&mud->mqtt_state.mqtt_connection, msg_id); - node = msg_enqueue(&(mud->mqtt_state.pending_msg_q), temp_msg, + msg_enqueue(&(mud->mqtt_state.pending_msg_q), temp_msg, msg_id, MQTT_MSG_TYPE_PUBACK, (int)mqtt_get_qos(temp_msg->data) ); } else if(msg_qos == 2){ temp_msg = mqtt_msg_pubrec(&mud->mqtt_state.mqtt_connection, msg_id); - node = msg_enqueue(&(mud->mqtt_state.pending_msg_q), temp_msg, + msg_enqueue(&(mud->mqtt_state.pending_msg_q), temp_msg, msg_id, MQTT_MSG_TYPE_PUBREC, (int)mqtt_get_qos(temp_msg->data) ); } if(msg_qos == 1 || msg_qos == 2){ @@ -317,11 +343,9 @@ READPACKET: break; if(mud->self_ref == LUA_NOREF) break; - if(mud->L == NULL) - break; - lua_rawgeti(mud->L, LUA_REGISTRYINDEX, mud->cb_puback_ref); - lua_rawgeti(mud->L, LUA_REGISTRYINDEX, mud->self_ref); // pass the userdata to callback func in lua - lua_call(mud->L, 1, 0); + lua_rawgeti(L, LUA_REGISTRYINDEX, mud->cb_puback_ref); + lua_rawgeti(L, LUA_REGISTRYINDEX, mud->self_ref); // pass the userdata to callback func in lua + lua_call(L, 1, 0); } break; @@ -331,7 +355,7 @@ READPACKET: // Note: actrually, should not destroy the msg until PUBCOMP is received. msg_destroy(msg_dequeue(&(mud->mqtt_state.pending_msg_q))); temp_msg = mqtt_msg_pubrel(&mud->mqtt_state.mqtt_connection, msg_id); - node = msg_enqueue(&(mud->mqtt_state.pending_msg_q), temp_msg, + msg_enqueue(&(mud->mqtt_state.pending_msg_q), temp_msg, msg_id, MQTT_MSG_TYPE_PUBREL, (int)mqtt_get_qos(temp_msg->data) ); NODE_DBG("MQTT: Response PUBREL\r\n"); } @@ -340,7 +364,7 @@ READPACKET: if(pending_msg && pending_msg->msg_type == MQTT_MSG_TYPE_PUBREC && pending_msg->msg_id == msg_id){ msg_destroy(msg_dequeue(&(mud->mqtt_state.pending_msg_q))); temp_msg = mqtt_msg_pubcomp(&mud->mqtt_state.mqtt_connection, msg_id); - node = msg_enqueue(&(mud->mqtt_state.pending_msg_q), temp_msg, + msg_enqueue(&(mud->mqtt_state.pending_msg_q), temp_msg, msg_id, MQTT_MSG_TYPE_PUBCOMP, (int)mqtt_get_qos(temp_msg->data) ); NODE_DBG("MQTT: Response PUBCOMP\r\n"); } @@ -353,16 +377,14 @@ READPACKET: break; if(mud->self_ref == LUA_NOREF) break; - if(mud->L == NULL) - break; - lua_rawgeti(mud->L, LUA_REGISTRYINDEX, mud->cb_puback_ref); - lua_rawgeti(mud->L, LUA_REGISTRYINDEX, mud->self_ref); // pass the userdata to callback func in lua - lua_call(mud->L, 1, 0); + lua_rawgeti(L, LUA_REGISTRYINDEX, mud->cb_puback_ref); + lua_rawgeti(L, LUA_REGISTRYINDEX, mud->self_ref); // pass the userdata to callback func in lua + lua_call(L, 1, 0); } break; case MQTT_MSG_TYPE_PINGREQ: temp_msg = mqtt_msg_pingresp(&mud->mqtt_state.mqtt_connection); - node = msg_enqueue(&(mud->mqtt_state.pending_msg_q), temp_msg, + msg_enqueue(&(mud->mqtt_state.pending_msg_q), temp_msg, msg_id, MQTT_MSG_TYPE_PINGRESP, (int)mqtt_get_qos(temp_msg->data) ); NODE_DBG("MQTT: Response PINGRESP\r\n"); break; @@ -391,21 +413,7 @@ READPACKET: break; } - if(node && (1==msg_size(&(mud->mqtt_state.pending_msg_q))) && mud->event_timeout == 0){ - mud->event_timeout = MQTT_SEND_TIMEOUT; - NODE_DBG("Sent: %d\n", node->msg.length); -#ifdef CLIENT_SSL_ENABLE - if( mud->secure ) - { - espconn_secure_send( pesp_conn, node->msg.data, node->msg.length ); - } - else -#endif - { - espconn_send( pesp_conn, node->msg.data, node->msg.length ); - } - } - NODE_DBG("receive, queue size: %d\n", msg_size(&(mud->mqtt_state.pending_msg_q))); + mqtt_send_if_possible(pesp_conn); NODE_DBG("leave mqtt_socket_received.\n"); return; } @@ -452,22 +460,7 @@ static void mqtt_socket_sent(void *arg) try_send = 0; } if (try_send) { - msg_queue_t *node = msg_peek(&(mud->mqtt_state.pending_msg_q)); - if (node) { - mud->event_timeout = MQTT_SEND_TIMEOUT; - NODE_DBG("Sent: %d\n", node->msg.length); -#ifdef CLIENT_SSL_ENABLE - if( mud->secure ) - { - (void) espconn_secure_send( mud->pesp_conn, node->msg.data, node->msg.length ); - } - else -#endif - { - (void) espconn_send( mud->pesp_conn, node->msg.data, node->msg.length ); - } - mud->keep_alive_tick = 0; - } + mqtt_send_if_possible(mud->pesp_conn); } NODE_DBG("sent2, queue size: %d\n", msg_size(&(mud->mqtt_state.pending_msg_q))); NODE_DBG("leave mqtt_socket_sent.\n"); @@ -561,42 +554,18 @@ void mqtt_socket_timer(void *arg) } else if(mud->connState == MQTT_DATA){ msg_queue_t *pending_msg = msg_peek(&(mud->mqtt_state.pending_msg_q)); if(pending_msg){ - mud->event_timeout = MQTT_SEND_TIMEOUT; -#ifdef CLIENT_SSL_ENABLE - if(mud->secure) - { - espconn_secure_send(mud->pesp_conn, pending_msg->msg.data, pending_msg->msg.length); - } - else -#endif - { - espconn_send(mud->pesp_conn, pending_msg->msg.data, pending_msg->msg.length); - } - mud->keep_alive_tick = 0; - NODE_DBG("id: %d - qos: %d, length: %d\n", pending_msg->msg_id, pending_msg->publish_qos, pending_msg->msg.length); + mqtt_send_if_possible(mud->pesp_conn); } else { // no queued event. mud->keep_alive_tick ++; if(mud->keep_alive_tick > mud->mqtt_state.connect_info->keepalive){ - mud->event_timeout = MQTT_SEND_TIMEOUT; uint8_t temp_buffer[MQTT_BUF_SIZE]; mqtt_msg_init(&mud->mqtt_state.mqtt_connection, temp_buffer, MQTT_BUF_SIZE); NODE_DBG("\r\nMQTT: Send keepalive packet\r\n"); mqtt_message_t* temp_msg = mqtt_msg_pingreq(&mud->mqtt_state.mqtt_connection); msg_queue_t *node = msg_enqueue( &(mud->mqtt_state.pending_msg_q), temp_msg, 0, MQTT_MSG_TYPE_PINGREQ, (int)mqtt_get_qos(temp_msg->data) ); - // only one message in queue, send immediately. -#ifdef CLIENT_SSL_ENABLE - if(mud->secure) - { - espconn_secure_send(mud->pesp_conn, temp_msg->data, temp_msg->length); - } - else -#endif - { - espconn_send(mud->pesp_conn, temp_msg->data, temp_msg->length); - } - mud->keep_alive_tick = 0; + mqtt_send_if_possible(mud->pesp_conn); } } } @@ -626,7 +595,6 @@ static int mqtt_socket_client( lua_State* L ) // create a object mud = (lmqtt_userdata *)lua_newuserdata(L, sizeof(lmqtt_userdata)); // pre-initialize it, in case of errors - mud->L = NULL; mud->self_ref = LUA_NOREF; mud->cb_connect_ref = LUA_NOREF; mud->cb_disconnect_ref = LUA_NOREF; @@ -651,8 +619,6 @@ static int mqtt_socket_client( lua_State* L ) luaL_getmetatable(L, "mqtt.socket"); lua_setmetatable(L, -2); - mud->L = L; // L for mqtt module. - if( lua_isstring(L,stack) ) // deal with the clientid string { clientId = luaL_checklstring( L, stack, &idl ); @@ -1187,51 +1153,49 @@ static int mqtt_socket_subscribe( lua_State* L ) { NODE_DBG("subscribe table\n"); lua_pushnil( L ); /* first key */ - uint8_t temp_buf[MQTT_BUF_SIZE]; - uint32_t temp_pos = 0; + int topic_count = 0; uint8_t overflow = 0; while( lua_next( L, stack ) != 0 ) { topic = luaL_checkstring( L, -2 ); qos = luaL_checkinteger( L, -1 ); - temp_msg = mqtt_msg_subscribe( &mud->mqtt_state.mqtt_connection, topic, qos, &msg_id ); + if (topic_count == 0) { + temp_msg = mqtt_msg_subscribe_init( &mud->mqtt_state.mqtt_connection, &msg_id ); + } + temp_msg = mqtt_msg_subscribe_topic( &mud->mqtt_state.mqtt_connection, topic, qos ); + topic_count++; + NODE_DBG("topic: %s - qos: %d, length: %d\n", topic, qos, temp_msg->length); - if (temp_pos + temp_msg->length > MQTT_BUF_SIZE){ + if (temp_msg->length == 0) { lua_pop(L, 1); overflow = 1; break; // too long message for the outbuffer. } - c_memcpy( temp_buf + temp_pos, temp_msg->data, temp_msg->length ); - temp_pos += temp_msg->length; lua_pop( L, 1 ); } - if (temp_pos == 0){ - luaL_error( L, "invalid data" ); - lua_pushboolean(L, 0); - return 1; + if (topic_count == 0){ + return luaL_error( L, "no topics found" ); } if (overflow != 0){ - luaL_error( L, "buffer overflow, can't enqueue all subscriptions" ); - lua_pushboolean(L, 0); - return 1; + return luaL_error( L, "buffer overflow, can't enqueue all subscriptions" ); + } + + temp_msg = mqtt_msg_subscribe_fini( &mud->mqtt_state.mqtt_connection ); + if (temp_msg->length == 0) { + return luaL_error( L, "buffer overflow, can't enqueue all subscriptions" ); } - c_memcpy( temp_buffer, temp_buf, temp_pos ); - temp_msg->data = temp_buffer; - temp_msg->length = temp_pos; stack++; } else { NODE_DBG("subscribe string\n"); topic = luaL_checklstring( L, stack, &il ); stack++; if( topic == NULL ){ - luaL_error( L, "need topic name" ); - lua_pushboolean(L, 0); - return 1; + return luaL_error( L, "need topic name" ); } qos = luaL_checkinteger( L, stack ); temp_msg = mqtt_msg_subscribe( &mud->mqtt_state.mqtt_connection, topic, qos, &msg_id ); @@ -1252,21 +1216,7 @@ static int mqtt_socket_subscribe( lua_State* L ) { sint8 espconn_status = ESPCONN_IF; - if(node && (1==msg_size(&(mud->mqtt_state.pending_msg_q))) && mud->event_timeout == 0){ - mud->event_timeout = MQTT_SEND_TIMEOUT; - NODE_DBG("Sent: %d\n", node->msg.length); -#ifdef CLIENT_SSL_ENABLE - if( mud->secure ) - { - espconn_status = espconn_secure_send( mud->pesp_conn, node->msg.data, node->msg.length ); - } - else -#endif - { - espconn_status = espconn_send( mud->pesp_conn, node->msg.data, node->msg.length ); - } - mud->keep_alive_tick = 0; - } + espconn_status = mqtt_send_if_possible(mud->pesp_conn); if(!node || espconn_status != ESPCONN_OK){ lua_pushboolean(L, 0); @@ -1304,17 +1254,13 @@ static int mqtt_socket_publish( lua_State* L ) } if(!mud->connected){ - luaL_error( L, "not connected" ); - lua_pushboolean(L, 0); - return 1; + return luaL_error( L, "not connected" ); } const char *topic = luaL_checklstring( L, stack, &l ); stack ++; if (topic == NULL){ - luaL_error( L, "need topic" ); - lua_pushboolean(L, 0); - return 1; + return luaL_error( L, "need topic" ); } const char *payload = luaL_checklstring( L, stack, &l ); @@ -1343,21 +1289,7 @@ static int mqtt_socket_publish( lua_State* L ) sint8 espconn_status = ESPCONN_OK; - if(node && (1==msg_size(&(mud->mqtt_state.pending_msg_q))) && mud->event_timeout == 0){ - mud->event_timeout = MQTT_SEND_TIMEOUT; - NODE_DBG("Sent: %d\n", node->msg.length); -#ifdef CLIENT_SSL_ENABLE - if( mud->secure ) - { - espconn_status = espconn_secure_send( mud->pesp_conn, node->msg.data, node->msg.length ); - } - else -#endif - { - espconn_status = espconn_send( mud->pesp_conn, node->msg.data, node->msg.length ); - } - mud->keep_alive_tick = 0; - } + espconn_status = mqtt_send_if_possible(mud->pesp_conn); if(!node || espconn_status != ESPCONN_OK){ lua_pushboolean(L, 0); diff --git a/app/mqtt/mqtt_msg.c b/app/mqtt/mqtt_msg.c index 9c405a7c..48c80e09 100644 --- a/app/mqtt/mqtt_msg.c +++ b/app/mqtt/mqtt_msg.c @@ -402,14 +402,19 @@ mqtt_message_t* mqtt_msg_pubcomp(mqtt_connection_t* connection, uint16_t message return fini_message(connection, MQTT_MSG_TYPE_PUBCOMP, 0, 0, 0); } -mqtt_message_t* mqtt_msg_subscribe(mqtt_connection_t* connection, const char* topic, int qos, uint16_t* message_id) +mqtt_message_t* mqtt_msg_subscribe_init(mqtt_connection_t* connection, uint16_t *message_id) { init_message(connection); - if(topic == NULL || topic[0] == '\0') + if((*message_id = append_message_id(connection, 0)) == 0) return fail_message(connection); - if((*message_id = append_message_id(connection, 0)) == 0) + return &connection->message; +} + +mqtt_message_t* mqtt_msg_subscribe_topic(mqtt_connection_t* connection, const char* topic, int qos) +{ + if(topic == NULL || topic[0] == '\0') return fail_message(connection); if(append_string(connection, topic, c_strlen(topic)) < 0) @@ -419,9 +424,29 @@ mqtt_message_t* mqtt_msg_subscribe(mqtt_connection_t* connection, const char* to return fail_message(connection); connection->buffer[connection->message.length++] = qos; + return &connection->message; +} + +mqtt_message_t* mqtt_msg_subscribe_fini(mqtt_connection_t* connection) +{ return fini_message(connection, MQTT_MSG_TYPE_SUBSCRIBE, 0, 1, 0); } +mqtt_message_t* mqtt_msg_subscribe(mqtt_connection_t* connection, const char* topic, int qos, uint16_t* message_id) +{ + mqtt_message_t* result; + + result = mqtt_msg_subscribe_init(connection, message_id); + if (result->length != 0) { + result = mqtt_msg_subscribe_topic(connection, topic, qos); + } + if (result->length != 0) { + result = mqtt_msg_subscribe_fini(connection); + } + + return result; +} + mqtt_message_t* mqtt_msg_unsubscribe(mqtt_connection_t* connection, const char* topic, uint16_t* message_id) { init_message(connection); diff --git a/app/mqtt/mqtt_msg.h b/app/mqtt/mqtt_msg.h index 225ba642..e1f372b9 100644 --- a/app/mqtt/mqtt_msg.h +++ b/app/mqtt/mqtt_msg.h @@ -120,6 +120,10 @@ mqtt_message_t* mqtt_msg_pingreq(mqtt_connection_t* connection); mqtt_message_t* mqtt_msg_pingresp(mqtt_connection_t* connection); mqtt_message_t* mqtt_msg_disconnect(mqtt_connection_t* connection); +mqtt_message_t* mqtt_msg_subscribe_init(mqtt_connection_t* connection, uint16_t* message_id); +mqtt_message_t* mqtt_msg_subscribe_topic(mqtt_connection_t* connection, const char* topic, int qos); +mqtt_message_t* mqtt_msg_subscribe_fini(mqtt_connection_t* connection); + #ifdef __cplusplus }