Move .observe() up to the cache to make things neater

synapse/rest/client/transactions.py

@@ -67,19 +67,18 @@ def fetch_or_execute(self, txn_key, fn, *args, **kwargs):
             txn_key (str): A key to ensure idempotency should fetch_or_execute be
             called again at a later point in time.
             fn (function): A function which returns a tuple of
-            (response_code, response_dict)d
+            (response_code, response_dict).
             *args: Arguments to pass to fn.
             **kwargs: Keyword arguments to pass to fn.
         Returns:
-            synapse.util.async.ObservableDeferred which resolves to a tuple
-            of (response_code, response_dict).
+            Deferred which resolves to a tuple of (response_code, response_dict).
         """
         try:
-            return self.transactions[txn_key]
+            return self.transactions[txn_key].observe()
         except KeyError:
             pass  # execute the function instead.
 
         deferred = fn(*args, **kwargs)
         observable = ObservableDeferred(deferred)
         self.transactions[txn_key] = observable
-        return observable
+        return observable.observe()

synapse/rest/client/v1/room.py

@@ -53,13 +53,10 @@ def register(self, http_server):
                                    client_path_patterns("/createRoom(?:/.*)?$"),
                                    self.on_OPTIONS)
 
-    @defer.inlineCallbacks
     def on_PUT(self, request, txn_id):
-        observable = self.txns.fetch_or_execute_request(
+        return self.txns.fetch_or_execute_request(
             request, self.on_POST, request
         )
-        res = yield observable.observe()
-        defer.returnValue(res)
 
     @defer.inlineCallbacks
     def on_POST(self, request):
@@ -208,13 +205,10 @@ def on_POST(self, request, room_id, event_type, txn_id=None):
     def on_GET(self, request, room_id, event_type, txn_id):
         return (200, "Not implemented")
 
-    @defer.inlineCallbacks
     def on_PUT(self, request, room_id, event_type, txn_id):
-        observable = self.txns.fetch_or_execute_request(
+        return self.txns.fetch_or_execute_request(
             request, self.on_POST, request, room_id, event_type, txn_id
         )
-        res = yield observable.observe()
-        defer.returnValue(res)
 
 
 # TODO: Needs unit testing for room ID + alias joins
@@ -271,13 +265,10 @@ def on_POST(self, request, room_identifier, txn_id=None):
 
         defer.returnValue((200, {"room_id": room_id}))
 
-    @defer.inlineCallbacks
     def on_PUT(self, request, room_identifier, txn_id):
-        observable = self.txns.fetch_or_execute_request(
+        return self.txns.fetch_or_execute_request(
             request, self.on_POST, request, room_identifier, txn_id
         )
-        res = yield observable.observe()
-        defer.returnValue(res)
 
 
 # TODO: Needs unit testing
@@ -519,13 +510,10 @@ def on_POST(self, request, room_id, txn_id=None):
 
         defer.returnValue((200, {}))
 
-    @defer.inlineCallbacks
     def on_PUT(self, request, room_id, txn_id):
-        observable = self.txns.fetch_or_execute_request(
+        return self.txns.fetch_or_execute_request(
             request, self.on_POST, request, room_id, txn_id
         )
-        res = yield observable.observe()
-        defer.returnValue(res)
 
 
 # TODO: Needs unit testing
@@ -597,13 +585,10 @@ def _has_3pid_invite_keys(self, content):
                 return False
         return True
 
-    @defer.inlineCallbacks
     def on_PUT(self, request, room_id, membership_action, txn_id):
-        observable = self.txns.fetch_or_execute_request(
+        return self.txns.fetch_or_execute_request(
             request, self.on_POST, request, room_id, membership_action, txn_id
         )
-        res = yield observable.observe()
-        defer.returnValue(res)
 
 
 class RoomRedactEventRestServlet(ClientV1RestServlet):
@@ -635,13 +620,10 @@ def on_POST(self, request, room_id, event_id, txn_id=None):
 
         defer.returnValue((200, {"event_id": event.event_id}))
 
-    @defer.inlineCallbacks
     def on_PUT(self, request, room_id, event_id, txn_id):
-        observable = self.txns.fetch_or_execute_request(
+        return self.txns.fetch_or_execute_request(
             request, self.on_POST, request, room_id, event_id, txn_id
         )
-        res = yield observable.observe()
-        defer.returnValue(res)
 
 
 class RoomTypingRestServlet(ClientV1RestServlet):

synapse/rest/client/v2_alpha/sendtodevice.py

@@ -43,13 +43,10 @@ def __init__(self, hs):
         self.txns = HttpTransactionCache()
         self.device_message_handler = hs.get_device_message_handler()
 
-    @defer.inlineCallbacks
     def on_PUT(self, request, message_type, txn_id):
-        observable = self.txns.fetch_or_execute_request(
+        return self.txns.fetch_or_execute_request(
             request, self._put, request, message_type, txn_id
         )
-        res = yield observable.observe()
-        defer.returnValue(res)
 
     @defer.inlineCallbacks
     def _put(self, request, message_type, txn_id):