Release of version 1.3.0

Author: liuszeng

AWSIoTPythonSDK/MQTTLib.py

@@ -378,8 +378,81 @@ def configureMQTTOperationTimeout(self, timeoutSecond):
         """
         self._mqtt_core.configure_operation_timeout_sec(timeoutSecond)
 
+    def configureUsernamePassword(self, username, password=None):
+        """
+        **Description**
+
+        Used to configure the username and password used in CONNECT packet.
+
+        **Syntax**
+
+        .. code:: python
+
+          # Configure user name and password
+          myAWSIoTMQTTClient.configureUsernamePassword("myUsername", "myPassword")
+
+        **Parameters**
+
+        *username* - Username used in the username field of CONNECT packet.
+
+        *password* - Password used in the password field of CONNECT packet.
+
+        **Returns**
+
+        None
+
+        """
+        self._mqtt_core.configure_username_password(username, password)
+
+    def enableMetricsCollection(self):
+        """
+        **Description**
+
+        Used to enable SDK metrics collection. Username field in CONNECT packet will be used to append the SDK name
+        and SDK version in use and communicate to AWS IoT cloud. This metrics collection is enabled by default.
+
+        **Syntax**
+
+        .. code:: python
+
+          myAWSIoTMQTTClient.enableMetricsCollection()
+
+        **Parameters**
+
+        None
+
+        **Returns**
+
+        None
+
+        """
+        self._mqtt_core.enable_metrics_collection()
+
+    def disableMetricsCollection(self):
+        """
+        **Description**
+
+        Used to disable SDK metrics collection.
+
+        **Syntax**
+
+        .. code:: python
+
+          myAWSIoTMQTTClient.disableMetricsCollection()
+
+        **Parameters**
+
+        None
+
+        **Returns**
+
+        None
+
+        """
+        self._mqtt_core.disable_metrics_collection()
+
     # MQTT functionality APIs
-    def connect(self, keepAliveIntervalSecond=30):
+    def connect(self, keepAliveIntervalSecond=600):
         """
         **Description**
 
@@ -389,15 +462,15 @@ def connect(self, keepAliveIntervalSecond=30):
 
         .. code:: python
 
-          # Connect to AWS IoT with default keepalive set to 30 seconds
+          # Connect to AWS IoT with default keepalive set to 600 seconds
           myAWSIoTMQTTClient.connect()
-          # Connect to AWS IoT with keepalive interval set to 55 seconds
-          myAWSIoTMQTTClient.connect(55)
+          # Connect to AWS IoT with keepalive interval set to 1200 seconds
+          myAWSIoTMQTTClient.connect(1200)
 
         **Parameters**
 
         *keepAliveIntervalSecond* - Time in seconds for interval of sending MQTT ping request. 
-        Default set to 30 seconds.
+        Default set to 600 seconds.
 
         **Returns**
 
@@ -407,7 +480,7 @@ def connect(self, keepAliveIntervalSecond=30):
         self._load_callbacks()
         return self._mqtt_core.connect(keepAliveIntervalSecond)
 
