diff --git a/README.rst b/README.rst
index 8d34d5d..c243652 100644
--- a/README.rst
+++ b/README.rst
@@ -282,6 +282,72 @@ A sample configuration ``logging.yaml`` would be:
                 level: DEBUG
                 propagate: False
 
+Asynchronous Communication
+~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Besides the regular interfaces - the event-based one provided by ``sender.FluentSender`` and the python logging one
+provided by ``handler.FluentHandler`` - there are also corresponding asynchronous versions in ``asyncsender`` and
+``asynchandler`` respectively. These versions use a separate thread to handle the communication with the remote fluentd
+server. In this way the client of the library won't be blocked during the logging of the events, and won't risk going
+into timeout if the fluentd server becomes unreachable. Also it won't be slowed down by the network overhead.
+
+The interfaces in ``asyncsender`` and ``asynchandler`` are exactly the same as those in ``sender`` and ``handler``, so it's
+just a matter of importing from a different module.
+
+For instance, for the event-based interface:
+
+.. code:: python
+
+    from fluent import asyncsender as sender
+
+    # for local fluent
+    sender.setup('app')
+
+    # for remote fluent
+    sender.setup('app', host='host', port=24224)
+
+    # do your work
+    ...
+
+    # IMPORTANT: before program termination, close the sender
+    sender.close()
+
+or for the python logging interface:
+
+.. code:: python
+
+    import logging
+    from fluent import asynchandler as handler
+
+    custom_format = {
+      'host': '%(hostname)s',
+      'where': '%(module)s.%(funcName)s',
+      'type': '%(levelname)s',
+      'stack_trace': '%(exc_text)s'
+    }
+
+    logging.basicConfig(level=logging.INFO)
+    l = logging.getLogger('fluent.test')
+    h = handler.FluentHandler('app.follow', host='host', port=24224, buffer_overflow_handler=overflow_handler)
+    formatter = handler.FluentRecordFormatter(custom_format)
+    h.setFormatter(formatter)
+    l.addHandler(h)
+    l.info({
+      'from': 'userA',
+      'to': 'userB'
+    })
+    l.info('{"from": "userC", "to": "userD"}')
+    l.info("This log entry will be logged with the additional key: 'message'.")
+
+    ...
+
+    # IMPORTANT: before program termination, close the handler
+    h.close()
+
+**NOTE**: please note that it's important to close the sender or the handler at program termination. This will make
+sure the communication thread terminates and it's joined correctly. Otherwise the program won't exit, waiting for
+the thread, unless forcibly killed.
+
 Testing
 -------
 
