Fix greengrass and discovery sample

Author: graebm

awsiot/greengrass_discovery.py

@@ -12,8 +12,7 @@
 # permissions and limitations under the License.
 
 from awscrt.http import HttpClientConnection, HttpRequest
-from awscrt import io
-from awscrt.io import ClientBootstrap, ClientTlsContext, TlsConnectionOptions, SocketOptions
+from awscrt.io import ClientBootstrap, ClientTlsContext, is_alpn_available, SocketOptions, TlsConnectionOptions
 import awsiot
 from concurrent.futures import Future
 import json
@@ -35,58 +34,59 @@ def __init__(self, bootstrap, socket_options, tls_context, region):
         self._tls_connection_options.set_server_name(self._gg_server_name)
         self.port = 8443
 
-        if io.is_alpn_available():
-            self._tls_connection_options.set_alpn_list('x-amzn-http-ca')
+        if is_alpn_available():
+            self._tls_connection_options.set_alpn_list(['x-amzn-http-ca'])
             self.port = 443
 
     def discover(self, thing_name):
-        ret_future = Future()
-        response_body = bytearray()
-        request = None
 
-        def on_incoming_body(response_chunk):
-            response_body.extend(response_chunk)
+        discovery = dict(
+            future=Future(),
+            response_body=bytearray())
+
+        def on_incoming_body(http_stream, response_chunk):
+            discovery['response_body'].extend(response_chunk)
 
         def on_request_complete(completion_future):
-            global request  
             try:
-                response_code = request.response_code
-                # marking request as global prevents the GC from reclaiming it,
-                # so force it to do it here.
-                request = None
+                response_code = completion_future.result()
                 if response_code == 200:
-                    payload_str = response_body.decode('utf-8')
+                    payload_str = discovery['response_body'].decode('utf-8')
                     discover_res = DiscoverResponse.from_payload(json.loads(payload_str))
-                    ret_future.set_result(discover_res)                    
-                else:                    
-                    ret_future.set_exception(DiscoveryException('Error during discover call: response code ={}'.format(response_code), response_code))
+                    discovery['future'].set_result(discover_res)
+                else:
+                    discovery['future'].set_exception(DiscoveryException('Error during discover call: response_code={}'.format(response_code), response_code))
 
             except Exception as e:
-                ret_future.set_exception(e)
+                discovery['future'].set_exception(e)
 
         def on_connection_completed(conn_future):
-            global request
             try:
-                connection = conn_future.result()                
-                request = connection.make_request(
-                    method='GET', 
-                    uri_str='/greengrass/discover/thing/{}'.format(thing_name), 
-                    outgoing_headers={'host':self._gg_server_name}, 
-                    on_outgoing_body=None, 
-                    on_incoming_body=on_incoming_body)                
+                connection = conn_future.result()
+                request = HttpRequest(
+                    method='GET',
+                    path='/greengrass/discover/thing/{}'.format(thing_name),
+                    headers=[('host', self._gg_server_name)])
+
+                http_stream = connection.request(
+                    request=request,
+                    on_body=on_incoming_body)
 
-                request.response_completed.add_done_callback(on_request_complete)
+                http_stream.completion_future.add_done_callback(on_request_complete)
 
             except Exception as e:
-                # marking request as global prevents the GC from reclaiming it,
-                # so force it to do it here.
-                request = None  
-                ret_future.set_exception(e)
+                discovery['future'].set_exception(e)
+
+        connect_future = HttpClientConnection.new(
+            host_name=self._gg_server_name,
+            port=self.port,
+            socket_options=self._socket_options,
+            tls_connection_options = self._tls_connection_options,
+            bootstrap = self._bootstrap)
 
-        connect_future = HttpClientConnection.new_connection(self._bootstrap, self._gg_server_name, self.port, self._socket_options, None, self._tls_connection_options)
         connect_future.add_done_callback(on_connection_completed)
-        
-        return ret_future
+
+        return discovery['future']
 
 class DiscoveryException(Exception):
     _slots_ = ['http_response_code', 'message']
@@ -102,7 +102,7 @@ class ConnectivityInfo(awsiot.ModeledClass):
     def ___init___(self):
         for slot in self.__slots__:
             setattr(self, slot, None)
-        
+
     @classmethod
     def from_payload(cls, payload):
         # type: (typing.Dict[str, typing.Any]) -> ConnectivityInfo
@@ -138,12 +138,12 @@ def from_payload(cls, payload):
         val = payload.get('Connectivity')
         if val is not None:
             new.connectivity = [ConnectivityInfo.from_payload(i) for i in val]
