diff options
| -rw-r--r-- | synapse/rest/client/transactions.py | 24 | ||||
| -rw-r--r-- | synapse/rest/client/v1/base.py | 2 | ||||
| -rw-r--r-- | synapse/rest/client/v2_alpha/sendtodevice.py | 2 | ||||
| -rw-r--r-- | synapse/util/__init__.py | 10 | ||||
| -rw-r--r-- | tests/rest/client/test_transactions.py | 69 |
5 files changed, 99 insertions, 8 deletions
diff --git a/synapse/rest/client/transactions.py b/synapse/rest/client/transactions.py index 8d69e12d36..351170edbc 100644 --- a/synapse/rest/client/transactions.py +++ b/synapse/rest/client/transactions.py @@ -41,12 +41,19 @@ def get_transaction_key(request): return request.path + "/" + token +CLEANUP_PERIOD_MS = 1000 * 60 * 30 # 30 mins + + class HttpTransactionCache(object): - def __init__(self): + def __init__(self, clock): + self.clock = clock self.transactions = { - # $txn_key: ObservableDeferred<(res_code, res_json_body)> + # $txn_key: (ObservableDeferred<(res_code, res_json_body)>, timestamp) } + # Try to clean entries every 30 mins. This means entries will exist + # for at *LEAST* 30 mins, and at *MOST* 60 mins. + self.cleaner = self.clock.looping_call(self._cleanup, CLEANUP_PERIOD_MS) def fetch_or_execute_request(self, request, fn, *args, **kwargs): """A helper function for fetch_or_execute which extracts @@ -74,11 +81,18 @@ class HttpTransactionCache(object): Deferred which resolves to a tuple of (response_code, response_dict). """ try: - return self.transactions[txn_key].observe() - except KeyError: + return self.transactions[txn_key][0].observe() + except (KeyError, IndexError): pass # execute the function instead. deferred = fn(*args, **kwargs) observable = ObservableDeferred(deferred) - self.transactions[txn_key] = observable + self.transactions[txn_key] = (observable, self.clock.time_msec()) return observable.observe() + + def _cleanup(self): + now = self.clock.time_msec() + for key in self.transactions.keys(): + ts = self.transactions[key][1] + if now > (ts + CLEANUP_PERIOD_MS): # after cleanup period + del self.transactions[key] diff --git a/synapse/rest/client/v1/base.py b/synapse/rest/client/v1/base.py index 07ff5b218c..c7aa0bbf59 100644 --- a/synapse/rest/client/v1/base.py +++ b/synapse/rest/client/v1/base.py @@ -60,4 +60,4 @@ class ClientV1RestServlet(RestServlet): self.hs = hs self.builder_factory = hs.get_event_builder_factory() self.auth = hs.get_v1auth() - self.txns = HttpTransactionCache() + self.txns = HttpTransactionCache(hs.get_clock()) diff --git a/synapse/rest/client/v2_alpha/sendtodevice.py b/synapse/rest/client/v2_alpha/sendtodevice.py index 2187350d42..ac660669f3 100644 --- a/synapse/rest/client/v2_alpha/sendtodevice.py +++ b/synapse/rest/client/v2_alpha/sendtodevice.py @@ -40,7 +40,7 @@ class SendToDeviceRestServlet(servlet.RestServlet): super(SendToDeviceRestServlet, self).__init__() self.hs = hs self.auth = hs.get_auth() - self.txns = HttpTransactionCache() + self.txns = HttpTransactionCache(hs.get_clock()) self.device_message_handler = hs.get_device_message_handler() def on_PUT(self, request, message_type, txn_id): diff --git a/synapse/util/__init__.py b/synapse/util/__init__.py index 2b3f0bef3c..c05b9450be 100644 --- a/synapse/util/__init__.py +++ b/synapse/util/__init__.py @@ -34,7 +34,7 @@ class Clock(object): """A small utility that obtains current time-of-day so that time may be mocked during unit-tests. - TODO(paul): Also move the sleep() functionallity into it + TODO(paul): Also move the sleep() functionality into it """ def time(self): @@ -46,6 +46,14 @@ class Clock(object): return int(self.time() * 1000) def looping_call(self, f, msec): + """Call a function repeatedly. + + Waits `msec` initially before calling `f` for the first time. + + Args: + f(function): The function to call repeatedly. + msec(float): How long to wait between calls in milliseconds. + """ l = task.LoopingCall(f) l.start(msec / 1000.0, now=False) return l diff --git a/tests/rest/client/test_transactions.py b/tests/rest/client/test_transactions.py new file mode 100644 index 0000000000..d7cea30260 --- /dev/null +++ b/tests/rest/client/test_transactions.py @@ -0,0 +1,69 @@ +from synapse.rest.client.transactions import HttpTransactionCache +from synapse.rest.client.transactions import CLEANUP_PERIOD_MS +from twisted.internet import defer +from mock import Mock, call +from tests import unittest +from tests.utils import MockClock + + +class HttpTransactionCacheTestCase(unittest.TestCase): + + def setUp(self): + self.clock = MockClock() + self.cache = HttpTransactionCache(self.clock) + + self.mock_http_response = (200, "GOOD JOB!") + self.mock_key = "foo" + + @defer.inlineCallbacks + def test_executes_given_function(self): + cb = Mock( + return_value=defer.succeed(self.mock_http_response) + ) + res = yield self.cache.fetch_or_execute( + self.mock_key, cb, "some_arg", keyword="arg" + ) + cb.assert_called_once_with("some_arg", keyword="arg") + self.assertEqual(res, self.mock_http_response) + + @defer.inlineCallbacks + def test_deduplicates_based_on_key(self): + cb = Mock( + return_value=defer.succeed(self.mock_http_response) + ) + for i in range(3): # invoke multiple times + res = yield self.cache.fetch_or_execute( + self.mock_key, cb, "some_arg", keyword="arg", changing_args=i + ) + self.assertEqual(res, self.mock_http_response) + # expect only a single call to do the work + cb.assert_called_once_with("some_arg", keyword="arg", changing_args=0) + + @defer.inlineCallbacks + def test_cleans_up(self): + cb = Mock( + return_value=defer.succeed(self.mock_http_response) + ) + yield self.cache.fetch_or_execute( + self.mock_key, cb, "an arg" + ) + # should NOT have cleaned up yet + self.clock.advance_time_msec(CLEANUP_PERIOD_MS / 2) + + yield self.cache.fetch_or_execute( + self.mock_key, cb, "an arg" + ) + # still using cache + cb.assert_called_once_with("an arg") + + self.clock.advance_time_msec(CLEANUP_PERIOD_MS) + + yield self.cache.fetch_or_execute( + self.mock_key, cb, "an arg" + ) + # no longer using cache + self.assertEqual(cb.call_count, 2) + self.assertEqual( + cb.call_args_list, + [call("an arg",), call("an arg",)] + ) |