guard the queue

Signed-off-by: Clemens Vasters <[email protected]>

Author: clemensv

mode-s/mode_s_kafka_bridge/mode_s.py

@@ -62,6 +62,9 @@ def handle_messages(self, messages):
         try:
             if self.stop_flag:
                 return
+            if (len(self.task_queue ) + 1) > 200:
+                logger.warning("Dropping input. Queue capacity exceeded.")   
+                return             
             msgs = []
             for msg, ts in messages:
                 try:
@@ -136,8 +139,6 @@ def handle_messages(self, messages):
             if len(msgs) > 0:
                 bundle = Messages(messages=msgs)
                 self.task_queue.append(bundle)
-                if len(self.task_queue) > 20:
-                    logger.warning("Queue length is now %d", len(self.task_queue))
         except Exception as e:
             logger.error("Error handling messages: %s", e)
             # we're going to observe a moment of silence and hope the problem goes away