implement UDS (Unix Domain Socket) sender

fixes #9 (thx @kmorey for the original patch)

Author: kzk

fluent/sender.py

@@ -28,7 +28,7 @@ def __init__(self,
                  tag,
                  host='localhost',
                  port=24224,
-                 bufmax=1*1024*1024,
+                 bufmax=1 * 1024 * 1024,
                  timeout=3.0,
                  verbose=False):
 
@@ -102,9 +102,14 @@ def _send_internal(self, bytes_):
 
     def _reconnect(self):
         if not self.socket:
-            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-            sock.settimeout(self.timeout)
-            sock.connect((self.host, self.port))
+            if self.host.startswith('unix://'):
+                sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+                sock.settimeout(self.timeout)
+                sock.connect(self.host[len('unix://'):])
+            else:
+                sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+                sock.settimeout(self.timeout)
+                sock.connect((self.host, self.port))
             self.socket = sock
 
     def _close(self):

tests/__init__.py

@@ -1,5 +1,6 @@
 # -*- coding: utf-8 -*-
 
 from tests.test_event import *
-from tests.test_sender import *
 from tests.test_handler import *
+from tests.test_sender import *
+from tests.test_unix_domain_socket_sender import *

tests/mockserver.py

@@ -16,9 +16,13 @@ class MockRecvServer(threading.Thread):
     """
     Single threaded server accepts one connection and recv until EOF.
     """
-    def __init__(self, port):
-        self._sock = socket.socket()
-        self._sock.bind(('localhost', port))
+    def __init__(self, host='localhost', port=24224):
+        if host.startswith('unix://'):
+            self._sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+            self._sock.bind(host[len('unix://'):])
+        else:
+            self._sock = socket.socket()
+            self._sock.bind((host, port))
         self._buf = BytesIO()
 
         threading.Thread.__init__(self)

tests/test_event.py

@@ -8,7 +8,7 @@
 sender.setup(server='localhost', tag='app')
 
 
-class TestHandler(unittest.TestCase):
+class TestEvent(unittest.TestCase):
     def testLogging(self):
         # send event with tag app.follow
         event.Event('follow', {

tests/test_handler.py

@@ -8,12 +8,12 @@
 from tests import mockserver
 
 
-class TestLogger(unittest.TestCase):
+class TestHandler(unittest.TestCase):
     def setUp(self):
-        super(TestLogger, self).setUp()
+        super(TestHandler, self).setUp()
         for port in range(10000, 20000):
             try:
-                self._server = mockserver.MockRecvServer(port)
+                self._server = mockserver.MockRecvServer('localhost', port)
                 self._port = port
                 break
             except IOError:

tests/test_sender.py

@@ -13,13 +13,11 @@ def setUp(self):
         super(TestSender, self).setUp()
         for port in range(10000, 20000):
             try:
-                self._server = mockserver.MockRecvServer(port)
+                self._server = mockserver.MockRecvServer('localhost', port)
                 break
             except IOError as exc:
                 print(exc)
-        self._sender = fluent.sender.FluentSender(tag='test',
-                                                  port=port,
-                                                  )
+        self._sender = fluent.sender.FluentSender(tag='test', port=port)
 
     def get_data(self):
         return self._server.get_recieved()