-    def connectAsync(self, keepAliveIntervalSecond=30, ackCallback=None):
+    def connectAsync(self, keepAliveIntervalSecond=600, ackCallback=None):
         """
         **Description**
 
@@ -417,15 +490,15 @@ def connectAsync(self, keepAliveIntervalSecond=30, ackCallback=None):
 
         .. code:: python
 
-          # Connect to AWS IoT with default keepalive set to 30 seconds and a custom CONNACK callback
+          # Connect to AWS IoT with default keepalive set to 600 seconds and a custom CONNACK callback
           myAWSIoTMQTTClient.connectAsync(ackCallback=my_connack_callback)
-          # Connect to AWS IoT with default keepalive set to 55 seconds and a custom CONNACK callback
-          myAWSIoTMQTTClient.connectAsync(keepAliveInternvalSecond=55, ackCallback=myConnackCallback)
+          # Connect to AWS IoT with default keepalive set to 1200 seconds and a custom CONNACK callback
+          myAWSIoTMQTTClient.connectAsync(keepAliveInternvalSecond=1200, ackCallback=myConnackCallback)
 
         **Parameters**
 
         *keepAliveIntervalSecond* - Time in seconds for interval of sending MQTT ping request.
-        Default set to 30 seconds.
+        Default set to 600 seconds.
 
         *ackCallback* - Callback to be invoked when the client receives a CONNACK. Should be in form
         :code:`customCallback(mid, data)`, where :code:`mid` is the packet id for the connect request
@@ -1036,8 +1109,81 @@ def configureMQTTOperationTimeout(self, timeoutSecond):
         # AWSIoTMQTTClient.configureMQTTOperationTimeout
         self._AWSIoTMQTTClient.configureMQTTOperationTimeout(timeoutSecond)
 
+    def configureUsernamePassword(self, username, password=None):
+        """
+        **Description**
+
+        Used to configure the username and password used in CONNECT packet.
+
+        **Syntax**
+
+        .. code:: python
+
+          # Configure user name and password
+          myAWSIoTMQTTShadowClient.configureUsernamePassword("myUsername", "myPassword")
+
+        **Parameters**
+
+        *username* - Username used in the username field of CONNECT packet.
+
+        *password* - Password used in the password field of CONNECT packet.
+
+        **Returns**
+
+        None
+
+        """
+        self._AWSIoTMQTTClient.configureUsernamePassword(username, password)
+
+    def enableMetricsCollection(self):
+        """
+        **Description**
+
+        Used to enable SDK metrics collection. Username field in CONNECT packet will be used to append the SDK name
+        and SDK version in use and communicate to AWS IoT cloud. This metrics collection is enabled by default.
+
+        **Syntax**
+
+        .. code:: python
+
+          myAWSIoTMQTTClient.enableMetricsCollection()
+
+        **Parameters**
+
+        None
+
+        **Returns**
+
+        None
+
+        """
+        self._AWSIoTMQTTClient.enableMetricsCollection()
+
+    def disableMetricsCollection(self):
+        """
+        **Description**
+
+        Used to disable SDK metrics collection.
+
+        **Syntax**
+
+        .. code:: python
+
+          myAWSIoTMQTTClient.disableMetricsCollection()
+
+        **Parameters**
+
+        None
+
+        **Returns**
+
+        None
+
+        """
+        self._AWSIoTMQTTClient.disableMetricsCollection()
+
     # Start the MQTT connection
-    def connect(self, keepAliveIntervalSecond=30):
+    def connect(self, keepAliveIntervalSecond=600):
         """
         **Description**
 
@@ -1047,10 +1193,10 @@ def connect(self, keepAliveIntervalSecond=30):
 
         .. code:: python
 
-          # Connect to AWS IoT with default keepalive set to 30 seconds
+          # Connect to AWS IoT with default keepalive set to 600 seconds
           myAWSIoTMQTTShadowClient.connect()
-          # Connect to AWS IoT with keepalive interval set to 55 seconds
-          myAWSIoTMQTTShadowClient.connect(55)
+          # Connect to AWS IoT with keepalive interval set to 1200 seconds
+          myAWSIoTMQTTShadowClient.connect(1200)
 
         **Parameters**
 

AWSIoTPythonSDK/__init__.py

@@ -1,3 +1,3 @@
-__version__ = "1.2.0"
+__version__ = "1.3.0"
 
 

AWSIoTPythonSDK/core/greengrass/discovery/models.py

@@ -460,7 +460,7 @@ def _decode_core_info(self, core_object, group_id):
             connectivity_info = ConnectivityInfo(connectivity_info_object[KEY_CONNECTIVITY_INFO_ID],
                                                  connectivity_info_object[KEY_HOST_ADDRESS],
                                                  connectivity_info_object[KEY_PORT_NUMBER],
-                                                 connectivity_info_object[KEY_METADATA])
+                                                 connectivity_info_object.get(KEY_METADATA,''))
             core_info.appendConnectivityInfo(connectivity_info)
 
         return core_info

AWSIoTPythonSDK/core/protocol/connection/cores.py

