def connect()

in lab/simulator/mqttclient.py [0:0]

• 23 lines of code
• 1 McCabe index (conditional complexity)

    def connect(self):
        """
        Method to connect to IoT MQTT via IAM mode.
        It uses the current exection role to setup the connection.
        """
        event_loop_group = io.EventLoopGroup()
        host_resolver = io.DefaultHostResolver(event_loop_group)
        client_bootstrap = io.ClientBootstrap(event_loop_group, host_resolver)
        credentials_provider = auth.AwsCredentialsProvider.new_default_chain(
            client_bootstrap)

        iot_client = boto3.client('iot')
        endpoint = iot_client.describe_endpoint(
            endpointType='iot:Data-ATS')['endpointAddress']
        region = endpoint.split(".")[2]

        mqtt_connection = mqtt_connection_builder.websockets_with_default_aws_signing(
            endpoint=endpoint,
            client_bootstrap=client_bootstrap,
            client_id=self.client_id,
            region=region,
            credentials_provider=credentials_provider,
            clean_session=False)

        logging.info("Connecting to {} with client ID '{}'...".format(endpoint, self.client_id))
        # Make the connect() call
        connect_future = mqtt_connection.connect()
        # Future.result() waits until a result is available
        connect_future.result()
        logging.info("Connected!")
        self.mqtt_connection = mqtt_connection
        return True