Merge pull request #64 from niolabs/add-timeout

Author: jamesleesaunders

.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/backend/base.py

@@ -18,6 +18,9 @@
 class CommandFrameException(KeyError):
     pass
 
+class TimeoutException(Exception):
+    pass
+
 
 class XBeeBase(object):
     """

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/thread/base.py

@@ -13,6 +13,7 @@
 """
 from xbee.frame import APIFrame
 from xbee.backend.base import XBeeBase as _XBeeBase
+from xbee.backend.base import TimeoutException as _TimeoutException
 import threading
 import time
 
@@ -93,7 +94,7 @@ def run(self):
                 if self._error_callback:
                     self._error_callback(e)
 
-    def wait_read_frame(self):
+    def wait_read_frame(self, timeout=None):
         """
         wait_read_frame: None -> frame info dictionary
 
@@ -102,10 +103,10 @@ def wait_read_frame(self):
         wait_read_frame attempts to parse the data contained within it
         and returns the resulting dictionary
         """
-        frame = self._wait_for_frame()
+        frame = self._wait_for_frame(timeout)
         return self._split_response(frame.data)
 
-    def _wait_for_frame(self):
+    def _wait_for_frame(self, timeout=None):
         """
         _wait_for_frame: None -> binary data
 
@@ -119,11 +120,17 @@ def _wait_for_frame(self):
         """
         frame = APIFrame(escaped=self._escaped)
 
+        deadline = 0
+        if timeout is not None and timeout > 0:
+            deadline = time.time() + timeout
+
         while True:
                 if self._callback and not self._thread_continue:
                     raise ThreadQuitException
 
                 if self.serial.inWaiting() == 0:
+                    if deadline and time.time() > deadline:
+                        raise _TimeoutException
                     time.sleep(.01)
                     continue
 

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'