@@ -32,6 +32,7 @@
 import hmac
 from AWSIoTPythonSDK.exception.AWSIoTExceptions import wssNoKeyInEnvironmentError
 from AWSIoTPythonSDK.exception.AWSIoTExceptions import wssHandShakeError
+from AWSIoTPythonSDK.core.protocol.internal.defaults import DEFAULT_CONNECT_DISCONNECT_TIMEOUT_SEC
 try:
     from urllib.parse import quote  # Python 3+
 except ImportError:
@@ -378,6 +379,8 @@ class SecuredWebSocketCore:
     _WebsocketConnectInit = -1
     _WebsocketDisconnected = 1
 
+    _logger = logging.getLogger(__name__)
+
     def __init__(self, socket, hostAddress, portNumber, AWSAccessKeyID="", AWSSecretAccessKey="", AWSSessionToken=""):
         self._connectStatus = self._WebsocketConnectInit
         # Handlers
@@ -481,21 +484,27 @@ def _handShake(self, hostAddress, portNumber):
         handshakeBytes = handshakeBytes.encode('utf-8')
         self._sslSocket.write(handshakeBytes)
         # Read it back (Non-blocking socket)
-        # Do we need a timeout here?
+        timeStart = time.time()
         wssHandshakeResponse = bytearray()
         while len(wssHandshakeResponse) == 0:
             try:
                 wssHandshakeResponse += self._sslSocket.read(1024)  # Response is always less than 1024 bytes
             except socket.error as err:
                 if err.errno == ssl.SSL_ERROR_WANT_READ or err.errno == ssl.SSL_ERROR_WANT_WRITE:
-                    pass
+                    if time.time() - timeStart > self._getTimeoutSec():
+                        raise err  # We make sure that reconnect gets retried in Paho upon a wss reconnect response timeout
+                else:
+                    raise err
         # Verify response
         # Now both wssHandshakeResponse and rawSecWebSocketKey are byte strings
         if not self._verifyWSSResponse(wssHandshakeResponse, rawSecWebSocketKey):
             raise wssHandShakeError()
         else:
             pass
 
+    def _getTimeoutSec(self):
+        return DEFAULT_CONNECT_DISCONNECT_TIMEOUT_SEC
+
     # Used to create a single wss frame
     # Assume that the maximum length of a MQTT packet never exceeds the maximum length
     # for a wss frame. Therefore, the FIN bit for the encoded frame will always be 1.

AWSIoTPythonSDK/core/protocol/internal/clients.py

@@ -97,6 +97,9 @@ def configure_last_will(self, topic, payload, qos, retain=False):
     def clear_last_will(self):
         self._paho_client.will_clear()
 
+    def set_username_password(self, username, password=None):
+        self._paho_client.username_pw_set(username, password)
+
     def configure_reconnect_back_off(self, base_reconnect_quiet_sec, max_reconnect_quiet_sec, stable_connection_sec):
         self._paho_client.setBackoffTiming(base_reconnect_quiet_sec, max_reconnect_quiet_sec, stable_connection_sec)
 

AWSIoTPythonSDK/core/protocol/internal/defaults.py

@@ -15,4 +15,5 @@
 
 DEFAULT_CONNECT_DISCONNECT_TIMEOUT_SEC = 30
 DEFAULT_OPERATION_TIMEOUT_SEC = 5
-DEFAULT_DRAINING_INTERNAL_SEC = 0.5
+DEFAULT_DRAINING_INTERNAL_SEC = 0.5
+METRICS_PREFIX = "?SDK=Python&Version="

AWSIoTPythonSDK/core/protocol/mqtt_core.py

@@ -13,6 +13,7 @@
 # * permissions and limitations under the License.
 # */
 
+import AWSIoTPythonSDK
 from AWSIoTPythonSDK.core.protocol.internal.clients import InternalAsyncMqttClient
 from AWSIoTPythonSDK.core.protocol.internal.clients import ClientStatusContainer
 from AWSIoTPythonSDK.core.protocol.internal.clients import ClientStatus
