Minor fixes and polish (#31)

Fixes:
- on_connection_resumed callbacks had misnamed argument
- GG Discovery had python-2 incompatible str->bytes conversion

Polish:
- add kwargs to appropriate callbacks
- misc

Author: graebm

awsiot/__init__.py

@@ -142,7 +142,7 @@ def on_suback(suback_future):
                 except Exception as e:
                     future.set_exception(e)
 
-            def callback_wrapper(topic, payload):
+            def callback_wrapper(topic, payload, **kwargs):
                 try:
                     payload_obj = json.loads(payload.decode())
                     event = payload_to_class_fn(payload_obj)

awsiot/greengrass_discovery.py

@@ -21,10 +21,10 @@ class DiscoveryClient(object):
     __slots__ = ['_bootstrap', '_tls_context', '_socket_options', '_region', '_tls_connection_options', '_gg_server_name', 'gg_url', 'port']
 
     def __init__(self, bootstrap, socket_options, tls_context, region):
-        assert bootstrap is not None and isinstance(bootstrap, ClientBootstrap)
-        assert socket_options is not None and isinstance(socket_options, SocketOptions)
-        assert tls_context is not None and isinstance(tls_context, ClientTlsContext)
-        assert region is not None and isinstance(region, str)
+        assert isinstance(bootstrap, ClientBootstrap)
+        assert isinstance(socket_options, SocketOptions)
+        assert isinstance(tls_context, ClientTlsContext)
+        assert isinstance(region, str)
 
         self._bootstrap = bootstrap
         self._socket_options = socket_options
@@ -44,7 +44,7 @@ def discover(self, thing_name):
             future=Future(),
             response_body=bytearray())
 
-        def on_incoming_body(http_stream, chunk):
+        def on_incoming_body(http_stream, chunk, **kwargs):
             discovery['response_body'].extend(chunk)
 
         def on_request_complete(completion_future):

samples/basic_discovery.py

@@ -75,12 +75,12 @@
     exit(0)
 
 
-def on_connection_interupted(connection, error_code):
-    print('connection interupted with error {}'.format(error_code))
+def on_connection_interupted(connection, error, **kwargs):
+    print('connection interrupted with error {}'.format(error))
 
 
-def on_connection_resumed(connection, error_code, session_present):
-    print('connection resumed with error {}, session present {}'.format(error_code, session_present))
+def on_connection_resumed(connection, return_code, session_present, **kwargs):
+    print('connection resumed with return code {}, session present {}'.format(return_code, session_present))
 
 
 # Try IoT endpoints until we find one that works
@@ -90,12 +90,19 @@ def try_iot_endpoints():
             for connectivity_info in gg_core.connectivity:
                 try:
                     print('Trying core {} at host {} port {}'.format(gg_core.thing_arn, connectivity_info.host_address, connectivity_info.port))
-                    mqtt_connection = mqtt_connection_builder.mtls_from_path(endpoint=connectivity_info.host_address, port=connectivity_info.port,
-                        cert_filepath=args.certificate_path, pri_key_filepath=args.private_key_path, client_bootstrap=client_bootstrap, 
-                        ca_bytes=bytes(gg_group.certificate_authorities[0], encoding='utf-8'), 
-                        on_connection_interrupted=on_connection_interupted, on_connection_resumed=on_connection_resumed,
-                        client_id=args.thing_name, clean_session=False, keep_alive_secs=6)
-                    
+                    mqtt_connection = mqtt_connection_builder.mtls_from_path(
+                        endpoint=connectivity_info.host_address,
+                        port=connectivity_info.port,
+                        cert_filepath=args.certificate_path,
+                        pri_key_filepath=args.private_key_path,
+                        client_bootstrap=client_bootstrap,
+                        ca_bytes=gg_group.certificate_authorities[0].encode('utf-8'),
+                        on_connection_interrupted=on_connection_interupted,
+                        on_connection_resumed=on_connection_resumed,
+                        client_id=args.thing_name,
+                        clean_session=False,
+                        keep_alive_secs=6)
+
                     connect_future = mqtt_connection.connect()
                     connect_future.result()
                     print('Connected!')