diff --git a/fluent/asynchandler.py b/fluent/asynchandler.py
new file mode 100644
index 0000000..e986313
--- /dev/null
+++ b/fluent/asynchandler.py
@@ -0,0 +1,18 @@
+# -*- coding: utf-8 -*-
+
+from fluent import asyncsender
+from fluent import handler
+from fluent.handler import FluentRecordFormatter
+
+
+class FluentHandler(handler.FluentHandler):
+    '''
+    Asynchronous Logging Handler for fluent.
+    '''
+
+    def getSenderClass(self):
+        return asyncsender.FluentSender
+
+    def close(self):
+        self.sender.close()
+        super(FluentHandler, self).close()
diff --git a/fluent/asyncsender.py b/fluent/asyncsender.py
new file mode 100644
index 0000000..178a0ba
--- /dev/null
+++ b/fluent/asyncsender.py
@@ -0,0 +1,195 @@
+# -*- coding: utf-8 -*-
+
+from __future__ import print_function
+import threading
+import time
+try:
+    from queue import Queue, Full, Empty
+except ImportError:
+    from Queue import Queue, Full, Empty
+
+from fluent import sender
+from fluent.sender import EventTime
+
+_global_sender = None
+
+DEFAULT_QUEUE_TIMEOUT = 0.05
+
+
+def _set_global_sender(sender):
+    """ [For testing] Function to set global sender directly
+    """
+    global _global_sender
+    _global_sender = sender
+
+
+def setup(tag, **kwargs):
+    global _global_sender
+    _global_sender = FluentSender(tag, **kwargs)
+
+
+def get_global_sender():
+    return _global_sender
+
+
+def close():
+    get_global_sender().close()
+
+
+class CommunicatorThread(threading.Thread):
+    def __init__(self, tag,
+                 host='localhost',
+                 port=24224,
+                 bufmax=1 * 1024 * 1024,
+                 timeout=3.0,
+                 verbose=False,
+                 buffer_overflow_handler=None,
+                 nanosecond_precision=False,
+                 msgpack_kwargs=None,
+                 queue_timeout=DEFAULT_QUEUE_TIMEOUT, *args, **kwargs):
+        super(CommunicatorThread, self).__init__(**kwargs)
+        self._queue = Queue()
+        self._do_run = True
+        self._queue_timeout = queue_timeout
+        self._conn_close_lock = threading.Lock()
+        self._sender = sender.FluentSender(tag=tag, host=host, port=port, bufmax=bufmax, timeout=timeout,
+                                           verbose=verbose, buffer_overflow_handler=buffer_overflow_handler,
+                                           nanosecond_precision=nanosecond_precision, msgpack_kwargs=msgpack_kwargs)
+
+    def send(self, bytes_):
+        try:
+            self._queue.put(bytes_)
+        except Full:
+            return False
+        return True
+
+    def run(self):
+        while self._do_run:
+            try:
+                bytes_ = self._queue.get(block=True, timeout=self._queue_timeout)
+            except Empty:
+                continue
+            self._conn_close_lock.acquire()
+            self._sender._send(bytes_)
+            self._conn_close_lock.release()
+
+    def close(self, flush=True, discard=True):
+        if discard:
+            while not self._queue.empty():
+                try:
+                    self._queue.get(block=False)
+                except Empty:
+                    break
+        while flush and (not self._queue.empty()):
+            time.sleep(0.1)
+        self._do_run = False
+        self._sender.close()
+
+    def _close(self):
+        self._conn_close_lock.acquire()
+        # self._sender.lock.acquire()
+        try:
+            self._sender._close()
+        finally:
+            # self._sender.lock.release()
+            self._conn_close_lock.release()
+            pass
+
+    @property
+    def last_error(self):
+        return self._sender.last_error
+
+    @last_error.setter
+    def last_error(self, err):
+        self._sender.last_error = err
+
+    def clear_last_error(self, _thread_id = None):
+        self._sender.clear_last_error(_thread_id=_thread_id)
+
+    @property
+    def queue_timeout(self):
+        return self._queue_timeout
+
+    @queue_timeout.setter
+    def queue_timeout(self, value):
+        self._queue_timeout = value
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, typ, value, traceback):
+        self.close()
+
+
+class FluentSender(sender.FluentSender):
+    def __init__(self,
+                 tag,
+                 host='localhost',
+                 port=24224,
+                 bufmax=1 * 1024 * 1024,
+                 timeout=3.0,
+                 verbose=False,
+                 buffer_overflow_handler=None,
+                 nanosecond_precision=False,
+                 msgpack_kwargs=None,
+                 queue_timeout=DEFAULT_QUEUE_TIMEOUT,
+                 **kwargs): # This kwargs argument is not used in __init__. This will be removed in the next major version.
+        super(FluentSender, self).__init__(tag=tag, host=host, port=port, bufmax=bufmax, timeout=timeout,
+                                           verbose=verbose, buffer_overflow_handler=buffer_overflow_handler,
+                                           nanosecond_precision=nanosecond_precision, msgpack_kwargs=msgpack_kwargs,
+                                           **kwargs)
+        self._communicator = CommunicatorThread(tag=tag, host=host, port=port, bufmax=bufmax, timeout=timeout,
+                                                verbose=verbose, buffer_overflow_handler=buffer_overflow_handler,
+                                                nanosecond_precision=nanosecond_precision, msgpack_kwargs=msgpack_kwargs,
+                                                queue_timeout=queue_timeout)
+        self._communicator.start()
+
+    def _send(self, bytes_):
+        return self._communicator.send(bytes_=bytes_)
+
+    def _close(self):
+        # super(FluentSender, self)._close()
+        self._communicator._close()
+
+    def _send_internal(self, bytes_):
+        return
+
+    def _send_data(self, bytes_):
+        return
+
+    # override reconnect, so we don't open a socket here (since it
+    # will be opened by the CommunicatorThread)
+    def _reconnect(self):
+        return
+
+    def close(self):
+        self._communicator.close(flush=True)
+        self._communicator.join()
+        return super(FluentSender, self).close()
+
+    @property
+    def last_error(self):
+        return self._communicator.last_error
+
+    @last_error.setter
+    def last_error(self, err):
+        self._communicator.last_error = err
+
+    def clear_last_error(self, _thread_id = None):
+        self._communicator.clear_last_error(_thread_id=_thread_id)
+
+    @property
+    def queue_timeout(self):
+        return self._communicator.queue_timeout
+
+    @queue_timeout.setter
+    def queue_timeout(self, value):
+        self._communicator.queue_timeout = value
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, typ, value, traceback):
+        # give time to the comm. thread to send its queued messages
+        time.sleep(0.2)
+        self.close()
diff --git a/fluent/handler.py b/fluent/handler.py
index 08320d2..e69af2b 100644
--- a/fluent/handler.py
+++ b/fluent/handler.py
@@ -141,14 +141,28 @@ def __init__(self,
                  nanosecond_precision=False):
 
         self.tag = tag
