WebSocket Server Connection Timeout Implementation and Testing

Question

I am implementing a WebSockets server in Tornado 3.2. The client connecting to the server will not be a browser.

In cases where there is feedback between the server and the client, I would like to add max. the time when the server will wait for the client to respond before closing the connection.

This is roughly what I tried:

import datetime import tornado class WSHandler(WebSocketHandler): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.timeout = None def _close_on_timeout(self): if self.ws_connection: self.close() def open(self): initialize() def on_message(self, message): # Remove previous timeout, if one exists. if self.timeout: tornado.ioloop.IOLoop.instance().remove_timeout(self.timeout) self.timeout = None if is_last_message: self.write_message(message) self.close() else: # Add a new timeout. self.timeout = tornado.ioloop.IOLoop.instance().add_timeout( datetime.timedelta(milliseconds=1000), self._close_on_timeout) self.write_message(message) 

I am klutz and is there a much easier way to do this? I can't even schedule a simple print statement through add_timeout above.

I also need help with testing. This is what I have so far:

 from tornado.websocket import websocket_connect from tornado.testing import AsyncHTTPTestCase, gen_test import time class WSTests(AsyncHTTPTestCase): @gen_test def test_long_response(self): ws = yield websocket_connect('ws://address', io_loop=self.io_loop) # First round trip. ws.write_message('First message.') result = yield ws.read_message() self.assertEqual(result, 'First response.') # Wait longer than the timeout. # The test is in its own IOLoop, so a blocking sleep should be okay? time.sleep(1.1) # Expect either write or read to fail because of a closed socket. ws.write_message('Second message.') result = yield ws.read_message() self.assertNotEqual(result, 'Second response.') 

The client has no problems with writing and reading from the socket. This is probably due to the fact that add_timeout does not start.

Is a test required to somehow allow the execution of a timeout callback on the server? I would not have thought, as the docs say the tests run in their own IOLoop.

Edit

This is a working version with Ben's suggestions.

 import datetime import tornado class WSHandler(WebSocketHandler): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.timeout = None def _close_on_timeout(self): if self.ws_connection: self.close() def open(self): initialize() def on_message(self, message): # Remove previous timeout, if one exists. if self.timeout: tornado.ioloop.IOLoop.current().remove_timeout(self.timeout) self.timeout = None if is_last_message: self.write_message(message) self.close() else: # Add a new timeout. self.timeout = tornado.ioloop.IOLoop.current().add_timeout( datetime.timedelta(milliseconds=1000), self._close_on_timeout) self.write_message(message) 

Test:

 from tornado.websocket import websocket_connect from tornado.testing import AsyncHTTPTestCase, gen_test import time class WSTests(AsyncHTTPTestCase): @gen_test def test_long_response(self): ws = yield websocket_connect('ws://address', io_loop=self.io_loop) # First round trip. ws.write_message('First message.') result = yield ws.read_message() self.assertEqual(result, 'First response.') # Wait a little more than the timeout. yield gen.Task(self.io_loop.add_timeout, datetime.timedelta(seconds=1.1)) # Expect either write or read to fail because of a closed socket. ws.write_message('Second message.') result = yield ws.read_message() self.assertEqual(result, None) 

Answer 1

In your first example, the timeout processing code looks like.

For testing, each test case gets its own IOLoop, but for the test and everything else, it has only one IOLoop, so you should use add_timeout instead of time.sleep () here to avoid server blocking.

Answer 2

Hey Ben, I know this issue has long been resolved, but I wanted to share with any user reading this decision I made for this. This is mostly based on yours, but it solves a problem from an external service that can be easily integrated into any website using composition instead of inheritance:

 class TimeoutWebSocketService(): _default_timeout_delta_ms = 10 * 60 * 1000 # 10 min def __init__(self, websocket, ioloop=None, timeout=None): # Timeout self.ioloop = ioloop or tornado.ioloop.IOLoop.current() self.websocket = websocket self._timeout = None self._timeout_delta_ms = timeout or TimeoutWebSocketService._default_timeout_delta_ms def _close_on_timeout(self): self._timeout = None if self.websocket.ws_connection: self.websocket.close() def refresh_timeout(self, timeout=None): timeout = timeout or self._timeout_delta_ms if timeout > 0: # Clean last timeout, if one exists self.clean_timeout() # Add a new timeout (must be None from clean). self._timeout = self.ioloop.add_timeout( datetime.timedelta(milliseconds=timeout), self._close_on_timeout) def clean_timeout(self): if self._timeout is not None: # Remove previous timeout, if one exists. self.ioloop.remove_timeout(self._timeout) self._timeout = None 

To use the service is as simple as creating a new instance of TimeoutWebService (optionally with a timeout in ms, as well as ioloop where it should be executed) and calling the 'refresh_timeout' method to set the timeout for the first time or reset an existing time or “clean_timeout” to stop the timeout service.

 class BaseWebSocketHandler(WebSocketHandler): def prepare(self): self.timeout_service = TimeoutWebSocketService(timeout=(1000*60)) ## Optionally starts the service here self.timeout_service.refresh_timeout() ## rest of prepare method def on_message(self): self.timeout_service.refresh_timeout() def on_close(self): self.timeout_service.clean_timeout() 

Thanks to this approach, you can control exactly when and under what conditions you want to restart the timeout, which may differ from application to application. As an example, you can only update the timeout if the user fulfills the X conditions or if the message is expected.

Hope ppl will like this solution!