@@ -111,7 +118,7 @@ def try_iot_endpoints():
 
 if args.mode == 'both' or args.mode == 'subscribe':
 
-    def on_publish(topic, payload):
+    def on_publish(topic, payload, **kwargs):
         print('Publish received on topic {}'.format(topic))
         print(payload)
 
@@ -125,8 +132,8 @@ def on_publish(topic, payload):
         message['message'] = args.message
         message['sequence'] = loop_count
         messageJson = json.dumps(message)
-        pub_future = mqtt_connection.publish(args.topic, messageJson, QoS.AT_MOST_ONCE)
-        pub_future[0].result()
+        pub_future, _ = mqtt_connection.publish(args.topic, messageJson, QoS.AT_MOST_ONCE)
+        pub_future.result()
         print('Published topic {}: {}\n'.format(args.topic, messageJson))
 
         loop_count += 1

samples/jobs.py

@@ -61,7 +61,7 @@
 parser.add_argument('--signing-region', default='us-east-1', help="If you specify --use-web-socket, this " +
     "is the region that will be used for computing the Sigv4 signature")
 parser.add_argument('--proxy-host', help="Hostname for proxy to connect to. Note: if you use this feature, " +
-    "you will likely need to set --root-ca to the ca for your proxy.")    
+    "you will likely need to set --root-ca to the ca for your proxy.")
 parser.add_argument('--proxy-port', type=int, default=8080, help="Port for proxy to connect to.")
 
 # Using globals to simplify sample code
@@ -244,18 +244,31 @@ def on_update_job_execution_rejected(rejected):
             proxy_options = http.HttpProxyOptions(host_name=args.proxy_host, port=args.proxy_port)
 
         credentials_provider = auth.AwsCredentialsProvider.new_default_chain(client_bootstrap)
-        mqtt_connection = mqtt_connection_builder.websockets_with_default_aws_signing(endpoint=args.endpoint,
-            client_bootstrap=client_bootstrap, region=args.signing_region, credentials_provider=credentials_provider, websocket_proxy_options=proxy_options,
-            ca_filepath=args.root_ca, client_id=args.client_id, clean_session=False, keep_alive_secs=6)
+        mqtt_connection = mqtt_connection_builder.websockets_with_default_aws_signing(
+            endpoint=args.endpoint,
+            client_bootstrap=client_bootstrap,
+            region=args.signing_region,
+            credentials_provider=credentials_provider,
+            websocket_proxy_options=proxy_options,
+            ca_filepath=args.root_ca,
+            client_id=args.client_id,
+            clean_session=False,
+            keep_alive_secs=6)
 
     else:
-        mqtt_connection = mqtt_connection_builder.mtls_from_path(endpoint=args.endpoint, cert_filepath=args.cert, pri_key_filepath=args.key,
-        client_bootstrap=client_bootstrap, ca_filepath=args.root_ca, client_id=args.client_id, 
-        clean_session=False, keep_alive_secs=6)  
+        mqtt_connection = mqtt_connection_builder.mtls_from_path(
+            endpoint=args.endpoint,
+            cert_filepath=args.cert,
+            pri_key_filepath=args.key,
+            client_bootstrap=client_bootstrap,
+            ca_filepath=args.root_ca,
+            client_id=args.client_id,
+            clean_session=False,
+            keep_alive_secs=6)
 
     print("Connecting to {} with client ID '{}'...".format(
         args.endpoint, args.client_id))
-    
+
     connected_future = mqtt_connection.connect()
 
     jobs_client = iotjobs.IotJobsClient(mqtt_connection)

samples/pubsub.py

@@ -16,6 +16,7 @@
 import argparse
 from awscrt import io, mqtt, auth, http
 from awsiot import mqtt_connection_builder
