def publish()

in static/code/worker-safety/deeplens-lambda/greengrasssdk/IoTDataPlane.py [0:0]

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

    def publish(self, **kwargs):
        r"""
        Publishes state information.

        :Keyword Arguments:
            * *topic* (``string``) --
              [REQUIRED]
              The name of the MQTT topic.
            * *payload* (``bytes or seekable file-like object``) --
              The state information, in JSON format.

        :returns: None
        """

        topic = self._get_required_parameter('topic', **kwargs)

        # payload is an optional parameter
        payload = kwargs.get('payload', b'')

        function_arn = ROUTER_FUNCTION_ARN
        client_context = {
            'custom': {
                'source': MY_FUNCTION_ARN,
                'subject': topic
            }
        }

        customer_logger.info('Publishing message on topic "{}" with Payload "{}"'.format(topic, payload))
        self.lambda_client._invoke_internal(
            function_arn,
            payload,
            base64.b64encode(json.dumps(client_context).encode())
        )