Release of version 1.1.0

Author: liuszeng

AWS-IoT-Arduino-Yun-Library/aws_iot_config_SDK.h

@@ -18,6 +18,6 @@
 
 #define MAX_BUF_SIZE 256										// maximum number of bytes to publish/receive
 #define MAX_SUB 15 												// maximum number of subscribe
-#define CMD_TIME_OUT 100										// maximum time to wait for feedback from AR9331, 100 = 10 sec
+#define CMD_TIME_OUT 200										// maximum time to wait for feedback from AR9331, 200 = 10 sec
 
 #endif

AWS-IoT-Arduino-Yun-Library/aws_iot_error.h

@@ -52,7 +52,8 @@ typedef enum {
 	SHADOW_DELETE_GENERIC_ERROR = -32,
 	SHADOW_REGISTER_DELTA_CALLBACK_GENERIC_ERROR = -33,
 	SHADOW_UNREGISTER_DELTA_CALLBACK_GENERIC_ERROR = -34,
-	YIELD_ERROR = -35
+	YIELD_ERROR = -35,
+	WEBSOCKET_CREDENTIAL_NOT_FOUND = -36
 } IoT_Error_t;
 
 #endif

AWS-IoT-Arduino-Yun-Library/aws_iot_mqtt.cpp

@@ -47,14 +47,14 @@ Baud_t aws_iot_mqtt_client::find_baud_type() {
 	return rc_type;
 }
 
