Executing only one pubsub callback concurrently

[philipp-eisen]

I have a problem, where I depend on stateful externals resources, that require me to only process one message at a time.

I was trying to accomplish that by setting the max_messages in the FlowControl to 1. However, the callbacks were still executed concurrently.

So I dug a bit deeper and found this commit. I thought I could pass an event and use that to block the thread until the execution of the callback is done.

However, it was not entirely clear to me how this is meant to be use. The only place I found, where an event can be passed is the constructor of the future. However, the docstring discourages to use this class.

So, is the intended way to use this, to implement a new policy? Because this is the place, where I would have assumed the event to be passed through.

If I'm completely off track and if there's some completely different way of going about this, please let me know.

I appreciate you work!

[dhermes]

@ph1l337 I recommend just using a lock in your callback:

class OnlyOne(object):

    def __init__(self):
        self.lock = threading.Lock()

    def __call__(self, message):
        with self.lock:
            return process_message(message)


callback = OnlyOne()
subscription.open(callback)

[dhermes]

I am going to pre-emptively close this since I think my recommendation works. Even if it needs tweaking, it sounds like your use case can be supported without any change to the actual google-cloud-pubsub package.

[philipp-eisen]

Thanks. @dhermes. Actually thought about that, but had some other error in my implementation. Do you think this could be useful to add to the documentation or is it to specific? If you point me to the place, where this would go I could write it and make PR.

[tseaver]

@ph1l337 Thanks for offering to improve our docs! I think the relevant file is the subscriber index, which has a section on "Subscription Callbacks".