Hi Guys ,
I have used AWS IOT Example as part of my IOT project to start the connection with AWS
I have problem now but I try to solve it but I can’t find or understand why this problem happen … I hope
you can help me in that
here what I noticed
when I used the code of NuMaker-mbed-AWS-IoT-example the initialization of the component it work fine and the conenction opened and can publish the message
but when I started RTOS timer and timer is expired to read date from sensor and then use the code to open the connection and send the value to aws the connection is opend but nothing is completed
Start from main …
configManager_init() …
startTimer()
TEMPERATURE timer start successfully…
temperatureTimerExpired()
sal_getTemperatureValue
temperatureValue : 34
LOG_INFO :Mode not changed still NORMAL
LOG_INFO : send Data Immediately to the cloud
LOG_INFO :Data sending to the cloud
LOG_INFO :cal_init
Starting AWS IoT test
Using Mbed OS 5.15.3
Connected to the network successfully. IP address: 192.168.2.101
Connecting with a2u5552lvo1kh8.iot.us-west-2.amazonaws.com:8883
the code I used :
/*******************************************************************************
-
File name : cal.cpp
-
Created on : Mar 22, 2020
-
Author : Esraa Hatem
-
Description : This file cal contain the implementation of all function related to CAL
*******************************************************************************/
/============================= INCLUDES ====================================/
#include"cal.h"
/============================= MACROS ====================================/
NetworkInterface *net
/============================= ClASSES ===================================/
#if AWS_IOT_MQTT_TEST
/**
- /brief AWS_IoT_MQTT_Test implements the logic with AWS IoT User/Thing Shadow topics through MQTT.
*/
class AWS_IoT_MQTT_Test {
public:
/**
* @brief AWS_IoT_MQTT_Test Constructor
*
* @param[in] domain Domain name of the MQTT server
* @param[in] port Port number of the MQTT server
* @param[in] net_iface Network interface
*/
AWS_IoT_MQTT_Test(const char * domain, const uint16_t port, NetworkInterface *net_iface) :
_domain(domain), _port(port) {
_tlssocket = new MyTLSSocket(net_iface, SSL_CA_CERT_PEM, SSL_USER_CERT_PEM, SSL_USER_PRIV_KEY_PEM);
/* Blocking mode */
_tlssocket->set_blocking(true);
/* Print Mbed TLS handshake log */
_tlssocket->set_debug(true);
_mqtt_client = new MQTT::Client<MyTLSSocket, Countdown, MAX_MQTT_PACKET_SIZE>(*_tlssocket);
}
/**
* @brief AWS_IoT_MQTT_Test Destructor
*/
~AWS_IoT_MQTT_Test() {
delete _mqtt_client;
_mqtt_client = NULL;
_tlssocket->close();
delete _tlssocket;
_tlssocket = NULL;
}
/**
* @brief Start AWS IoT test through MQTT
*/
void start_test() {
int tls_rc;
int mqtt_rc;
do {
/* Connect to the server */
/* Initialize TLS-related stuff */
printf("Connecting with %s:%d\n", _domain, _port);
tls_rc = _tlssocket->connect(_domain, _port);
if (tls_rc != NSAPI_ERROR_OK) {
printf("Connects with %s:%d failed: %d\n", _domain, _port, tls_rc);
break;
}
printf("Connects with %s:%d OK\n", _domain, _port);
/* See the link below for AWS IoT support for MQTT:
* http://docs.aws.amazon.com/iot/latest/developerguide/protocols.html */
/* MQTT connect */
/* The message broker does not support persistent sessions (connections made with
* the cleanSession flag set to false. */
MQTTPacket_connectData conn_data = MQTTPacket_connectData_initializer;
/* AWS IoT message broker implementation is based on MQTT version 3.1.1
* 3 = 3.1
* 4 = 3.1.1 */
conn_data.MQTTVersion = 4;
/* Version number of this structure. Must be 0 */
conn_data.struct_version = 0;
/* The message broker uses the client ID to identify each client. The client ID is passed
* in from the client to the message broker as part of the MQTT payload. Two clients with
* the same client ID are not allowed to be connected concurrently to the message broker.
* When a client connects to the message broker using a client ID that another client is using,
* a CONNACK message will be sent to both clients and the currently connected client will be
* disconnected. */
conn_data.clientID.cstring = AWS_IOT_MQTT_CLIENTNAME;
/* The message broker does not support persistent sessions (connections made with
* the cleanSession flag set to false. The AWS IoT message broker assumes all sessions
* are clean sessions and messages are not stored across sessions. If an MQTT client
* attempts to connect to the AWS IoT message broker with the cleanSession set to false,
* the client will be disconnected. */
conn_data.cleansession = 1;
//conn_data.username.cstring = "USERNAME";
//conn_data.password.cstring = "PASSWORD";
MQTT::connackData connack_data;
/* _tlssocket must connect to the network endpoint before calling this. */
printf("MQTT connecting");
if ((mqtt_rc = _mqtt_client->connect(conn_data, connack_data)) != 0) {
printf("\rMQTT connects failed: %d\n", mqtt_rc);
break;
}
printf("\rMQTT connects OK\n\n");
/* Subscribe/publish user topic */
printf("Subscribing/publishing user topic\n");
if (! sub_pub_topic(USER_MQTT_TOPIC, USER_MQTT_TOPIC_FILTERS, sizeof (USER_MQTT_TOPIC_FILTERS) / sizeof (USER_MQTT_TOPIC_FILTERS[0]), USER_MQTT_TOPIC_PUBLISH_MESSAGE)) {
break;
}
printf("Subscribes/publishes user topic OK\n\n");
/* Subscribe/publish UpdateThingShadow topic */
printf("Subscribing/publishing UpdateThingShadow topic\n");
if (! sub_pub_topic(UPDATETHINGSHADOW_MQTT_TOPIC, UPDATETHINGSHADOW_MQTT_TOPIC_FILTERS, sizeof (UPDATETHINGSHADOW_MQTT_TOPIC_FILTERS) / sizeof (UPDATETHINGSHADOW_MQTT_TOPIC_FILTERS[0]), UPDATETHINGSHADOW_MQTT_TOPIC_PUBLISH_MESSAGE)) {
break;
}
printf("Subscribes/publishes UpdateThingShadow topic OK\n\n");
/* Subscribe/publish GetThingShadow topic */
printf("Subscribing/publishing GetThingShadow topic\n");
if (! sub_pub_topic(GETTHINGSHADOW_MQTT_TOPIC, GETTHINGSHADOW_MQTT_TOPIC_FILTERS, sizeof (GETTHINGSHADOW_MQTT_TOPIC_FILTERS) / sizeof (GETTHINGSHADOW_MQTT_TOPIC_FILTERS[0]), GETTHINGSHADOW_MQTT_TOPIC_PUBLISH_MESSAGE)) {
break;
}
printf("Subscribes/publishes GetThingShadow topic OK\n\n");
/* Subscribe/publish DeleteThingShadow topic */
printf("Subscribing/publishing DeleteThingShadow topic\n");
if (! sub_pub_topic(DELETETHINGSHADOW_MQTT_TOPIC, DELETETHINGSHADOW_MQTT_TOPIC_FILTERS, sizeof (DELETETHINGSHADOW_MQTT_TOPIC_FILTERS) / sizeof (DELETETHINGSHADOW_MQTT_TOPIC_FILTERS[0]), DELETETHINGSHADOW_MQTT_TOPIC_PUBLISH_MESSAGE)) {
break;
}
printf("Subscribes/publishes DeleteThingShadow topic OK\n\n");
} while (0);
printf("MQTT disconnecting");
if ((mqtt_rc = _mqtt_client->disconnect()) != 0) {
printf("\rMQTT disconnects failed %d\n\n", mqtt_rc);
}
printf("\rMQTT disconnects OK\n\n");
_tlssocket->close();
}
protected:
/**
* @brief Subscribe/publish specific topic
*/
bool sub_pub_topic(const char *topic, const char **topic_filters, size_t topic_filters_size, const char *publish_message_body) {
bool ret = false;
int mqtt_rc;
do {
const char **topic_filter;
const char **topic_filter_end = topic_filters + topic_filters_size;
for (topic_filter = topic_filters; topic_filter != topic_filter_end; topic_filter ++) {
/* AWS IoT does not support publishing and subscribing with QoS 2.
* The AWS IoT message broker does not send a PUBACK or SUBACK when QoS 2 is requested. */
printf("MQTT subscribing to %s", *topic_filter);
if ((mqtt_rc = _mqtt_client->subscribe(*topic_filter, MQTT::QOS1, message_arrived)) != 0) {
printf("\rMQTT subscribes to %s failed: %d\n", *topic_filter, mqtt_rc);
continue;
}
printf("\rMQTT subscribes to %s OK\n", *topic_filter);
}
/* Clear count of received message with subscribed topic */
clear_message_arrive_count();
MQTT::Message message;
int _bpos;
_bpos = snprintf(_buffer, sizeof (_buffer) - 1, publish_message_body);
if (_bpos < 0 || ((size_t) _bpos) > (sizeof (_buffer) - 1)) {
printf("snprintf failed: %d\n", _bpos);
break;
}
_buffer[_bpos] = 0;
/* AWS IoT does not support publishing and subscribing with QoS 2.
* The AWS IoT message broker does not send a PUBACK or SUBACK when QoS 2 is requested. */
message.qos = MQTT::QOS1;
message.retained = false;
message.dup = false;
message.payload = _buffer;
message.payloadlen = strlen(_buffer);
/* Print publish message */
printf("Message to publish:\n");
printf("%s\n", _buffer);
printf("MQTT publishing message to %s", topic);
if ((mqtt_rc = _mqtt_client->publish(topic, message)) != 0) {
printf("\rMQTT publishes message to %s failed: %d\n", topic, mqtt_rc);
break;
}
printf("\rMQTT publishes message to %s OK\n", topic);
/* Receive message with subscribed topic */
printf("MQTT receives message with subscribed %s...\n", topic);
Timer timer;
timer.start();
while (! _message_arrive_count) {
if (timer.read_ms() >= MQTT_RECEIVE_MESSAGE_WITH_SUBSCRIBED_TOPIC_TIMEOUT_MS) {
printf("MQTT receives message with subscribed %s TIMEOUT\n", topic);
break;
}
_mqtt_client->yield(100);
}
if (_message_arrive_count) {
printf("MQTT receives message with subscribed %s OK\n", topic);
}
printf("\n");
/* Unsubscribe
* We meet second unsubscribe failed. This is caused by MQTT lib bug. */
for (topic_filter = topic_filters; topic_filter != topic_filter_end; topic_filter ++) {
printf("MQTT unsubscribing from %s", *topic_filter);
if ((mqtt_rc = _mqtt_client->unsubscribe(*topic_filter)) != 0) {
printf("\rMQTT unsubscribes from %s failed: %d\n", *topic_filter, mqtt_rc);
continue;
}
printf("\rMQTT unsubscribes from %s OK\n", *topic_filter);
}
ret = true;
} while (0);
return ret;
}
protected:
MyTLSSocket * _tlssocket;
MQTT::Client<MyTLSSocket, Countdown, MAX_MQTT_PACKET_SIZE> * _mqtt_client;
const char *_domain; /**< Domain name of the MQTT server */
const uint16_t _port; /**< Port number of the MQTT server */
char _buffer[MQTT_USER_BUFFER_SIZE]; /**< User buffer */
private:
static volatile uint16_t _message_arrive_count;
static void message_arrived(MQTT::MessageData& md) {
MQTT::Message &message = md.message;
printf("Message arrived: qos %d, retained %d, dup %d, packetid %d\r\n", message.qos, message.retained, message.dup, message.id);
printf("Payload:\n");
printf("%.*s\n", message.payloadlen, (char*)message.payload);
++ _message_arrive_count;
}
static void clear_message_arrive_count() {
_message_arrive_count = 0;
}
};
volatile uint16_t AWS_IoT_MQTT_Test::_message_arrive_count = 0;
#endif // End of AWS_IOT_MQTT_TEST
#if AWS_IOT_HTTPS_TEST
/**
-
/brief AWS_IoT_HTTPS_Test implements the logic with AWS IoT User/Thing Shadow topics (publish-only)
-
and Thing Shadow RESTful API through HTTPS.
*/
class AWS_IoT_HTTPS_Test {
public:
/**
* @brief AWS_IoT_HTTPS_Test Constructor
*
* @param[in] domain Domain name of the HTTPS server
* @param[in] port Port number of the HTTPS server
* @param[in] net_iface Network interface
*/
AWS_IoT_HTTPS_Test(const char * domain, const uint16_t port, NetworkInterface *net_iface) :
_domain(domain), _port(port) {
_tlssocket = new MyTLSSocket(net_iface, SSL_CA_CERT_PEM, SSL_USER_CERT_PEM, SSL_USER_PRIV_KEY_PEM);
/* Non-blocking mode */
_tlssocket->set_blocking(false);
/* Print Mbed TLS handshake log */
_tlssocket->set_debug(true);
}
/**
* @brief AWS_IoT_HTTPS_Test Destructor
*/
~AWS_IoT_HTTPS_Test() {
_tlssocket->close();
delete _tlssocket;
_tlssocket = NULL;
}
/**
* @brief Start AWS IoT test through HTTPS
*
* @param[in] path The path of the file to fetch from the HTTPS server
*/
void start_test() {
int tls_rc;
do {
/* Connect to the server */
/* Initialize TLS-related stuff */
printf("Connecting with %s:%d\n", _domain, _port);
tls_rc = _tlssocket->connect(_domain, _port);
if (tls_rc != NSAPI_ERROR_OK) {
printf("Connects with %s:%d failed: %d\n", _domain, _port, tls_rc);
break;
}
printf("Connects with %s:%d OK\n\n", _domain, _port);
/* Publish to user topic through HTTPS/POST */
printf("Publishing to user topic through HTTPS/POST\n");
if (! run_req_resp(USER_TOPIC_HTTPS_PATH, USER_TOPIC_HTTPS_REQUEST_METHOD, USER_TOPIC_HTTPS_REQUEST_MESSAGE_BODY)) {
break;
}
printf("Publishes to user topic through HTTPS/POST OK\n\n");
/* Update thing shadow by publishing to UpdateThingShadow topic through HTTPS/POST */
printf("Updating thing shadow by publishing to Update Thing Shadow topic through HTTPS/POST\n");
if (! run_req_resp(UPDATETHINGSHADOW_TOPIC_HTTPS_PATH, UPDATETHINGSHADOW_TOPIC_HTTPS_REQUEST_METHOD, UPDATETHINGSHADOW_TOPIC_HTTPS_REQUEST_MESSAGE_BODY)) {
break;
}
printf("Update thing shadow by publishing to Update Thing Shadow topic through HTTPS/POST OK\n\n");
/* Get thing shadow by publishing to GetThingShadow topic through HTTPS/POST */
printf("Getting thing shadow by publishing to GetThingShadow topic through HTTPS/POST\n");
if (! run_req_resp(GETTHINGSHADOW_TOPIC_HTTPS_PATH, GETTHINGSHADOW_TOPIC_HTTPS_REQUEST_METHOD, GETTHINGSHADOW_TOPIC_HTTPS_REQUEST_MESSAGE_BODY)) {
break;
}
printf("Get thing shadow by publishing to GetThingShadow topic through HTTPS/POST OK\n\n");
/* Delete thing shadow by publishing to DeleteThingShadow topic through HTTPS/POST */
printf("Deleting thing shadow by publishing to DeleteThingShadow topic through HTTPS/POST\n");
if (! run_req_resp(DELETETHINGSHADOW_TOPIC_HTTPS_PATH, DELETETHINGSHADOW_TOPIC_HTTPS_REQUEST_METHOD, DELETETHINGSHADOW_TOPIC_HTTPS_REQUEST_MESSAGE_BODY)) {
break;
}
printf("Delete thing shadow by publishing to DeleteThingShadow topic through HTTPS/POST OK\n\n");
/* Update thing shadow RESTfully through HTTPS/POST */
printf("Updating thing shadow RESTfully through HTTPS/POST\n");
if (! run_req_resp(UPDATETHINGSHADOW_THING_HTTPS_PATH, UPDATETHINGSHADOW_THING_HTTPS_REQUEST_METHOD, UPDATETHINGSHADOW_THING_HTTPS_REQUEST_MESSAGE_BODY)) {
break;
}
printf("Update thing shadow RESTfully through HTTPS/POST OK\n\n");
/* Get thing shadow RESTfully through HTTPS/GET */
printf("Getting thing shadow RESTfully through HTTPS/GET\n");
if (! run_req_resp(GETTHINGSHADOW_THING_HTTPS_PATH, GETTHINGSHADOW_THING_HTTPS_REQUEST_METHOD, GETTHINGSHADOW_THING_HTTPS_REQUEST_MESSAGE_BODY)) {
break;
}
printf("Get thing shadow RESTfully through HTTPS/GET OK\n\n");
/* Delete thing shadow RESTfully through HTTPS/DELETE */
printf("Deleting thing shadow RESTfully through HTTPS/DELETE\n");
if (! run_req_resp(DELETETHINGSHADOW_THING_HTTPS_PATH, DELETETHINGSHADOW_THING_HTTPS_REQUEST_METHOD, DELETETHINGSHADOW_THING_HTTPS_REQUEST_MESSAGE_BODY)) {
break;
}
printf("Delete thing shadow RESTfully through HTTPS/DELETE OK\n\n");
} while (0);
/* Close socket */
_tlssocket->close();
}
protected:
/**
* @brief Run request/response through HTTPS
*/
bool run_req_resp(const char *https_path, const char *https_request_method, const char *https_request_message_body) {
bool ret = false;
do {
int tls_rc;
bool _got200 = false;
int _bpos;
/* Fill the request buffer */
_bpos = snprintf(_buffer, sizeof(_buffer) - 1,
"%s %s HTTP/1.1\r\n" "Host: %s\r\n" "Content-Length: %d\r\n" "\r\n" "%s",
https_request_method, https_path, AWS_IOT_HTTPS_SERVER_NAME, strlen(https_request_message_body), https_request_message_body);
if (_bpos < 0 || ((size_t) _bpos) > (sizeof (_buffer) - 1)) {
printf("snprintf failed: %d\n", _bpos);
break;
}
_buffer[_bpos] = 0;
/* Print request message */
printf("HTTPS: Request message:\n");
printf("%s\n", _buffer);
int offset = 0;
do {
tls_rc = _tlssocket->send((const unsigned char *) _buffer + offset, _bpos - offset);
if (tls_rc > 0) {
offset += tls_rc;
}
} while (offset < _bpos &&
(tls_rc > 0 || tls_rc == MBEDTLS_ERR_SSL_WANT_READ || tls_rc == MBEDTLS_ERR_SSL_WANT_WRITE));
if (tls_rc < 0) {
print_mbedtls_error("_tlssocket->send", tls_rc);
break;
}
/* Read data out of the socket */
offset = 0;
size_t content_length = 0;
size_t offset_end = 0;
char *line_beg = _buffer;
char *line_end = NULL;
do {
tls_rc = _tlssocket->recv((unsigned char *) _buffer + offset, sizeof(_buffer) - offset - 1);
if (tls_rc > 0) {
offset += tls_rc;
}
/* Make it null-terminated */
_buffer[offset] = 0;
/* Scan response message
*
* 1. A status line which includes the status code and reason message (e.g., HTTP/1.1 200 OK)
* 2. Response header fields (e.g., Content-Type: text/html)
* 3. An empty line (\r\n)
* 4. An optional message body
*/
if (! offset_end) {
line_end = strstr(line_beg, "\r\n");
if (line_end) {
/* Scan status line */
if (! _got200) {
_got200 = strstr(line_beg, HTTPS_OK_STR) != NULL;
}
/* Scan response header fields for Content-Length
*
* NOTE: Assume chunked transfer (Transfer-Encoding: chunked) is not used
* NOTE: Assume response field name are in lower case
*/
if (content_length == 0) {
sscanf(line_beg, "content-length:%d", &content_length);
}
/* An empty line indicates end of response header fields */
if (line_beg == line_end) {
offset_end = line_end - _buffer + 2 + content_length;
}
/* Go to next line */
line_beg = line_end + 2;
line_end = NULL;
}
}
} while ((offset_end == 0 || offset < offset_end) &&
(tls_rc > 0 || tls_rc == MBEDTLS_ERR_SSL_WANT_READ || tls_rc == MBEDTLS_ERR_SSL_WANT_WRITE));
if (tls_rc < 0 &&
tls_rc != MBEDTLS_ERR_SSL_WANT_READ &&
tls_rc != MBEDTLS_ERR_SSL_WANT_WRITE) {
print_mbedtls_error("_tlssocket->read", tls_rc);
break;
}
_bpos = offset;
_buffer[_bpos] = 0;
/* Print status messages */
printf("HTTPS: Received %d chars from server\n", _bpos);
printf("HTTPS: Received 200 OK status ... %s\n", _got200 ? "[OK]" : "[FAIL]");
printf("HTTPS: Received message:\n");
printf("%s\n", _buffer);
ret = true;
} while (0);
return ret;
}
protected:
MyTLSSocket * _tlssocket;
const char *_domain; /**< Domain name of the HTTPS server */
const uint16_t _port; /**< Port number of the HTTPS server */
char _buffer[HTTPS_USER_BUFFER_SIZE]; /**< User buffer */
};
#endif // End of AWS_IOT_HTTPS_TEST
int cal_init() {
LOG_INFO("cal_init\n");
/* The default 9600 bps is too slow to print full TLS debug info and could
* cause the other party to time out. */
printf("\nStarting AWS IoT test\n");
#if defined(MBED_MAJOR_VERSION)
printf("Using Mbed OS %d.%d.%d\n", MBED_MAJOR_VERSION, MBED_MINOR_VERSION, MBED_PATCH_VERSION);
#else
printf("Using Mbed OS from master.\n");
#endif
NetworkInterface *net = NetworkInterface::get_default_instance();
if (NULL == net) {
printf("Connecting to the network failed. See serial output.\n");
return 1;
}
nsapi_error_t status = net->connect();
if (status != NSAPI_ERROR_OK) {
printf("Connecting to the network failed %d!\n", status);
return -1;
}
printf("Connected to the network successfully. IP address: %s\n", net->get_ip_address());
#if AWS_IOT_MQTT_TEST
AWS_IoT_MQTT_Test *mqtt_test = new AWS_IoT_MQTT_Test(AWS_IOT_MQTT_SERVER_NAME, AWS_IOT_MQTT_SERVER_PORT, net);
mqtt_test->start_test();
delete mqtt_test;
#endif // End of AWS_IOT_MQTT_TEST
#if AWS_IOT_HTTPS_TEST
AWS_IoT_HTTPS_Test *https_test = new AWS_IoT_HTTPS_Test(AWS_IOT_HTTPS_SERVER_NAME, AWS_IOT_HTTPS_SERVER_PORT, net);
https_test->start_test();
delete https_test;
#endif // End of AWS_IOT_HTTPS_TEST
/* Some cellular modems e.g.: QUECTEL EC2X need graceful exit; otherwise, they will break in next reboot. */
status = net->disconnect();
if (status != NSAPI_ERROR_OK) {
printf("\n\nDisconnect from network interface failed %d\n", status);
}
return 1;
}
void cal_sendData(t_uint32 * t_data ,t_uint32 t_length, std::string t_dataType) {
LOG_INFO(“Data sending to the cloud”);
cal_init();
}