dotlambda
diff --git a/‎README.rst
Lines changed: 27 additions & 0 deletions b/‎README.rst
Lines changed: 27 additions & 0 deletions
diff --git a/‎fluent/sender.py
Lines changed: 35 additions & 6 deletions b/‎fluent/sender.py
Lines changed: 35 additions & 6 deletions
@@ -93,6 +93,33 @@ fluent, with tag 'app.follow' and the attributes 'from' and 'to'.
       'to':   'userB'
     })
 
+If you want to shutdown the client, call `close()` method.
+
+.. code:: python
+
+    sender.close()
+
+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``.
+
+This handler is also called when pending events exist during `close()`.
+
 Python logging.Handler interface
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 
@@ -27,6 +27,8 @@ def setup(tag, **kwargs):
 def get_global_sender():
     return _global_sender
 
+def close():
+    get_global_sender().close()
 
 class FluentSender(object):
     def __init__(self,
@@ -36,6 +38,7 @@ def __init__(self,
                  bufmax=1 * 1024 * 1024,
                  timeout=3.0,
                  verbose=False,
+                 buffer_overflow_handler=None,
                  **kwargs):
 
         self.tag = tag
@@ -44,6 +47,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
@@ -69,6 +73,21 @@ def emit_with_time(self, label, timestamp, data):
                                         "traceback": traceback.format_exc()})
         self._send(bytes_)
 
+    def close(self):
+
self.lock.acquire()        self.lock.acquire()
+        try:
+            if self.pendings:
+                try:
+                    self._send_data(self.pendings)
+                except Exception:
+                    self._call_buffer_overflow_handler(self.pendings)
+
+            self._close()
+            self.pendings = None
+        finally:
+            self.lock.release()
+
+
     def _make_packet(self, label, timestamp, data):
         if label:
             tag = '.'.join((self.tag, label))
@@ -93,11 +112,7 @@ def _send_internal(self, bytes_):
             bytes_ = self.pendings
 
         try:
-            # reconnect if possible
-            self._reconnect()
-
-            # send message
-            self.socket.sendall(bytes_)
+            self._send_data(bytes_)
 
             # send finished
             self.pendings = None
@@ -106,11 +121,17 @@ 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_
 
+    def _send_data(self, bytes_):
+        # reconnect if possible
+        self._reconnect()
+        # send message
+        self.socket.sendall(bytes_)
+
     def _reconnect(self):
         if not self.socket:
             if self.host.startswith('unix://'):
@@ -123,6 +144,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()