-        self.sender = sender.FluentSender(tag,
-                                          host=host, port=port,
-                                          timeout=timeout, verbose=verbose,
-                                          buffer_overflow_handler=buffer_overflow_handler,
-                                          msgpack_kwargs=msgpack_kwargs,
-                                          nanosecond_precision=nanosecond_precision)
+        self.sender = self.getSenderInstance(tag,
+                                             host=host, port=port,
+                                             timeout=timeout, verbose=verbose,
+                                             buffer_overflow_handler=buffer_overflow_handler,
+                                             msgpack_kwargs=msgpack_kwargs,
+                                             nanosecond_precision=nanosecond_precision)
         logging.Handler.__init__(self)
 
+    def getSenderClass(self):
+        return sender.FluentSender
+
+    def getSenderInstance(self, tag, host, port, timeout, verbose,
+                          buffer_overflow_handler, msgpack_kwargs,
+                          nanosecond_precision):
+        sender_class = self.getSenderClass()
+        return sender_class(tag,
+                            host=host, port=port,
+                            timeout=timeout, verbose=verbose,
+                            buffer_overflow_handler=buffer_overflow_handler,
+                            msgpack_kwargs=msgpack_kwargs,
+                            nanosecond_precision=nanosecond_precision)
+
     def emit(self, record):
         data = self.format(record)
         return self.sender.emit(None, data)
