Add Support for timeout on the tornado IOLoop

Author: David Walker

.travis.yml

@@ -4,6 +4,7 @@ python:
   - "3.4"
   - "3.5"
 install:
-    pip install -e .[tornado]
+  - if [[ $TRAVIS_PYTHON_VERSION == 2* ]]; then travis_retry pip install mock; fi
+  - pip install -e .[tornado]
 # command to run tests
 script: py.test

setup.py

@@ -29,6 +29,6 @@
     packages=find_packages(exclude=['tests', '*.tests']),
     install_requires=['pyserial'],
     extras_require={
-        'tornado': ['tornado']
+        'tornado': ['tornado~=4.5']
     }
 )

xbee/tests/Fake.py

@@ -29,6 +29,7 @@ def __init__(self, port='/dev/null', baudrate=19200, timeout=1,
         self.xonxoff = xonxoff
         self.rtscts = rtscts
         self._is_open = True
+        self.fd = 0
 
         self._data_written = ""
         self._read_data = ""

xbee/tornado/base.py

@@ -11,6 +11,7 @@
 """
 from xbee.frame import APIFrame
 from xbee.backend.base import XBeeBase as _XBeeBase
+from xbee.backend.base import TimeoutException as _TimeoutException
 from tornado import ioloop, gen
 from tornado.locks import Event
 from tornado.concurrent import Future
@@ -45,27 +46,28 @@ class XBeeBase(_XBeeBase):
                  argument is also used.
     """
     def __init__(self, *args, **kwargs):
+        if 'io_loop' in kwargs:
+            self._ioloop = kwargs.pop('io_loop')
+        else:
+            self._ioloop = ioloop.IOLoop.current()
+
         super(XBeeBase, self).__init__(*args, **kwargs)
 
         self._running = Event()
         self._running.set()
 
-        if 'io_loop' in kwargs:
-            self._ioloop = kwargs['io_loop']
-        else:
-            self._ioloop = ioloop.IOLoop.current()
-
         self._frame_future = None
         self._frame_queue = deque()
 
         if self._callback:
             # Make Non-Blocking
             self.serial.timeout = 0
-            self._ioloop.add_handler(self.serial.fd,
-                                     self._process_input,
-                                     ioloop.IOLoop.READ)
             self.process_frames()
 
+        self._ioloop.add_handler(self.serial.fd,
+                                 self._process_input,
+                                 ioloop.IOLoop.READ)
+
     def halt(self):
         """
         halt: None -> None
@@ -99,15 +101,25 @@ def process_frames(self):
                     self._error_callback(e)
 
     @gen.coroutine
-    def wait_read_frame(self):
-        frame = yield self._get_frame()
+    def wait_read_frame(self, timeout=None):
+        frame = yield self._get_frame(timeout=timeout)
         raise gen.Return(self._split_response(frame.data))
 
-    def _get_frame(self):
+    def _get_frame(self, timeout=None):
         future = Future()
         if self._frame_queue:
             future.set_result(self._frame_queue.popleft())
         else:
+            if timeout is not None:
+                def on_timeout():
+                    future.set_exception(_TimeoutException())
+
+                handle = self._ioloop.add_timeout(
+                    self._ioloop.time() + timeout, on_timeout
+                )
+                future.add_done_callback(lambda _:
+                                         self._ioloop.remove_timeout(handle))
+
             self._frame_future = future
 
         return future

xbee/tornado/tests/test_base.py

@@ -9,11 +9,17 @@
 """
 
 import unittest
+try:
+    from unittest.mock import Mock
+except ImportError:
+    from mock import Mock
+
 from xbee.tornado import has_tornado
 
 if not has_tornado:
     raise unittest.SkipTest("Requires Tornado")
 
+from tornado import ioloop # noqa
 from tornado.testing import AsyncTestCase, gen_test  # noqa
 from tornado.test.util import unittest  # noqa
 from xbee.tornado.base import XBeeBase  # noqa
@@ -26,14 +32,19 @@ class TestReadFromDevice(AsyncTestCase):
     API frame
     """
 
+    def setUp(self):
+        super(TestReadFromDevice, self).setUp()
+        self._patch_io = ioloop.IOLoop.current()
+        self._patch_io.add_handler = Mock()
+
     @gen_test
     def test_read(self):
         """
         _wait_for_frame should properly read a frame of data
         """
         device = Serial()
         device.set_read_data(b'\x7E\x00\x01\x00\xFF')
-        xbee = XBeeBase(device)
+        xbee = XBeeBase(device, io_loop=self._patch_io)
 
         xbee._process_input(None, None)
         frame = yield xbee._get_frame()
@@ -46,7 +57,7 @@ def test_read_invalid_followed_by_valid(self):
         """
         device = Serial()
         device.set_read_data(b'\x7E\x00\x01\x00\xFA' + b'\x7E\x00\x01\x05\xFA')
-        xbee = XBeeBase(device)
+        xbee = XBeeBase(device, io_loop=self._patch_io)
 
         xbee._process_input(None, None)
         # First process ends with no good frame, process next
@@ -64,7 +75,7 @@ def test_read_escaped(self):
         device.set_read_data(
             b'\x7E\x00\x04\x7D\x5E\x7D\x5D\x7D\x31\x7D\x33\xE0')
 
-        xbee = XBeeBase(device, escaped=True)
+        xbee = XBeeBase(device, escaped=True, io_loop=self._patch_io)
 
         xbee._process_input(None, None)
         frame = yield xbee._get_frame()

xbee/tornado/tests/test_digimesh.py

@@ -7,11 +7,18 @@
 Tests the XBee DigiMesh implementation class for API compliance
 """
 import unittest
+try:
+    from unittest.mock import Mock
+except ImportError:
+    from mock import Mock
+
 from xbee.tornado import has_tornado
 
 if not has_tornado:
     raise unittest.SkipTest("Requires Tornado")
 
+from tornado import ioloop # noqa
+from xbee.tests.Fake import Serial  # noqa
 from xbee.tornado.digimesh import DigiMesh  # noqa
 
 
@@ -22,8 +29,11 @@ class TestDigiMesh(unittest.TestCase):
     """
 
     def setUp(self):
-        self.digimesh = DigiMesh(None)
         super(TestDigiMesh, self).setUp()
+        patch_io = ioloop.IOLoop.current()
+        patch_io.add_handler = Mock()
+        serial_port = Serial()
+        self.digimesh = DigiMesh(serial_port, io_loop=patch_io)
 
     def test_split_tx_status(self):
             data = b'\x8b\x01\xff\xff\x01\x01\x01'

xbee/tornado/tests/test_ieee.py

@@ -8,6 +8,10 @@
 Tests the XBee (IEEE 802.15.4) implementation class for XBee API compliance
 """
 import unittest
+try:
+    from unittest.mock import Mock
+except ImportError:
+    from mock import Mock
 from xbee.tornado import has_tornado
 
 if not has_tornado:
@@ -17,6 +21,7 @@
 from xbee.tornado.ieee import XBee  # noqa
 from xbee.frame import APIFrame  # noqa
 from xbee.python2to3 import intToByte, stringToBytes  # noqa
+from tornado import ioloop  # noqa
 from tornado.testing import AsyncTestCase, gen_test  # noqa
 from tornado.test.util import unittest  # noqa
 import sys  # noqa
@@ -33,7 +38,10 @@ def setUp(self):
         Initialize XBee object
         """
         super(InitXBee, self).setUp()
-        self.xbee = XBee(None)
+        self._patch_io = ioloop.IOLoop.current()
+        self._patch_io.add_handler = Mock()
+        serial_port = Serial()
+        self.xbee = XBee(serial_port, io_loop=self._patch_io)
 
 
 class TestBuildCommand(InitXBee):
@@ -462,7 +470,7 @@ def test_send_at_command(self):
         """
 
         serial_port = Serial()
-        xbee = XBee(serial_port)
+        xbee = XBee(serial_port, io_loop=self._patch_io)
 
         # Send an AT command
         xbee.send('at', frame_id=stringToBytes('A'),
@@ -480,7 +488,7 @@ def test_send_at_command_with_param(self):
         """
 
         serial_port = Serial()
-        xbee = XBee(serial_port)
+        xbee = XBee(serial_port, io_loop=self._patch_io)
 
         # Send an AT command
         xbee.send(
@@ -508,7 +516,7 @@ def setUp(self):
         """
         super(TestSendShorthand, self).setUp()
         self.ser = Serial()
-        self.xbee = XBee(self.ser)
+        self.xbee = XBee(self.ser, io_loop=self._patch_io)
 
     def test_send_at_command(self):
         """
@@ -552,7 +560,7 @@ def test_shorthand_disabled(self):
         When shorthand is disabled, any attempt at calling a
         non-existant attribute should raise AttributeError
         """
-        self.xbee = XBee(self.ser, shorthand=False)
+        self.xbee = XBee(self.ser, shorthand=False, io_loop=self._patch_io)
 
         try:
             self.xbee.at
@@ -575,7 +583,7 @@ def test_read_at(self):
         """
         device = Serial()
         device.set_read_data(b'\x7E\x00\x05\x88DMY\x01\x8c')
-        xbee = XBee(device)
+        xbee = XBee(device, io_loop=self._patch_io)
 
         xbee._process_input(None, None)
         info = yield xbee.wait_read_frame()
@@ -592,7 +600,7 @@ def test_read_at_params(self):
         """
         device = Serial()
         device.set_read_data(b'\x7E\x00\x08\x88DMY\x01\x00\x00\x00\x8c')
-        xbee = XBee(device)
+        xbee = XBee(device, io_loop=self._patch_io)
 
         xbee._process_input(None, None)
         info = yield xbee.wait_read_frame()
@@ -620,7 +628,7 @@ def test_is_response_parsed_as_io(self):
 
         device = Serial()
         device.set_read_data(APIFrame(data=b'\x88DIS\x00' + data).output())
-        xbee = XBee(device)
+        xbee = XBee(device, io_loop=self._patch_io)
 
         xbee._process_input(None, None)
         info = yield xbee.wait_read_frame()
@@ -655,7 +663,7 @@ def test_is_remote_response_parsed_as_io(self):
             data=b'\x97D\x00\x13\xa2\x00@oG\xe4v\x1aIS\x00' + data).output()
         )
 
-        xbee = XBee(device)
+        xbee = XBee(device, io_loop=self._patch_io)
 
         xbee._process_input(None, None)
         info = yield xbee.wait_read_frame()
@@ -693,7 +701,7 @@ def test_read_io_data(self):
 
         device = Serial()
         device.set_read_data(b'\x7E\x00\x0C' + rx_io_resp + data + b'\xfd')
-        xbee = XBee(device)
+        xbee = XBee(device, io_loop=self._patch_io)
 
         xbee._process_input(None, None)
         info = yield xbee.wait_read_frame()
@@ -737,7 +745,7 @@ def read(self, length=1):
                     return super(BadReadDevice, self).read()
 
         badDevice = BadReadDevice(1, b'\x7E\x00\x05\x88DMY\x01\x8c')
-        xbee = XBee(badDevice)
+        xbee = XBee(badDevice, io_loop=self._patch_io)
 
         try:
             xbee._process_input(None, None)
@@ -755,7 +763,7 @@ def test_read_at_params_in_escaped_mode(self):
         """
         device = Serial()
         device.set_read_data(b'~\x00\t\x88DMY\x01}^}]}1}3m')
-        xbee = XBee(device, escaped=True)
+        xbee = XBee(device, escaped=True, io_loop=self._patch_io)
 
         xbee._process_input(None, None)
         info = yield xbee.wait_read_frame()
@@ -773,7 +781,7 @@ def test_empty_frame_ignored(self):
         """
         device = Serial()
         device.set_read_data(b'\x7E\x00\x00\xFF\x7E\x00\x05\x88DMY\x01\x8c')
-        xbee = XBee(device)
+        xbee = XBee(device, io_loop=self._patch_io)
 
         xbee._process_input(None, None)
         xbee._process_input(None, None)
@@ -791,7 +799,7 @@ def test_read_rx_with_close_brace(self):
         """
         device = Serial()
         device.set_read_data(APIFrame(b'\x81\x01\x02\x55\x00{test=1}').output())
-        xbee = XBee(device)
+        xbee = XBee(device, io_loop=self._patch_io)
 
         xbee._process_input(None, None)
         info = yield xbee.wait_read_frame()
@@ -810,7 +818,7 @@ def test_read_rx_with_close_brace_escaped(self):
         device = Serial()
         device.set_read_data(APIFrame(b'\x81\x01\x02\x55\x00{test=1}',
                                       escaped=True).output())
-        xbee = XBee(device, escaped=True)
+        xbee = XBee(device, escaped=True, io_loop=self._patch_io)
 
         xbee._process_input(None, None)
         info = yield xbee.wait_read_frame()

xbee/tornado/tests/test_zigbee.py

@@ -7,11 +7,16 @@
 Tests the XBee ZB (ZigBee) implementation class for API compliance
 """
 import unittest
+try:
+    from unittest.mock import Mock
+except ImportError:
+    from mock import Mock
 from xbee.tornado import has_tornado
 
 if not has_tornado:
     raise unittest.SkipTest("Requires Tornado")
 
+from tornado import ioloop  # noqa
 from xbee.tornado.zigbee import ZigBee  # noqa
 from xbee.tests.Fake import Serial  # noqa
 
@@ -22,15 +27,18 @@ class TestZigBee(unittest.TestCase):
     """
 
     def setUp(self):
-        self.zigbee = ZigBee(None)
         super(TestZigBee, self).setUp()
+        self._patch_io = ioloop.IOLoop.current()
+        self._patch_io.add_handler = Mock()
+        serial_port = Serial()
+        self.zigbee = ZigBee(serial_port, io_loop=self._patch_io)
 
     def test_send(self):
         """
         Test send() with AT command.
         """
         device = Serial()
-        xbee = ZigBee(device)
+        xbee = ZigBee(device, io_loop=self._patch_io)
         xbee.send('at', command='MY')
         result = device.get_data_written()
         expected = b'~\x00\x04\x08\x01MYP'
@@ -236,8 +244,11 @@ class TestParseZigBeeIOData(unittest.TestCase):
     """
 
     def setUp(self):
-        self.zigbee = ZigBee(None)
         super(TestParseZigBeeIOData, self).setUp()
+        patch_io = ioloop.IOLoop.current()
+        patch_io.add_handler = Mock()
+        serial_port = Serial()
+        self.zigbee = ZigBee(serial_port, io_loop=patch_io)
 
     def test_parse_dio_adc(self):
             data = b'\x01\x08\x00\x0e\x08\x00\x00\x00\x02P\x02\x06'