Skip to content

feat(pubsub): add stop method #9365

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.

Sign up for GitHub

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

Merged
pradn merged 15 commits into from
Nov 7, 2019
Merged

feat(pubsub): add stop method #9365

Changes from 1 commit
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
dad43aa
feat(pubsub): add stop method
Oct 1, 2019
ec7677c
feat(pubsub): add unit-test for stop method
Oct 1, 2019
93b5ec8
Add comment for _futures. Change wait() conditions.
Oct 22, 2019
a0d82e6
Move attribute comment
Oct 23, 2019
210abf0
Make stop() method non-blocking.
Oct 25, 2019
99bfc26
Add one more assert for stop() method.
Oct 25, 2019
3216d84
Fix comment.
Oct 25, 2019
94d3023
Comment fix.
Oct 25, 2019
3f53333
Add stopping lock.
Oct 28, 2019
b63c176
Spelling mistake fix.
Oct 29, 2019
3ef412c
Preventing race conditions while stopping publisher.
Nov 4, 2019
b1d515d
Fix test.
Nov 5, 2019
2534467
Lock method stop() completely.
Nov 6, 2019
45a6bf8
Small refactor.
Nov 7, 2019
ba01ec1
Update pubsub/google/cloud/pubsub_v1/publisher/client.py
Nov 7, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions pubsub/google/cloud/pubsub_v1/publisher/_batch/thread.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,9 @@ def __init__(self, client, topic, settings, autocommit=True):
# These members are all communicated between threads; ensure that
# any writes to them use the "state lock" to remain atomic.
self._futures = []
# _futures list should remain unchanged after batch
# status changed from ACCEPTING_MESSAGES to any other
# in order to avoid race conditions
self._messages = []
self._size = 0
self._status = base.BatchStatus.ACCEPTING_MESSAGES
Expand Down Expand Up @@ -138,13 +141,13 @@ def status(self):
return self._status

def wait(self):
"""If commit in progress, waits until all of the futures resolved.
"""If commit is in progress, waits until all of the futures resolved.

.. note::

This method blocks until all futures of this batch resolved.
"""
if self._status in (base.BatchStatus.STARTING, base.BatchStatus.IN_PROGRESS):
if self._status != base.BatchStatus.ACCEPTING_MESSAGES:
for future in self._futures:
future.result()

Expand Down