+import sys
 import threading
 import time
 
@@ -45,8 +46,8 @@
 parser.add_argument('--signing-region', default='us-east-1', help="If you specify --use-web-socket, this " +
     "is the region that will be used for computing the Sigv4 signature")
 parser.add_argument('--proxy-host', help="Hostname for proxy to connect to. Note: if you use this feature, " +
-    "you will likely need to set --root-ca to the ca for your proxy.")    
-parser.add_argument('--proxy-port', type=int, default=8080, help="Port for proxy to connect to.")                                                                
+    "you will likely need to set --root-ca to the ca for your proxy.")
+parser.add_argument('--proxy-port', type=int, default=8080, help="Port for proxy to connect to.")
 
 # Using globals to simplify sample code
 args = parser.parse_args()
@@ -55,15 +56,15 @@
 received_all_event = threading.Event()
 
 # Callback when connection is accidentally lost.
-def on_connection_interrupted(connection, error):
-    print("Connection interrupted. error_code: {}".format(error))
+def on_connection_interrupted(connection, error, **kwargs):
+    print("Connection interrupted. error: {}".format(error))
 
 
 # Callback when an interrupted connection is re-established.
-def on_connection_resumed(connection, error, session_present):
-    print("Connection resumed. error_code: {} session_present: {}".format(error, session_present))
+def on_connection_resumed(connection, return_code, session_present, **kwargs):
+    print("Connection resumed. return_code: {} session_present: {}".format(return_code, session_present))
 
-    if not error and not session_present:
+    if return_code == mqtt.ConnectReturnCode.ACCEPTED and not session_present:
         print("Session did not persist. Resubscribing to existing topics...")
         resubscribe_future, _ = connection.resubscribe_existing_topics()
 
@@ -78,11 +79,11 @@ def on_resubscribe_complete(resubscribe_future):
 
         for topic, qos in resubscribe_results['topics']:
             if qos is None:
-                exit("Server rejected resubscribe to topic: {}".format(topic))
+                sys.exit("Server rejected resubscribe to topic: {}".format(topic))
 
 
 # Callback when the subscribed topic receives a message
-def on_message_received(topic, payload):
+def on_message_received(topic, payload, **kwargs):
     print("Received message from topic '{}': {}".format(topic, payload))
     global received_count
     received_count += 1
@@ -101,15 +102,31 @@ def on_message_received(topic, payload):
             proxy_options = http.HttpProxyOptions(host_name=args.proxy_host, port=args.proxy_port)
 
         credentials_provider = auth.AwsCredentialsProvider.new_default_chain(client_bootstrap)
-        mqtt_connection = mqtt_connection_builder.websockets_with_default_aws_signing(endpoint=args.endpoint,
-            client_bootstrap=client_bootstrap, region=args.signing_region, credentials_provider=credentials_provider, websocket_proxy_options=proxy_options,
-            ca_filepath=args.root_ca, on_connection_interrupted=on_connection_interrupted, on_connection_resumed=on_connection_resumed,
-            client_id=args.client_id, clean_session=False, keep_alive_secs=6)
+        mqtt_connection = mqtt_connection_builder.websockets_with_default_aws_signing(
+            endpoint=args.endpoint,
+            client_bootstrap=client_bootstrap,
+            region=args.signing_region,
+            credentials_provider=credentials_provider,
+            websocket_proxy_options=proxy_options,
+            ca_filepath=args.root_ca,
+            on_connection_interrupted=on_connection_interrupted,
+            on_connection_resumed=on_connection_resumed,
+            client_id=args.client_id,
+            clean_session=False,
+            keep_alive_secs=6)
 
     else:
