def run_forever()

in python-example-elasticsearch-extension/extensions/logs_api_elasticsearch_extension.py [0:0]

• 8 lines of code
• 7 McCabe index (conditional complexity)

    def run_forever(self):
        print(f"Serving LogsAPIHTTPExternalExtension {self.agent_name}")
        while True:
            resp = self.extensions_api_client.next(self.agent_id)
            # Process the received batches if any.
            while not self.queue.empty():
                batch = self.queue.get_nowait()
                # This line logs the events received to CloudWatch.
                # Replace it to send logs to elsewhere.
                # If you've subscribed to extension logs, e.g. "types": ["platform", "function", "extension"],
                # you'll receive the logs of this extension back from Logs API.
                # And if you log it again with the line below, it will create a cycle since you receive it back again.
                # Use `extension` log type if you'll egress it to another endpoint,
                # or make sure you've implemented a protocol to handle this case.
                for item in batch:
                    self.es_producer.send(item)