summary refs log tree commit diff
path: root/synapse/util/async_helpers.py
diff options
context:
space:
mode:
authorHubert Chathi <hubert@uhoreg.ca>2019-10-30 10:17:10 -0400
committerHubert Chathi <hubert@uhoreg.ca>2019-10-30 10:17:10 -0400
commit7d7eac61be337b7a3ed7274a13e131f41d0966d9 (patch)
treee6f70c499902b00ed0450e24bb4132e3079698e6 /synapse/util/async_helpers.py
parentblack (diff)
parentMerge pull request #6240 from matrix-org/erikj/split_out_persistence_store (diff)
downloadsynapse-7d7eac61be337b7a3ed7274a13e131f41d0966d9.tar.xz
Merge branch 'develop' into cross-signing_federation
Diffstat (limited to 'synapse/util/async_helpers.py')
-rw-r--r--synapse/util/async_helpers.py16
1 files changed, 8 insertions, 8 deletions
diff --git a/synapse/util/async_helpers.py b/synapse/util/async_helpers.py
index 804dbca443..b60a604474 100644
--- a/synapse/util/async_helpers.py
+++ b/synapse/util/async_helpers.py
@@ -86,11 +86,12 @@ class ObservableDeferred(object):
 
         deferred.addCallbacks(callback, errback)
 
-    def observe(self):
+    def observe(self) -> defer.Deferred:
         """Observe the underlying deferred.
 
-        Can return either a deferred if the underlying deferred is still pending
-        (or has failed), or the actual value. Callers may need to use maybeDeferred.
+        This returns a brand new deferred that is resolved when the underlying
+        deferred is resolved. Interacting with the returned deferred does not
+        effect the underdlying deferred.
         """
         if not self._result:
             d = defer.Deferred()
@@ -105,7 +106,7 @@ class ObservableDeferred(object):
             return d
         else:
             success, res = self._result
-            return res if success else defer.fail(res)
+            return defer.succeed(res) if success else defer.fail(res)
 
     def observers(self):
         return self._observers
@@ -138,7 +139,7 @@ def concurrently_execute(func, args, limit):
     the number of concurrent executions.
 
     Args:
-        func (func): Function to execute, should return a deferred.
+        func (func): Function to execute, should return a deferred or coroutine.
         args (list): List of arguments to pass to func, each invocation of func
             gets a signle argument.
         limit (int): Maximum number of conccurent executions.
@@ -148,11 +149,10 @@ def concurrently_execute(func, args, limit):
     """
     it = iter(args)
 
-    @defer.inlineCallbacks
-    def _concurrently_execute_inner():
+    async def _concurrently_execute_inner():
         try:
             while True:
-                yield func(next(it))
+                await maybe_awaitable(func(next(it)))
         except StopIteration:
             pass