@@ -24,6 +25,7 @@
 from AWSIoTPythonSDK.core.protocol.internal.requests import QueueableRequest
 from AWSIoTPythonSDK.core.protocol.internal.defaults import DEFAULT_CONNECT_DISCONNECT_TIMEOUT_SEC
 from AWSIoTPythonSDK.core.protocol.internal.defaults import DEFAULT_OPERATION_TIMEOUT_SEC
+from AWSIoTPythonSDK.core.protocol.internal.defaults import METRICS_PREFIX
 from AWSIoTPythonSDK.core.protocol.internal.events import FixedEventMids
 from AWSIoTPythonSDK.core.protocol.paho.client import MQTT_ERR_SUCCESS
 from AWSIoTPythonSDK.exception.AWSIoTExceptions import connectError
@@ -60,6 +62,9 @@ class MqttCore(object):
     _logger = logging.getLogger(__name__)
 
     def __init__(self, client_id, clean_session, protocol, use_wss):
+        self._username = ""
+        self._password = None
+        self._enable_metrics_collection = True
         self._event_queue = Queue()
         self._event_cv = Condition()
         self._event_producer = EventProducer(self._event_cv, self._event_queue)
@@ -77,7 +82,6 @@ def __init__(self, client_id, clean_session, protocol, use_wss):
         self._operation_timeout_sec = DEFAULT_OPERATION_TIMEOUT_SEC
         self._init_offline_request_exceptions()
         self._init_workers()
-        self._start_workers()
         self._logger.info("MqttCore initialized")
         self._logger.info("Client id: %s" % client_id)
         self._logger.info("Protocol version: %s" % ("MQTTv3.1" if protocol == MQTTv31 else "MQTTv3.1.1"))
@@ -153,6 +157,17 @@ def clear_last_will(self):
         self._logger.info("Clearing last will...")
         self._internal_async_client.clear_last_will()
 
+    def configure_username_password(self, username, password=None):
+        self._logger.info("Configuring username and password...")
+        self._username = username
+        self._password = password
+
+    def enable_metrics_collection(self):
+        self._enable_metrics_collection = True
+
+    def disable_metrics_collection(self):
+        self._enable_metrics_collection = False
+
     def configure_offline_requests_queue(self, max_size, drop_behavior):
         self._logger.info("Configuring offline requests queueing: max queue size: %d", max_size)
         self._offline_requests_manager = OfflineRequestsManager(max_size, drop_behavior)
@@ -174,7 +189,9 @@ def connect(self, keep_alive_sec):
     def connect_async(self, keep_alive_sec, ack_callback=None):
         self._logger.info("Performing async connect...")
         self._logger.info("Keep-alive: %f sec" % keep_alive_sec)
+        self._start_workers()
         self._load_callbacks()
+        self._load_username_password()
         self._client_status.set_status(ClientStatus.CONNECT)
         rc = self._internal_async_client.connect(keep_alive_sec, ack_callback)
         if MQTT_ERR_SUCCESS != rc:
@@ -188,6 +205,13 @@ def _load_callbacks(self):
         self._internal_async_client.on_offline = self.on_offline
         self._internal_async_client.on_message = self.on_message
 
+    def _load_username_password(self):
+        username_candidate = self._username
+        if self._enable_metrics_collection:
+            username_candidate += METRICS_PREFIX
+            username_candidate += AWSIoTPythonSDK.__version__
+        self._internal_async_client.set_username_password(username_candidate, self._password)
+
     def disconnect(self):
         self._logger.info("Performing sync disconnect...")
         event = Event()

AWSIoTPythonSDK/core/shadow/deviceShadow.py

@@ -178,6 +178,10 @@ def _parseTopicShadowName(self, srcTopic):
 
     def _timerHandler(self, srcActionName, srcToken):
         with self._dataStructureLock:
+            # Don't crash if we try to remove an unknown token
+            if srcToken not in self._tokenPool:
+                self._logger.warn('Tried to remove non-existent token from pool: %s' % str(srcToken))
+                return
             # Remove the token
             del self._tokenPool[srcToken]
             # Need to unsubscribe?