-      
+
         return new
 
 class GGGroup(awsiot.ModeledClass):
     __slots__ = ['gg_group_id', 'cores', 'certificate_authorities']
-        
+
     def ___init___(self):
        for slot in self.__slots__:
            setattr(self, slot, None)
@@ -160,23 +160,23 @@ def from_payload(cls, payload):
             new.cores = [GGCore.from_payload(i) for i in val]
         val = payload.get('CAs')
         if val is not None:
-            new.certificate_authorities = val 
+            new.certificate_authorities = val
 
-        return new 
+        return new
 
 class DiscoverResponse(awsiot.ModeledClass):
     __slots__ = ['gg_groups']
 
     def ___init___(self):
         for slot in self.__slots__:
             setattr(self, slot, None)
-        
+
     @classmethod
     def from_payload(cls, payload):
         # type: (typing.Dict[str, typing.Any]) -> DiscoverResponse
         new = cls()
         val = payload.get('GGGroups')
         if val is not None:
             new.gg_groups = [GGGroup.from_payload(i) for i in val]
-       
-        return new           
+
+        return new

samples/basic_discovery.py

@@ -63,19 +63,15 @@
 socket_options = io.SocketOptions()
 socket_options.connect_timeout_ms = 3000
 
+print('Performing greengrass discovery...')
 discovery_client = DiscoveryClient(client_bootstrap, socket_options, tls_context, args.region)
 resp_future = discovery_client.discover(args.thing_name)
-resp = resp_future.result()
+discover_response = resp_future.result()
 
+print(discover_response)
 if args.print_discover_resp_only:
-    print(resp)
     exit(0)
 
-gg_core_tls_options = io.TlsContextOptions.create_client_with_mtls_from_path(args.certificate_path, args.private_key_path)
-gg_core_tls_options.override_default_trust_store(bytes(resp.gg_groups[0].certificate_authorities[0], encoding='utf-8'))
-gg_core_tls_ctx = io.ClientTlsContext(gg_core_tls_options)
-mqtt_client = Client(client_bootstrap, gg_core_tls_ctx)
-
 
 def on_connection_interupted(connection, error_code):
     print('connection interupted with error {}'.format(error_code))
@@ -85,27 +81,44 @@ 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:
-    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
-
-if connection_succeeded != True:
-    print('All connection attempts for core {} failed'.format(resp.gg_groups[0].cores[0].thing_arn))
-    exit(-1)
+# Try endpoints until we find one that works
+def connect_to_greengrass():
+    for gg_group in discover_response.gg_groups:
+
+        gg_core_tls_options = io.TlsContextOptions.create_client_with_mtls_from_path(args.certificate_path, args.private_key_path)
+        gg_core_tls_options.override_default_trust_store(bytes(gg_group.certificate_authorities[0], encoding='utf-8'))
+        gg_core_tls_ctx = io.ClientTlsContext(gg_core_tls_options)
+        mqtt_client = Client(client_bootstrap, gg_core_tls_ctx)
+
+        for gg_core in gg_group.cores:
+            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 = Connection(
+                        mqtt_client,
+                        on_connection_interrupted=on_connection_interupted,
+                        on_connection_resumed=on_connection_resumed)
+                    connect_future = mqtt_connection.connect(
+                        client_id=args.thing_name,
+                        host_name=connectivity_info.host_address,
+                        port=connectivity_info.port,
+                        clean_session=False)
+                    connect_future.result()
+                    print('Connected!')
+                    return mqtt_connection
+
+                except Exception as e:
+                    print('Connection failed with exception {}'.format(e))
+                    continue
+
+    exit('All connection attempts failed')
+
+mqtt_connection = connect_to_greengrass()
 
 if args.mode == 'both' or args.mode == 'subscribe':
 
     def on_publish(topic, message):
-        print('publish recieved on topic {}'.format(topic))
+        print('Publish received on topic {}'.format(topic))
         print(message)
 
     subscribe_future = mqtt_connection.subscribe(args.topic, QoS.AT_MOST_ONCE, on_publish)

setup.py

@@ -17,7 +17,7 @@
 
 setup(
     name='awsiotsdk',
-    version='0.2.9',
+    version='0.3.0',
     description='AWS IoT SDK based on the AWS Common Runtime',
     author='AWS SDK Common Runtime Team',
     url='https://github.com/awslabs/aws-iot-device-sdk-python-v2',