Adapt to awscrt.io and awscrt.mqtt API changes

Author: graebm

samples/basic_discovery.py

@@ -50,7 +50,7 @@
     io.init_logging(LogLevel.Info, 'stderr')
 elif args.verbosity.lower() == 'debug':
     io.init_logging(LogLevel.Debug, 'stderr')
-elif args.verbosity.lower() == 'trace':    
+elif args.verbosity.lower() == 'trace':
     io.init_logging(LogLevel.Trace, 'stderr')
 
 event_loop_group = io.EventLoopGroup(1)
@@ -77,26 +77,26 @@
 mqtt_client = Client(client_bootstrap, gg_core_tls_ctx)
 
 
-def on_connection_interupted(error_code):
+def on_connection_interupted(connection, error_code):
     print('connection interupted with error {}'.format(error_code))
 
 
-def on_connection_resumed(error_code, session_present):
+def on_connection_resumed(connection, error_code, session_present):
     print('connection resumed with error {}, session present {}'.format(error_code, session_present))
 
 
 mqtt_connection = Connection(mqtt_client, on_connection_interrupted=on_connection_interupted, on_connection_resumed=on_connection_resumed)
 
 connection_succeeded = False
-for conectivity_info in resp.gg_groups[0].cores[0].connectivity:    
+for conectivity_info in resp.gg_groups[0].cores[0].connectivity:
     try:
         connect_future = mqtt_connection.connect(args.thing_name, resp.gg_groups[0].cores[0].connectivity[0].host_address, resp.gg_groups[0].cores[0].connectivity[0].port, clean_session=False)
         connect_future.result()
         connection_succeeded = True
         break
     except Exception as e:
         print('connection failed with exception {}'.format(e))
-        continue  
+        continue
 
 if connection_succeeded != True:
     print('All connection attempts for core {} failed'.format(resp.gg_groups[0].cores[0].thing_arn))
@@ -121,6 +121,6 @@ def on_publish(topic, message):
         pub_future = mqtt_connection.publish(args.topic, messageJson, QoS.AT_MOST_ONCE)
         pub_future[0].result()
         print('Published topic {}: {}\n'.format(args.topic, messageJson))
-       
+
         loop_count += 1
     time.sleep(1)

samples/jobs.py

@@ -230,7 +230,7 @@ def on_update_job_execution_rejected(rejected):
 
     tls_options = io.TlsContextOptions.create_client_with_mtls_from_path(args.cert, args.key)
     if args.root_ca:
-        tls_options.override_default_trust_store_from_path(ca_path=None, ca_file=args.root_ca)
+        tls_options.override_default_trust_store_from_path(ca_dirpath=None, ca_filepath=args.root_ca)
     tls_context = io.ClientTlsContext(tls_options)
 
     mqtt_client = mqtt.Client(client_bootstrap, tls_context)

samples/pubsub.py

@@ -46,29 +46,29 @@
 
 # Callback when connection is accidentally lost.
 def on_connection_interrupted(connection, error_code):
-    print("Connection interrupted. error_code:{}".format(error_code))
+    print("Connection interrupted. error_code: {}".format(error_code))
+
 
 # Callback when an interrupted connection is re-established.
 def on_connection_resumed(connection, error_code, session_present):
-    print("Connection resumed. error_code:{} session_present:{}".format(error_code, session_present))
+    print("Connection resumed. error_code: {} session_present: {}".format(error_code, session_present))
 
-    if not session_present:
-        print("Server did not save state. Resubscribing to existing topics...")
+    if not error_code and not session_present:
+        print("Session did not persist. Resubscribing to existing topics...")
         resubscribe_future, _ = connection.resubscribe_existing_topics()
 
         # Cannot synchronously wait for resubscribe result because we're on the connection's event-loop thread,
         # evaluate result with a callback instead.
-        def on_resubscribed(resubscribe_future):
-            try:
-                resubscribe_results = resubscribe_future.result()
-                print("Resubscribe complete. results:{}".format(resubscribe_results))
-                for topic, qos in resubscribe_results['topics']:
-                    assert(qos)
-            except Exception as e:
-                print("Resubscribe failed. Exiting", e)
-                exit(-1)
+        resubscribe_future.add_done_callback(on_resubscribe_complete)
+
+
+def on_resubscribe_complete(resubscribe_future):
+        resubscribe_results = resubscribe_future.result()
+        print("Resubscribe results: {}".format(resubscribe_results))
 
-        resubscribe_future.add_done_callback(on_resubscribed)
+        for topic, qos in resubscribe_results['topics']:
+            if qos is None:
+                exit("Server rejected resubscribe to topic: {}".format(topic))
 
 
 # Callback when the subscribed topic receives a message
@@ -121,7 +121,9 @@ def on_message_received(topic, message):
         qos=mqtt.QoS.AT_LEAST_ONCE,
         callback=on_message_received)
 
-    subscribe_future.result()
+    subscribe_result = subscribe_future.result()
+    if subscribe_result['qos'] is None:
+        raise RuntimeError("Server rejected subscription")
     print("Subscribed!")
 
     # Publish message to server desired number of times.

samples/shadow.py

@@ -235,7 +235,7 @@ def user_input_thread_fn():
 
     tls_options = io.TlsContextOptions.create_client_with_mtls_from_path(args.cert, args.key)
     if args.root_ca:
-        tls_options.override_default_trust_store_from_path(ca_path=None, ca_file=args.root_ca)
+        tls_options.override_default_trust_store_from_path(ca_dirpath=None, ca_filepath=args.root_ca)
     tls_context = io.ClientTlsContext(tls_options)
 
     mqtt_client = mqtt.Client(client_bootstrap, tls_context)