New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
feat: make publish futures compatible with concurrent.futures.as_completed() #397
Merged
Merged
Changes from all commits
Commits
Show all changes
6 commits
Select commit
Hold shift + click to select a range
318676a
feat: make futures compatible with as_completed()
plamut 64d0d6c
Fix two unit tests in pre-Python 3.8
plamut 46b1607
Cover missing code line with a test
plamut bb3dd72
Use double underscore for internal cancelled flag
plamut 79f11b6
Prefix manager reference with double underscore
plamut 6c8b501
Remove Future's completed parameter altogether
plamut File filter
Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -14,173 +14,45 @@ | |
|
||
from __future__ import absolute_import | ||
|
||
import threading | ||
import uuid | ||
import concurrent.futures | ||
|
||
import google.api_core.future | ||
from google.cloud.pubsub_v1.publisher import exceptions | ||
|
||
|
||
class Future(google.api_core.future.Future): | ||
class Future(concurrent.futures.Future, google.api_core.future.Future): | ||
"""Encapsulation of the asynchronous execution of an action. | ||
|
||
This object is returned from asychronous Pub/Sub calls, and is the | ||
interface to determine the status of those calls. | ||
|
||
This object should not be created directly, but is returned by other | ||
methods in this library. | ||
|
||
Args: | ||
completed (Optional[Any]): An event, with the same interface as | ||
:class:`threading.Event`. This is provided so that callers | ||
with different concurrency models (e.g. ``threading`` or | ||
``multiprocessing``) can supply an event that is compatible | ||
with that model. The ``wait()`` and ``set()`` methods will be | ||
used. If this argument is not provided, then a new | ||
:class:`threading.Event` will be created and used. | ||
""" | ||
|
||
# This could be a sentinel object or None, but the sentinel object's ID | ||
# can change if the process is forked, and None has the possibility of | ||
# actually being a result. | ||
_SENTINEL = uuid.uuid4() | ||
|
||
def __init__(self, completed=None): | ||
self._result = self._SENTINEL | ||
self._exception = self._SENTINEL | ||
self._callbacks = [] | ||
if completed is None: | ||
completed = threading.Event() | ||
self._completed = completed | ||
|
||
def cancel(self): | ||
"""Actions in Pub/Sub generally may not be canceled. | ||
|
||
This method always returns False. | ||
""" | ||
return False | ||
|
||
def cancelled(self): | ||
"""Actions in Pub/Sub generally may not be canceled. | ||
|
||
This method always returns False. | ||
""" | ||
return False | ||
|
||
def running(self): | ||
"""Actions in Pub/Sub generally may not be canceled. | ||
"""Return ``True`` if the associated Pub/Sub action has not yet completed. | ||
|
||
Returns: | ||
bool: ``True`` if this method has not yet completed, or | ||
``False`` if it has completed. | ||
Returns: bool: | ||
""" | ||
return not self.done() | ||
|
||
def done(self): | ||
"""Return True the future is done, False otherwise. | ||
|
||
This still returns True in failure cases; checking :meth:`result` or | ||
:meth:`exception` is the canonical way to assess success or failure. | ||
""" | ||
return self._exception != self._SENTINEL or self._result != self._SENTINEL | ||
|
||
def result(self, timeout=None): | ||
"""Resolve the future and return a value where appropriate. | ||
|
||
Args: | ||
timeout (Union[int, float]): The number of seconds before this call | ||
times out and raises TimeoutError. | ||
|
||
Raises: | ||
concurrent.futures.TimeoutError: If the request times out. | ||
Exception: For undefined exceptions in the underlying | ||
call execution. | ||
""" | ||
# Attempt to get the exception if there is one. | ||
# If there is not one, then we know everything worked, and we can | ||
# return an appropriate value. | ||
err = self.exception(timeout=timeout) | ||
if err is None: | ||
return self._result | ||
raise err | ||
|
||
def exception(self, timeout=None): | ||
"""Return the exception raised by the call, if any. | ||
|
||
Args: | ||
timeout (Union[int, float]): The number of seconds before this call | ||
times out and raises TimeoutError. | ||
|
||
Raises: | ||
concurrent.futures.TimeoutError: If the request times out. | ||
|
||
Returns: | ||
Exception: The exception raised by the call, if any. | ||
""" | ||
# Wait until the future is done. | ||
if not self._completed.wait(timeout=timeout): | ||
raise exceptions.TimeoutError("Timed out waiting for result.") | ||
|
||
# If the batch completed successfully, this should return None. | ||
if self._result != self._SENTINEL: | ||
return None | ||
|
||
# Okay, this batch had an error; this should return it. | ||
return self._exception | ||
|
||
def add_done_callback(self, callback): | ||
"""Attach the provided callable to the future. | ||
|
||
The provided function is called, with this future as its only argument, | ||
when the future finishes running. | ||
|
||
Args: | ||
callback (Callable): The function to call. | ||
|
||
Returns: | ||
None | ||
""" | ||
if self.done(): | ||
return callback(self) | ||
self._callbacks.append(callback) | ||
def set_running_or_notify_cancel(self): | ||
raise NotImplementedError( | ||
"Only used by executors from `concurrent.futures` package." | ||
) | ||
|
||
def set_result(self, result): | ||
"""Set the result of the future to the provided result. | ||
"""Set the return value of work associated with the future. | ||
|
||
Args: | ||
result (Any): The result | ||
Do not use this method, it should only be used internally by the library and its | ||
unit tests. | ||
""" | ||
# Sanity check: A future can only complete once. | ||
if self.done(): | ||
raise RuntimeError("set_result can only be called once.") | ||
|
||
# Set the result and trigger the future. | ||
self._result = result | ||
self._trigger() | ||
return super().set_result(result=result) | ||
|
||
def set_exception(self, exception): | ||
"""Set the result of the future to the given exception. | ||
|
||
Args: | ||
exception (:exc:`Exception`): The exception raised. | ||
""" | ||
# Sanity check: A future can only complete once. | ||
if self.done(): | ||
raise RuntimeError("set_exception can only be called once.") | ||
|
||
# Set the exception and trigger the future. | ||
self._exception = exception | ||
self._trigger() | ||
|
||
def _trigger(self): | ||
"""Trigger all callbacks registered to this Future. | ||
|
||
This method is called internally by the batch once the batch | ||
completes. | ||
"""Set the result of the future as being the given exception. | ||
|
||
Args: | ||
message_id (str): The message ID, as a string. | ||
Do not use this method, it should only be used internally by the library and its | ||
unit tests. | ||
""" | ||
self._completed.set() | ||
for callback in self._callbacks: | ||
callback(self) | ||
return super().set_exception(exception=exception) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Why do we need this override? |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -25,6 +25,20 @@ class Future(futures.Future): | |
ID, unless an error occurs. | ||
""" | ||
|
||
def cancel(self): | ||
"""Actions in Pub/Sub generally may not be canceled. | ||
|
||
This method always returns ``False``. | ||
""" | ||
return False | ||
|
||
def cancelled(self): | ||
"""Actions in Pub/Sub generally may not be canceled. | ||
|
||
This method always returns ``False``. | ||
""" | ||
return False | ||
|
||
def result(self, timeout=None): | ||
"""Return the message ID or raise an exception. | ||
|
||
|
@@ -43,10 +57,4 @@ def result(self, timeout=None): | |
Exception: For undefined exceptions in the underlying | ||
call execution. | ||
""" | ||
# Attempt to get the exception if there is one. | ||
# If there is not one, then we know everything worked, and we can | ||
# return an appropriate value. | ||
err = self.exception(timeout=timeout) | ||
if err is None: | ||
return self._result | ||
raise err | ||
return super().result(timeout=timeout) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Why do we need this override? |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Why do we need this override?