Add buffer_overflow_handler

repeatedly · repeatedly · commit 3909c4309c90 · 2016-07-21T17:03:58.000+09:00
diff --git a/README.rst b/README.rst
@@ -93,6 +93,25 @@ fluent, with tag 'app.follow' and the attributes 'from' and 'to'.
       'to':   'userB'
     })
 
+Handler for buffer overflow
+~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+You can inject your own custom proc to handle buffer overflow in the event of connection failure. This will mitigate the loss of data instead of simply throwing data away.
+
+.. code:: python
+
+    import msgpack
+    from io import BytesIO
+
+    def handler(pendings):
+        unpacker = msgpack.Unpacker(BytesIO(pendings))
+        for unpacked in unpacker:
+            print(unpacked)
+
+    sender.setup('app', host='host', port=24224, buffer_overflow_handler=handler)
+
+You should handle any exception in handler. fluent-logger ignores exceptions from ``buffer_overflow_handler``.
+
 Python logging.Handler interface
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
diff --git a/fluent/sender.py b/fluent/sender.py
@@ -36,6 +36,7 @@ def __init__(self,
                  bufmax=1 * 1024 * 1024,
                  timeout=3.0,
                  verbose=False,
+                 buffer_overflow_handler=None,
                  **kwargs):
 
         self.tag = tag
@@ -44,6 +45,7 @@ def __init__(self,
         self.bufmax = bufmax
         self.timeout = timeout
         self.verbose = verbose
+        self.buffer_overflow_handler = buffer_overflow_handler
 
         self.socket = None
         self.pendings = None
@@ -106,7 +108,7 @@ def _send_internal(self, bytes_):
             self._close()
             # clear buffer if it exceeds max bufer size
             if self.pendings and (len(self.pendings) > self.bufmax):
-                # TODO: add callback handler here
+                self._call_buffer_overflow_handler(self.pendings)
                 self.pendings = None
             else:
                 self.pendings = bytes_
@@ -123,6 +125,14 @@ def _reconnect(self):
                 sock.connect((self.host, self.port))
             self.socket = sock
 
+    def _call_buffer_overflow_handler(self, pending_events):
+        try:
+            if self.buffer_overflow_handler:
+                self.buffer_overflow_handler(pending_events)
+        except Exception as e:
+            # User should care any exception in handler
+            pass
+
     def _close(self):
         if self.socket:
             self.socket.close()