-IoT_Error_t aws_iot_mqtt_client::setup_exec(char* client_id, bool clean_session, MQTTv_t MQTT_version) {
+IoT_Error_t aws_iot_mqtt_client::setup_exec(char* client_id, bool clean_session, MQTTv_t MQTT_version, bool useWebsocket) {
 	// Serial1 is started before this call
 	IoT_Error_t rc = NONE_ERROR;
 	exec_cmd("cd /root/AWS-IoT-Python-Runtime/runtime/\n", false, false);
 	exec_cmd("python run.py\n", false, false);
 
 	// Create obj
-	exec_cmd("4\n", false, false);
+	exec_cmd("5\n", false, false);
 
 	exec_cmd("i\n", false, false);
 
@@ -66,6 +66,11 @@ IoT_Error_t aws_iot_mqtt_client::setup_exec(char* client_id, bool clean_session,
 	exec_cmd(rw_buf, false, false);
 
 	sprintf(rw_buf, "%u\n", MQTT_version);
+	exec_cmd(rw_buf, false, false);
+
+	// Websocket flag
+	num_temp = useWebsocket ? 1 : 0;
+	sprintf(rw_buf, "%d\n", num_temp);
 	exec_cmd(rw_buf, true, false);
 
 	if(strncmp(rw_buf, "I T", 3) != 0) {
@@ -76,7 +81,7 @@ IoT_Error_t aws_iot_mqtt_client::setup_exec(char* client_id, bool clean_session,
 	return rc;
 }
 
-IoT_Error_t aws_iot_mqtt_client::setup(char* client_id, bool clean_session, MQTTv_t MQTT_version) {
+IoT_Error_t aws_iot_mqtt_client::setup(char* client_id, bool clean_session, MQTTv_t MQTT_version, bool useWebsocket) {
 	IoT_Error_t rc = NONE_ERROR;
 	if(client_id == NULL) {rc = NULL_VALUE_ERROR;}
 	else if(strlen(client_id) >= MAX_BUF_SIZE) {rc = OVERFLOW_ERROR;}
@@ -86,13 +91,17 @@ IoT_Error_t aws_iot_mqtt_client::setup(char* client_id, bool clean_session, MQTT
 		// Communication failed due to baud rate issue
 		if(BAUD_TYPE_UNKNOWN == baud_type) {rc = SERIAL1_COMMUNICATION_ERROR;}
 		else {
-			rc = setup_exec(client_id, clean_session, MQTT_version);
+			rc = setup_exec(client_id, clean_session, MQTT_version, useWebsocket);
 		}
 	}
 
 	return rc;
 }
 
+IoT_Error_t aws_iot_mqtt_client::configWss(char* host, int port, char* cafile_path) {
+	return config(host, port, cafile_path, "", ""); // No need for key and cert, IAM credentials are used.
+}
+
 IoT_Error_t aws_iot_mqtt_client::config(char* host, int port, char* cafile_path, char* keyfile_path, char* certfile_path) {
 	IoT_Error_t rc = NONE_ERROR;
 
@@ -154,6 +163,7 @@ IoT_Error_t aws_iot_mqtt_client::connect(int keepalive_interval) {
 		else if(strncmp(rw_buf, "C4F", 3) == 0) {rc = CONNECT_ERROR;}
 		else if(strncmp(rw_buf, "C5F", 3) == 0) {rc = CONNECT_TIMEOUT;}
 		else if(strncmp(rw_buf, "C6F", 3) == 0) {rc = CONNECT_CREDENTIAL_NOT_FOUND;}
+		else if(strncmp(rw_buf, "C7F", 3) == 0) {rc = WEBSOCKET_CREDENTIAL_NOT_FOUND;}
 		else if(strncmp(rw_buf, "CFF", 3) == 0) {rc = CONNECT_GENERIC_ERROR;}
 		else rc = GENERIC_ERROR;
 	}
@@ -609,10 +619,10 @@ void aws_iot_mqtt_client::exec_cmd(const char* cmd, bool wait, bool single_line)
 		timeout_flag = false;
 		timeout_sec = 0;
 		// step2: waiting
-		delay(10);
+		delay(6);
 		if(wait) {
 			while(timeout_sec < CMD_TIME_OUT && !Serial1.available()) {
-				delay(100); // 0.1 sec
+				delay(50); // 50 ms
 				timeout_sec++;
 			}
 		}
@@ -637,7 +647,6 @@ void aws_iot_mqtt_client::exec_cmd(const char* cmd, bool wait, bool single_line)
 	}
 	timeout_flag = false; // Clear timeout flag
 	rw_buf[ptr] = '\0'; // add terminator in case of garbage data in rw_buf
-	// printf("%s\n------\n", rw_buf);
 }
 
 int aws_iot_mqtt_client::find_unused_subgroup() {

AWS-IoT-Arduino-Yun-Library/aws_iot_mqtt.h

@@ -47,8 +47,9 @@ class aws_iot_mqtt_client {
 				sub_group[i].callback = NULL;
 			}
 		}
-		IoT_Error_t setup(char* client_id, bool clean_session=true, MQTTv_t MQTT_version=MQTTv311);
+		IoT_Error_t setup(char* client_id, bool clean_session=true, MQTTv_t MQTT_version=MQTTv311, bool useWebsocket=false);
 		IoT_Error_t config(char* host, int port, char* cafile_path, char* keyfile_path, char* certfile_path);
+		IoT_Error_t configWss(char* host, int port, char* cafile_path);
 		IoT_Error_t connect(int keepalive_interval=60);
 		IoT_Error_t publish(char* topic, char* payload, int payload_len, int qos, bool retain);
 		IoT_Error_t subscribe(char* topic, int qos, message_callback cb);
@@ -74,7 +75,7 @@ class aws_iot_mqtt_client {
 		bool timeout_flag; // Is there a timeout when executing RPC
 		IoT_Error_t base_subscribe(char* topic, int qos, message_callback cb, int is_delta);
 		Baud_t find_baud_type();
-		IoT_Error_t setup_exec(char* client_id, bool clean_session, MQTTv_t MQTT_version);
+		IoT_Error_t setup_exec(char* client_id, bool clean_session, MQTTv_t MQTT_version, bool useWebsocket);
 		void exec_cmd(const char* cmd, bool wait, bool single_line);
 		int find_unused_subgroup();
 		void clearProtocolOnSerialBegin(long baudrate);

AWS-IoT-Arduino-Yun-Library/aws_iot_version.h

@@ -17,8 +17,8 @@
 #define version_h
 
 #define VERSION_MAJOR 1
-#define VERSION_MINOR 0
-#define VERSION_PATCH 3
+#define VERSION_MINOR 1
+#define VERSION_PATCH 0
 #define VERSION_TAG ""
 
 #endif

AWS-IoT-Arduino-Yun-Library/keywords.txt

@@ -4,6 +4,7 @@ aws_iot_version		KEYWORD1
 aws_iot_config		KEYWORD1
 setup			KEYWORD2
 config			KEYWORD2
+configWss		KEYWORD2
 connect			KEYWORD2
 publish			KEYWORD2
 subscribe		KEYWORD2

AWS-IoT-Python-Runtime/certs/delete.me

@@ -0,0 +1 @@
+Please place your credentials in this directory.

AWS-IoT-Python-Runtime/lib/command/commandConnect.py

@@ -55,6 +55,8 @@ def execute(self):
                 returnMessage = "C5F: " + str(e.message)
             except IOError as e:
                 returnMessage = "C6F: " + "Credentials not found."
+            except ValueError as e:
+                returnMessage = "C7F: " + "Key/KeyID not in $ENV."
             except Exception as e:
                 returnMessage = "CFF: " + "Unknown error."
         self._serialCommServerHandler.writeToInternalProtocol(returnMessage)

AWS-IoT-Python-Runtime/lib/exception/AWSIoTExceptions.py

@@ -75,3 +75,14 @@ def __init__(self, errorCode):
 class unsubscribeError(operationError.operationError):
     def __init__(self, errorCode):
         self.message = "Unsubscribe Error: " + str(errorCode)
+
+
+# Websocket Error
+class wssNoKeyInEnvironmentError(operationError.operationError):
+    def __init__(self):
+        self.message = "No AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY detected in $ENV."
+
+
+class wssHandShakeError(operationError.operationError):
+    def __init__(self):
+        self.message = "Error in WSS handshake."

AWS-IoT-Python-Runtime/lib/protocol/mqttCore.py

@@ -20,7 +20,8 @@
 sys.path.append("../lib/")
 import ssl
 import time
-import paho.mqtt.client as mqtt
+import thread
+import protocol.paho.client as mqtt
 from threading import Lock
 from exception.AWSIoTExceptions import connectError
 from exception.AWSIoTExceptions import connectTimeoutException
@@ -45,6 +46,10 @@ class mqttCore:
     _unsubscribeSent = False
     _connectdisconnectTimeout = 0  # Default connect/disconnect timeout set to 0 second
     _mqttOperationTimeout = 0  # Default MQTT operation timeout set to 0 second
+    # Use websocket
+    _useWebsocket = False
+    # Subscribe record
+    _subscribePool = dict()
     # Broker information
     _host = ""
     _port = -1
@@ -76,13 +81,25 @@ def getMQTTOperationTimeoutSecond(self):
     def setUserData(self, srcUserData):
         self._pahoClient.user_data_set(srcUserData)
 
-    def createPahoClient(self, clientID, cleanSession, userdata, protocol):
-        return mqtt.Client(clientID, cleanSession, userdata, protocol)  # Throw exception when error happens
+    def createPahoClient(self, clientID, cleanSession, userdata, protocol, useWebsocket):
+        return mqtt.Client(clientID, cleanSession, userdata, protocol, useWebsocket)  # Throw exception when error happens
+
+    def _doResubscribe(self):
+        if self._connectResultCode == 0:  # If this is a successful connect...
+            for key in self._subscribePool.keys():
+                qos, callback = self._subscribePool.get(key)
+                try:
+                    self.subscribe(key, qos, callback)
+                except subscribeError:
+                    pass
+                except subscribeTimeoutException:
+                    pass  # Subscribe error resulted from network error, will redo subscription in the next re-connect
 
     # Callbacks
     def on_connect(self, client, userdata, flags, rc):
         self._disconnectResultCode = sys.maxint
         self._connectResultCode = rc
+        thread.start_new_thread(self._doResubscribe, ())
         self._log.writeLog("Connect result code " + str(rc))
 
     def on_disconnect(self, client, userdata, rc):
@@ -103,18 +120,19 @@ def on_message(self, client, userdata, message):
         self._log.writeLog("Received (No custom callback registered) : message: " + str(message.payload) + " from topic: " + str(message.topic))
 
     ####### API starts here #######
-    def __init__(self, clientID, cleanSession, protocol, srcLogManager):
+    def __init__(self, clientID, cleanSession, protocol, srcLogManager, srcUseWebsocket=False):
         if clientID is None or cleanSession is None or protocol is None or srcLogManager is None:
             raise TypeError("None type inputs detected.")
         self._log = srcLogManager
         self._clientID = clientID
-        self._pahoClient = self.createPahoClient(clientID, cleanSession, None, protocol)  # User data is set to None as default
+        self._pahoClient = self.createPahoClient(clientID, cleanSession, None, protocol, srcUseWebsocket)  # User data is set to None as default
         self._log.writeLog("Paho MQTT Client init.")
         self._pahoClient.on_connect = self.on_connect
         self._pahoClient.on_disconnect = self.on_disconnect
         self._pahoClient.on_message = self.on_message
         self._pahoClient.on_subscribe = self.on_subscribe
         self._pahoClient.on_unsubscribe = self.on_unsubscribe
+        self._useWebsocket = srcUseWebsocket
         self._log.writeLog("Register Paho MQTT Client callbacks.")
         self._log.writeLog("mqttCore init.")
 
@@ -135,7 +153,10 @@ def connect(self, keepAliveInterval=60):
         # Return connect succeeded/failed
         ret = False
         # TLS configuration
-        self._pahoClient.tls_set(self._cafile, self._cert, self._key, ssl.CERT_REQUIRED, ssl.PROTOCOL_SSLv23)  # Throw exception...
+        if self._useWebsocket:
+            self._pahoClient.tls_set(ca_certs=self._cafile, cert_reqs=ssl.CERT_REQUIRED, tls_version=ssl.PROTOCOL_SSLv23)
+        else:
+            self._pahoClient.tls_set(self._cafile, self._cert, self._key, ssl.CERT_REQUIRED, ssl.PROTOCOL_SSLv23)  # Throw exception...
         # Connect
         self._pahoClient.connect(self._host, self._port, keepAliveInterval)  # Throw exception...
         self._pahoClient.loop_start()
@@ -215,6 +236,7 @@ def subscribe(self, topic, qos, callback):
         if(self._subscribeSent):
             ret = rc == 0
             if(ret):
+                self._subscribePool[topic] = (qos, callback)
                 self._log.writeLog("Subscribe request " + str(mid) + " succeeded. Time consumption: " + str(float(TenmsCount) * 10) + "ms.")
             else:
                 if(callback is not None):
@@ -252,6 +274,10 @@ def unsubscribe(self, topic):
         if(self._unsubscribeSent):
             ret = rc == 0
             if(ret):
+                try:
+                    del self._subscribePool[topic]
+                except KeyError:
+                    pass  # Ignore topic that is never subscribed to
                 self._log.writeLog("Unsubscribe request " + str(mid) + " succeeded. Time consumption: " + str(float(TenmsCount) * 10) + "ms.")
                 self._pahoClient.message_callback_remove(topic)
                 self._log.writeLog("Remove the callback.")