Add futures for subscriptions.

[lukesneeringer]

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.

[theacodes]

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)
defcallback(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]

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.

[theacodes]

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

[lukesneeringer]

Maybe.

[lukesneeringer]

@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]

@lukesneeringer totally fair

[theacodes]

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