-        mqtt_connection = mqtt_connection_builder.mtls_from_path(endpoint=args.endpoint, cert_filepath=args.cert, pri_key_filepath=args.key,
-        client_bootstrap=client_bootstrap, ca_filepath=args.root_ca, on_connection_interrupted=on_connection_interrupted, on_connection_resumed=on_connection_resumed,
-            client_id=args.client_id, clean_session=False, keep_alive_secs=6)  
+        mqtt_connection = mqtt_connection_builder.mtls_from_path(
+            endpoint=args.endpoint,
+            cert_filepath=args.cert,
+            pri_key_filepath=args.key,
+            client_bootstrap=client_bootstrap,
+            ca_filepath=args.root_ca,
+            on_connection_interrupted=on_connection_interrupted,
+            on_connection_resumed=on_connection_resumed,
+            client_id=args.client_id,
+            clean_session=False,
+            keep_alive_secs=6)
 
     print("Connecting to {} with client ID '{}'...".format(
         args.endpoint, args.client_id))

samples/shadow.py

@@ -58,7 +58,7 @@
 parser.add_argument('--signing-region', default='us-east-1', help="If you specify --use-web-socket, this " +
     "is the region that will be used for computing the Sigv4 signature")
 parser.add_argument('--proxy-host', help="Hostname for proxy to connect to. Note: if you use this feature, " +
-    "you will likely need to set --root-ca to the ca for your proxy.")    
+    "you will likely need to set --root-ca to the ca for your proxy.")
 parser.add_argument('--proxy-port', type=int, default=8080, help="Port for proxy to connect to.")
 
 # Using globals to simplify sample code
@@ -249,13 +249,27 @@ def user_input_thread_fn():
             proxy_options = http.HttpProxyOptions(host_name=args.proxy_host, port=args.proxy_port)
 
         credentials_provider = auth.AwsCredentialsProvider.new_default_chain(client_bootstrap)
-        mqtt_connection = mqtt_connection_builder.websockets_with_default_aws_signing(endpoint=args.endpoint,
-            client_bootstrap=client_bootstrap, region=args.signing_region, credentials_provider=credentials_provider, websocket_proxy_options=proxy_options,
-            ca_filepath=args.root_ca, client_id=args.client_id, clean_session=False, keep_alive_secs=6)
+        mqtt_connection = mqtt_connection_builder.websockets_with_default_aws_signing(
+            endpoint=args.endpoint,
+            client_bootstrap=client_bootstrap,
+            region=args.signing_region,
+            credentials_provider=credentials_provider,
+            websocket_proxy_options=proxy_options,
+            ca_filepath=args.root_ca,
+            client_id=args.client_id,
+            clean_session=False,
+            keep_alive_secs=6)
 
     else:
-        mqtt_connection = mqtt_connection_builder.mtls_from_path(endpoint=args.endpoint, cert_filepath=args.cert, pri_key_filepath=args.key,
-        client_bootstrap=client_bootstrap, ca_filepath=args.root_ca, client_id=args.client_id, clean_session=False, keep_alive_secs=6)  
+        mqtt_connection = mqtt_connection_builder.mtls_from_path(
+            endpoint=args.endpoint,
+            cert_filepath=args.cert,
+            pri_key_filepath=args.key,
+            client_bootstrap=client_bootstrap,
+            ca_filepath=args.root_ca,
+            client_id=args.client_id,
+            clean_session=False,
+            keep_alive_secs=6)
 
     print("Connecting to {} with client ID '{}'...".format(
         args.endpoint, args.client_id))

setup.py

@@ -17,13 +17,13 @@
 
 setup(
     name='awsiotsdk',
-    version='1.0.2',
+    version='1.0.3',
     description='AWS IoT SDK based on the AWS Common Runtime',
     author='AWS SDK Common Runtime Team',
     url='https://github.com/aws/aws-iot-device-sdk-python-v2',
     packages = ['awsiot'],
     install_requires=[
-        'awscrt==0.5.2',
+        'awscrt==0.5.5',
         'futures;python_version<"3.2"',
         'typing;python_version<"3.5"',
     ],