Wait at most 30s for empty socket write buffer

Closes #83.

Author: eht16

docs/config.rst

@@ -267,6 +267,16 @@ for easy modification.
     *Default*: ``5.0``
 
 
+``constants.SOCKET_CLOSE_WAIT_TIMEOUT``
+
+    Maximum time in seconds to wait for the socket's write buffer to get empty.
+    Set to 0 to disable waiting for the socket write buffer to get empty.
+
+    *Type*: ``float``
+
+    *Default*: ``30.0``
+
+
 ``constants.QUEUE_CHECK_INTERVAL``
 
     Interval in seconds to check the internal queue for new messages

logstash_async/constants.py

@@ -13,6 +13,8 @@ class Constants:
     """
     # timeout in seconds for TCP connections
     SOCKET_TIMEOUT = 5.0
+    # maximum time in seconds to wait for the socket's write buffer to get empty
+    SOCKET_CLOSE_WAIT_TIMEOUT = 30.0
     # interval in seconds to check the internal queue for new messages to be cached in the database
     QUEUE_CHECK_INTERVAL = 2.0
     # interval in seconds to send cached events from the database to Logstash

logstash_async/transport.py

@@ -18,6 +18,7 @@
 import pylogbeat
 import requests
 
+from logstash_async.constants import constants
 from logstash_async.utils import ichunked
 
 
@@ -126,12 +127,22 @@ def _convert_data_to_send(self, data):
     def _close(self, force=False):
         if not self._keep_connection or force:
             if self._sock:
-                while not self._is_sock_write_buff_empty():
-                    time.sleep(0.05)
+                self._wait_for_socket_buffer_empty()
                 self._sock.shutdown(socket.SHUT_WR)
                 self._sock.close()
                 self._sock = None
 
+    # ----------------------------------------------------------------------
+    def _wait_for_socket_buffer_empty(self):
+        wait_timeout = constants.SOCKET_CLOSE_WAIT_TIMEOUT
+        interval = 0.05
+        time_waited = 0
+        # wait until the socket's write buffer is empty
+        # but do not wait longer than SOCKET_CLOSE_WAIT_TIMEOUT
+        while not self._is_sock_write_buff_empty() and time_waited < wait_timeout:
+            time_waited += interval
+            time.sleep(interval)
+
     # ----------------------------------------------------------------------
     def _is_sock_write_buff_empty(self):
         socket_fd = self._sock.fileno()