diff --git a/tests/test_asynchandler.py b/tests/test_asynchandler.py
new file mode 100644
index 0000000..2725c6a
--- /dev/null
+++ b/tests/test_asynchandler.py
@@ -0,0 +1,279 @@
+# -*- coding: utf-8 -*-
+
+import logging
+import sys
+import unittest
+import time
+
+import fluent.handler
+import fluent.asynchandler
+
+from tests import mockserver
+
+
+class TestHandler(unittest.TestCase):
+    def setUp(self):
+        super(TestHandler, self).setUp()
+        self._server = mockserver.MockRecvServer('localhost')
+        self._port = self._server.port
+        self.handler = None
+
+    def get_handler_class(self):
+        # return fluent.handler.FluentHandler
+        return fluent.asynchandler.FluentHandler
+
+    def get_data(self):
+        return self._server.get_recieved()
+
+    def test_simple(self):
+        handler = self.get_handler_class()('app.follow', port=self._port)
+        self.handler = handler
+
+        logging.basicConfig(level=logging.INFO)
+        log = logging.getLogger('fluent.test')
+        handler.setFormatter(fluent.handler.FluentRecordFormatter())
+        log.addHandler(handler)
+        log.info({
+            'from': 'userA',
+            'to': 'userB'
+        })
+
+        # wait, giving time to the communicator thread to send the messages
+        time.sleep(0.5)
+        # close the handler, to join the thread and let the test suite to terminate
+        handler.close()
+
+        data = self.get_data()
+        eq = self.assertEqual
+        eq(1, len(data))
+        eq(3, len(data[0]))
+        eq('app.follow', data[0][0])
+        eq('userA', data[0][2]['from'])
+        eq('userB', data[0][2]['to'])
+        self.assertTrue(data[0][1])
+        self.assertTrue(isinstance(data[0][1], int))
+
+    def test_custom_fmt(self):
+        handler = self.get_handler_class()('app.follow', port=self._port)
+
+        logging.basicConfig(level=logging.INFO)
+        log = logging.getLogger('fluent.test')
+        handler.setFormatter(
+            fluent.handler.FluentRecordFormatter(fmt={
+                'name': '%(name)s',
+                'lineno': '%(lineno)d',
+                'emitted_at': '%(asctime)s',
+            })
+        )
+        log.addHandler(handler)
+        log.info({'sample': 'value'})
+        # wait, giving time to the communicator thread to send the messages
+        time.sleep(0.5)
+        # close the handler, to join the thread and let the test suite to terminate
+        handler.close()
+
+        data = self.get_data()
+        self.assertTrue('name' in data[0][2])
+        self.assertEqual('fluent.test', data[0][2]['name'])
+        self.assertTrue('lineno' in data[0][2])
+        self.assertTrue('emitted_at' in data[0][2])
+
+    @unittest.skipUnless(sys.version_info[0:2] >= (3, 2), 'supported with Python 3.2 or above')
+    def test_custom_fmt_with_format_style(self):
+        handler = self.get_handler_class()('app.follow', port=self._port)
+
+        logging.basicConfig(level=logging.INFO)
+        log = logging.getLogger('fluent.test')
+        handler.setFormatter(
+            fluent.handler.FluentRecordFormatter(fmt={
+                'name': '{name}',
+                'lineno': '{lineno}',
+                'emitted_at': '{asctime}',
+            }, style='{')
+        )
+        log.addHandler(handler)
+        log.info({'sample': 'value'})
+        # wait, giving time to the communicator thread to send the messages
+        time.sleep(0.5)
+        # close the handler, to join the thread and let the test suite to terminate
+        handler.close()
+
+        data = self.get_data()
+        self.assertTrue('name' in data[0][2])
+        self.assertEqual('fluent.test', data[0][2]['name'])
+        self.assertTrue('lineno' in data[0][2])
+        self.assertTrue('emitted_at' in data[0][2])
+
+    @unittest.skipUnless(sys.version_info[0:2] >= (3, 2), 'supported with Python 3.2 or above')
+    def test_custom_fmt_with_template_style(self):
+        handler = self.get_handler_class()('app.follow', port=self._port)
+
+        logging.basicConfig(level=logging.INFO)
+        log = logging.getLogger('fluent.test')
+        handler.setFormatter(
+            fluent.handler.FluentRecordFormatter(fmt={
+                'name': '${name}',
+                'lineno': '${lineno}',
+                'emitted_at': '${asctime}',
+            }, style='$')
+        )
+        log.addHandler(handler)
+        log.info({'sample': 'value'})
+        # wait, giving time to the communicator thread to send the messages
+        time.sleep(0.5)
+        # close the handler, to join the thread and let the test suite to terminate
+        handler.close()
+
+        data = self.get_data()
+        self.assertTrue('name' in data[0][2])
+        self.assertEqual('fluent.test', data[0][2]['name'])
+        self.assertTrue('lineno' in data[0][2])
+        self.assertTrue('emitted_at' in data[0][2])
+
+    def test_custom_field_raise_exception(self):
+        handler = self.get_handler_class()('app.follow', port=self._port)
+
+        logging.basicConfig(level=logging.INFO)
+        log = logging.getLogger('fluent.test')
+        handler.setFormatter(
+            fluent.handler.FluentRecordFormatter(fmt={
+                'name': '%(name)s',
+                'custom_field': '%(custom_field)s'
+            })
+        )
+        log.addHandler(handler)
+        with self.assertRaises(KeyError):
+            log.info({'sample': 'value'})
+        log.removeHandler(handler)
+        # wait, giving time to the communicator thread to send the messages
+        time.sleep(0.5)
+        # close the handler, to join the thread and let the test suite to terminate
+        handler.close()
+
+    def test_custom_field_fill_missing_fmt_key_is_true(self):
+        handler = self.get_handler_class()('app.follow', port=self._port)
+
+        logging.basicConfig(level=logging.INFO)
+        log = logging.getLogger('fluent.test')
+        handler.setFormatter(
+            fluent.handler.FluentRecordFormatter(fmt={
+                'name': '%(name)s',
+                'custom_field': '%(custom_field)s'
+                },
+                fill_missing_fmt_key=True
+            )
+        )
+        log.addHandler(handler)
+        log.info({'sample': 'value'})
+        log.removeHandler(handler)
+        # wait, giving time to the communicator thread to send the messages
+        time.sleep(0.5)
+        # close the handler, to join the thread and let the test suite to terminate
+        handler.close()
+
+        data = self.get_data()
+        self.assertTrue('name' in data[0][2])
+        self.assertEqual('fluent.test', data[0][2]['name'])
+        self.assertTrue('custom_field' in data[0][2])
+        # field defaults to none if not in log record
+        self.assertIsNone(data[0][2]['custom_field'])
+
+    def test_json_encoded_message(self):
+        handler = self.get_handler_class()('app.follow', port=self._port)
+
+        logging.basicConfig(level=logging.INFO)
+        log = logging.getLogger('fluent.test')
+        handler.setFormatter(fluent.handler.FluentRecordFormatter())
+        log.addHandler(handler)
+        log.info('{"key": "hello world!", "param": "value"}')
+        # wait, giving time to the communicator thread to send the messages
+        time.sleep(0.5)
+        # close the handler, to join the thread and let the test suite to terminate
+        handler.close()
+
+        data = self.get_data()
+        self.assertTrue('key' in data[0][2])
+        self.assertEqual('hello world!', data[0][2]['key'])
+
+    def test_unstructured_message(self):
+        handler = self.get_handler_class()('app.follow', port=self._port)
+
+        logging.basicConfig(level=logging.INFO)
+        log = logging.getLogger('fluent.test')
+        handler.setFormatter(fluent.handler.FluentRecordFormatter())
+        log.addHandler(handler)
+        log.info('hello %s', 'world')
+        # wait, giving time to the communicator thread to send the messages
+        time.sleep(0.5)
+        # close the handler, to join the thread and let the test suite to terminate
+        handler.close()
+
+        data = self.get_data()
+        self.assertTrue('message' in data[0][2])
+        self.assertEqual('hello world', data[0][2]['message'])
+
+    def test_unstructured_formatted_message(self):
+        handler = self.get_handler_class()('app.follow', port=self._port)
+
+        logging.basicConfig(level=logging.INFO)
+        log = logging.getLogger('fluent.test')
+        handler.setFormatter(fluent.handler.FluentRecordFormatter())
+        log.addHandler(handler)
+        log.info('hello world, %s', 'you!')
+        # wait, giving time to the communicator thread to send the messages
+        time.sleep(0.5)
+        # close the handler, to join the thread and let the test suite to terminate
+        handler.close()
+
+        data = self.get_data()
+        self.assertTrue('message' in data[0][2])
+        self.assertEqual('hello world, you!', data[0][2]['message'])
+
+    def test_number_string_simple_message(self):
+        handler = self.get_handler_class()('app.follow', port=self._port)
+
+        logging.basicConfig(level=logging.INFO)
+        log = logging.getLogger('fluent.test')
+        handler.setFormatter(fluent.handler.FluentRecordFormatter())
+        log.addHandler(handler)
+        log.info("1")
+        # wait, giving time to the communicator thread to send the messages
+        time.sleep(0.5)
+        # close the handler, to join the thread and let the test suite to terminate
+        handler.close()
+
+        data = self.get_data()
+        self.assertTrue('message' in data[0][2])
+
+    def test_non_string_simple_message(self):
+        handler = self.get_handler_class()('app.follow', port=self._port)
+
+        logging.basicConfig(level=logging.INFO)
+        log = logging.getLogger('fluent.test')
+        handler.setFormatter(fluent.handler.FluentRecordFormatter())
+        log.addHandler(handler)
+        log.info(42)
+        # wait, giving time to the communicator thread to send the messages
+        time.sleep(0.5)
+        # close the handler, to join the thread and let the test suite to terminate
+        handler.close()
+
+        data = self.get_data()
+        self.assertTrue('message' in data[0][2])
+
+    def test_non_string_dict_message(self):
+        handler = self.get_handler_class()('app.follow', port=self._port)
+
+        logging.basicConfig(level=logging.INFO)
+        log = logging.getLogger('fluent.test')
+        handler.setFormatter(fluent.handler.FluentRecordFormatter())
+        log.addHandler(handler)
+        log.info({42: 'root'})
+        # wait, giving time to the communicator thread to send the messages
+        time.sleep(0.5)
+        # close the handler, to join the thread and let the test suite to terminate
+        handler.close()
+
+        data = self.get_data()
+        # For some reason, non-string keys are ignored
+        self.assertFalse(42 in data[0][2])
diff --git a/tests/test_asyncsender.py b/tests/test_asyncsender.py
new file mode 100644
index 0000000..477a768
--- /dev/null
+++ b/tests/test_asyncsender.py
@@ -0,0 +1,197 @@
+# -*- coding: utf-8 -*-
+
+from __future__ import print_function
+import unittest
+import socket
+import msgpack
+import time
+
+import fluent.asyncsender
+from tests import mockserver
+
+
+class TestSetup(unittest.TestCase):
+    def tearDown(self):
+        from fluent.asyncsender import _set_global_sender
+        _set_global_sender(None)
+
+    def test_no_kwargs(self):
+        fluent.asyncsender.setup("tag")
+        actual = fluent.asyncsender.get_global_sender()
+        self.assertEqual(actual.tag, "tag")
+        self.assertEqual(actual.host, "localhost")
+        self.assertEqual(actual.port, 24224)
+        self.assertEqual(actual.timeout, 3.0)
+        actual.close()
+
+    def test_host_and_port(self):
+        fluent.asyncsender.setup("tag", host="myhost", port=24225)
+        actual = fluent.asyncsender.get_global_sender()
+        self.assertEqual(actual.tag, "tag")
+        self.assertEqual(actual.host, "myhost")
+        self.assertEqual(actual.port, 24225)
+        self.assertEqual(actual.timeout, 3.0)
+        actual.close()
+
+    def test_tolerant(self):
+        fluent.asyncsender.setup("tag", host="myhost", port=24225, timeout=1.0)
+        actual = fluent.asyncsender.get_global_sender()
+        self.assertEqual(actual.tag, "tag")
+        self.assertEqual(actual.host, "myhost")
+        self.assertEqual(actual.port, 24225)
+        self.assertEqual(actual.timeout, 1.0)
+        actual.close()
+
+
+class TestSender(unittest.TestCase):
+    def setUp(self):
+        super(TestSender, self).setUp()
+        self._server = mockserver.MockRecvServer('localhost')
+        self._sender = fluent.asyncsender.FluentSender(tag='test',
+                                                  port=self._server.port)
+
+    def tearDown(self):
+        self._sender.close()
+
+    def get_data(self):
+        return self._server.get_recieved()
+
+    def test_simple(self):
+        sender = self._sender
+        sender.emit('foo', {'bar': 'baz'})
+        time.sleep(0.5)
+        sender._close()
+        data = self.get_data()
+        eq = self.assertEqual
+        eq(1, len(data))
+        eq(3, len(data[0]))
+        eq('test.foo', data[0][0])
+        eq({'bar': 'baz'}, data[0][2])
+        self.assertTrue(data[0][1])
+        self.assertTrue(isinstance(data[0][1], int))
+
+    def test_decorator_simple(self):
+        with self._sender as sender:
+            sender.emit('foo', {'bar': 'baz'})
+        data = self.get_data()
+        eq = self.assertEqual
+        eq(1, len(data))
+        eq(3, len(data[0]))
+        eq('test.foo', data[0][0])
+        eq({'bar': 'baz'}, data[0][2])
+        self.assertTrue(data[0][1])
+        self.assertTrue(isinstance(data[0][1], int))
+
+    def test_nanosecond(self):
+        sender = self._sender
+        sender.nanosecond_precision = True
+        sender.emit('foo', {'bar': 'baz'})
+        time.sleep(0.5)
+        sender._close()
+        data = self.get_data()
+        eq = self.assertEqual
+        eq(1, len(data))
+        eq(3, len(data[0]))
+        eq('test.foo', data[0][0])
+        eq({'bar': 'baz'}, data[0][2])
+        self.assertTrue(isinstance(data[0][1], msgpack.ExtType))
+        eq(data[0][1].code, 0)
+
+    def test_nanosecond_coerce_float(self):
+        time_ = 1490061367.8616468906402588
+        sender = self._sender
+        sender.nanosecond_precision = True
+        sender.emit_with_time('foo', time_, {'bar': 'baz'})
+        time.sleep(0.5)
+        sender._close()
+        data = self.get_data()
+        eq = self.assertEqual
+        eq(1, len(data))
+        eq(3, len(data[0]))
+        eq('test.foo', data[0][0])
+        eq({'bar': 'baz'}, data[0][2])
+        self.assertTrue(isinstance(data[0][1], msgpack.ExtType))
+        eq(data[0][1].code, 0)
+        eq(data[0][1].data, b'X\xd0\x8873[\xb0*')
+
+    def test_no_last_error_on_successful_emit(self):
+        sender = self._sender
+        sender.emit('foo', {'bar': 'baz'})
+        sender._close()
+
+        self.assertEqual(sender.last_error, None)
+
+    def test_last_error_property(self):
+        EXCEPTION_MSG = "custom exception for testing last_error property"
+        self._sender.last_error = socket.error(EXCEPTION_MSG)
+
+        self.assertEqual(self._sender.last_error.args[0], EXCEPTION_MSG)
+
+    def test_clear_last_error(self):
+        EXCEPTION_MSG = "custom exception for testing clear_last_error"
+        self._sender.last_error = socket.error(EXCEPTION_MSG)
+        self._sender.clear_last_error()
+
+        self.assertEqual(self._sender.last_error, None)
+
+    @unittest.skip("This test failed with 'TypeError: catching classes that do not inherit from BaseException is not allowed' so skipped")
+    #@patch('fluent.asyncsender.socket')
+    def test_connect_exception_during_sender_init(self, mock_socket):
+        # Make the socket.socket().connect() call raise a custom exception
+        mock_connect = mock_socket.socket.return_value.connect
+        EXCEPTION_MSG = "a sender init socket connect() exception"
+        mock_connect.side_effect = socket.error(EXCEPTION_MSG)
+
+        self.assertEqual(self._sender.last_error.args[0], EXCEPTION_MSG)
+
+
+class TestSenderWithTimeout(unittest.TestCase):
+    def setUp(self):
+        super(TestSenderWithTimeout, self).setUp()
+        self._server = mockserver.MockRecvServer('localhost')
+        self._sender = fluent.asyncsender.FluentSender(tag='test',
+                                                       port=self._server.port,
+                                                       queue_timeout=0.04)
+
+    def tearDown(self):
+        self._sender.close()
+
+    def get_data(self):
+        return self._server.get_recieved()
+
+    def test_simple(self):
+        sender = self._sender
+        sender.emit('foo', {'bar': 'baz'})
+        time.sleep(0.5)
+        sender._close()
+        data = self.get_data()
+        eq = self.assertEqual
+        eq(1, len(data))
+        eq(3, len(data[0]))
+        eq('test.foo', data[0][0])
+        eq({'bar': 'baz'}, data[0][2])
+        self.assertTrue(data[0][1])
+        self.assertTrue(isinstance(data[0][1], int))
+
+    def test_simple_with_timeout_props(self):
+        sender = self._sender
+        sender.queue_timeout = 0.06
+        assert sender.queue_timeout == 0.06
+        sender.emit('foo', {'bar': 'baz'})
+        time.sleep(0.5)
+        sender._close()
+        data = self.get_data()
+        eq = self.assertEqual
+        eq(1, len(data))
+        eq(3, len(data[0]))
+        eq('test.foo', data[0][0])
+        eq({'bar': 'baz'}, data[0][2])
+        self.assertTrue(data[0][1])
+        self.assertTrue(isinstance(data[0][1], int))
+
+
+class TestEventTime(unittest.TestCase):
+    def test_event_time(self):
+        time = fluent.asyncsender.EventTime(1490061367.8616468906402588)
+        self.assertEqual(time.code, 0)
+        self.assertEqual(time.data, b'X\xd0\x8873[\xb0*')