Add queue overflow handler in asyncsender.

Paweł Guz · Paweł Guz · commit 50119c55a1c6 · 2020-01-29T13:07:51.000+01:00
diff --git a/fluent/asyncsender.py b/fluent/asyncsender.py
@@ -55,6 +55,7 @@ def __init__(self,
                  msgpack_kwargs=None,
                  queue_maxsize=DEFAULT_QUEUE_MAXSIZE,
                  queue_circular=DEFAULT_QUEUE_CIRCULAR,
+                 queue_overflow_handler=None,
                  **kwargs):
         """
         :param kwargs: This kwargs argument is not used in __init__. This will be removed in the next major version.
@@ -66,6 +67,7 @@ def __init__(self,
                                            **kwargs)
         self._queue_maxsize = queue_maxsize
         self._queue_circular = queue_circular
+        self._queue_overflow_handler = queue_overflow_handler
 
         self._thread_guard = threading.Event()  # This ensures visibility across all variables
         self._closed = False
@@ -109,7 +111,8 @@ def _send(self, bytes_):
             if self._queue_circular and self._queue.full():
                 # discard oldest
                 try:
-                    self._queue.get(block=False)
+                    discarded_bytes = self._queue.get(block=False)
+                    self._call_queue_overflow_handler(discarded_bytes)
                 except Empty:  # pragma: no cover
                     pass
             try:
@@ -132,5 +135,13 @@ def _send_loop(self):
         finally:
             self._close()
 
+    def _call_queue_overflow_handler(self, discarded_bytes):
+        try:
+            if self._queue_overflow_handler:
+                self._queue_overflow_handler(discarded_bytes)
+        except Exception as e:
+            # User should care any exception in handler
+            pass
+
     def __exit__(self, exc_type, exc_val, exc_tb):
         self.close()
