Add futures for subscriptions. #4265

lukesneeringer · 2017-10-26T18:19:35Z

This causes the .open method on subscriptions to return a future, which can be used to block the main thread or trap exceptions.

Fixes #3888.
Fixes #4019.

@jonparrott A couple items we should nail down and possibly tweak before merging this:

The current interface has a three-object pass to start consuming messages: instantiate a client, instantiate a subscription, then call .open on the subscription, which now returns a future. However, we currently have some shortcut logic that lets you pass a callback on step 2. You still get the subscription back, but .open is called implicitly; there's no good place for me to return the future in that case.
- This is kind of a big deal; this future is basically the way to both block and handle errata.
- On the other hand, maybe this is fine; .future is a public property.
  - At the very minimum, I should update all documentation to show the full baton pass and highlight the future and what it is for.
There is not really any good way to stop consuming automatically. I can not do it in on_exception because that runs in one of the threads that the helper_threads.stop_all() method is trying to burn down, so it raises RuntimeError.
- I could have an .add_done_callback call in Future.__init__ that adds a callback that does the right thing, but this feels like an interface violation to me.
- Maybe this is fine too; we (a) leave it up to users to decide how to handle errors, albeit (b) also fix bugs like Pub/Sub Subscriber does not catch & retry UNAVAILABLE errors #4234 so they are far less common.

Holding off on updating tests and docs until we make a decision.

This causes the `.open` method on subscriptions to return a future, which can be used to block the main thread or trap exceptions.

theacodes · 2017-10-26T18:28:53Z

On the other hand, maybe this is fine; .future is a public property.

Seems fine:

    subscriber = pubsub_v1.SubscriberClient()
    subscription_path = subscriber.subscription_path(
        project, subscription_name)

    def callback(message):
        print('Received message: {}'.format(message))
        message.ack()

    subcription = subscriber.subscribe(subscription_path, callback=callback)

    # The subscriber is non-blocking, so we must keep the main thread from
    # exiting to allow it to process messages in the background.
    print('Listening for messages on {}'.format(subscription_path))
    subscription.future.result()

There is not really any good way to stop consuming automatically.

doesn't .close() cover this? Or am I missing something?

lukesneeringer · 2017-10-26T18:30:09Z

doesn't .close() cover this? Or am I missing something?

It does, but if you call it in .on_exception, you get a RuntimeError because a thread tries to join itself.

pubsub/google/cloud/pubsub_v1/publisher/batch/thread.py


        # These objects are all communicated between threads; ensure that
        # any writes to them are atomic.
+        # import pdb ; pdb.set_trace()


theacodes · 2017-10-26T19:09:09Z

Can we have close() send a signal to exit if it's being invoked from the "wrong" thread?

lukesneeringer · 2017-10-26T19:16:41Z

Maybe.

lukesneeringer · 2017-10-26T20:08:10Z

@jonparrott This is ready for review. I am going to punt on .close() for now because I think it is more valuable to push this as is and spend cycles on a couple of the other bugs. I am also not 100% sure that .close() is even the right behavior.

theacodes · 2017-10-26T20:10:04Z

@lukesneeringer totally fair

theacodes

Looks mostly fine, but some concerns about making the future public.