Add shutdown routine to clean up properly

Author: repeatedly

README.rst

@@ -93,6 +93,12 @@ 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
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
@@ -112,6 +118,8 @@ You can inject your own custom proc to handle buffer overflow in the event of co
 
 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
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

fluent/sender.py

@@ -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,
@@ -71,6 +73,20 @@ def emit_with_time(self, label, timestamp, data):
                                         "traceback": traceback.format_exc()})
         self._send(bytes_)
 
+    def close(self):
+        self.lock.acquire()
+        try:
+            if self.pendings:
+                try:
+                    self._send_data(self.pendings)
+                except Exception:
+                    self._call_buffer_overflow_handler(self.pendings)
+        finally:
+            self.lock.release()
+
+        self._close()
+        self.pendings = None
+
     def _make_packet(self, label, timestamp, data):
         if label:
             tag = '.'.join((self.tag, label))
@@ -95,11 +111,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
@@ -113,6 +125,12 @@ def _send_internal(self, bytes_):
             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://'):