clear queue when socket.gaierror occurs

Signed-off-by: enjoy-binbin <[email protected]>

Author: enjoy-binbin

.gitignore

@@ -10,3 +10,5 @@
 /build
 /dist
 .idea/
+venv/
+env/

fluent/asyncsender.py

@@ -1,5 +1,6 @@
 # -*- coding: utf-8 -*-
 
+import socket
 import threading
 from queue import Queue, Full, Empty
 
@@ -81,11 +82,7 @@ def close(self, flush=True):
                 return
             self._closed = True
             if not flush:
-                while True:
-                    try:
-                        self._queue.get(block=False)
-                    except Empty:
-                        break
+                self._clear_queue()
             self._queue.put(_TOMBSTONE)
             self._send_thread.join()
 
@@ -101,6 +98,13 @@ def queue_blocking(self):
     def queue_circular(self):
         return self._queue_circular
 
+    def _clear_queue(self):
+        while True:
+            try:
+                self._queue.get(block=False)
+            except Empty:
+                break
+
     def _send(self, bytes_):
         with self.lock:
             if self._closed:
@@ -120,8 +124,20 @@ def _send(self, bytes_):
 
             return True
 
+    def _send_internal(self, bytes_):
+        send_internal_result = super(FluentSender, self)._send_internal(bytes_)
+        if send_internal_result is False:
+            # when send_result is False, super() caught socket.error
+            # and assigned the error to self.last_error
+            if isinstance(self.last_error, socket.gaierror):
+                # clear the queue to avoid blocking and print the log
+                self._clear_queue()
+                print("%s. Please check address: (%s, %s)" % (str(self.last_error), self.host, self.port))
+
+        return send_internal_result
+
     def _send_loop(self):
-        send_internal = super(FluentSender, self)._send_internal
+        send_internal = self._send_internal
 
         try:
             while True:

tests/test_asyncsender.py

@@ -330,7 +330,6 @@ def test_simple(self):
 
 
 class TestSenderUnlimitedSize(unittest.TestCase):
-    Q_SIZE = 3
 
     def setUp(self):
         super(TestSenderUnlimitedSize, self).setUp()
@@ -374,3 +373,21 @@ def test_simple(self):
         eq(3, len(el))
         eq("test.foo{}".format(NUM), el[0])
         eq({'bar': "baz{}".format(NUM)}, el[2])
+
+
+class TestSenderSocketGaierror(unittest.TestCase):
+    def setUp(self):
+        super(TestSenderSocketGaierror, self).setUp()
+        self._sender = fluent.asyncsender.FluentSender(host='localhost_error', tag='test')
+
+    def tearDown(self):
+        self._sender.close()
+
+    def test_simple(self):
+        with self._sender as sender:
+            for _ in range(100):
+                sender._queue.put(b"1")
+
+            sender._queue.put(fluent.asyncsender._TOMBSTONE)
+
+        self.assertEqual(self._sender._queue.qsize(), 0)