Break out of PubSub Subscription.Receive

[mbinette]

I've followed the basic examples of PubSub Subscription.Receive and it's working for me except I can't figure out how to get it to exit after all messages have been received.

I am writing a simple cli application that should connect, receive and process all messages, then exit if no messages are left. I saw samples on exiting after receiving a certain amount of messages, say 10, but I have no idea how many will be waiting for me when I call Receive and want to process all of them and then exit.

I tried context.WithTimeout but I can't see a way to reset the timeout during each of me Receive calls that process the message. If I could reset the timeout then I could reset it to something short like 30 seconds on each call so that the application would exit 30 seconds after the last message is processed.

I really don't want to drop down to the apiv1 code if I don't have to.

[jba]

As you suggest with your discussion of timeout, there's no clear meaning to "all the messages." As long as the subscription exists, it could receive messages.

You could also structure your system to publish an "I'm done" message, but the problem there is that PubSub doesn't guarantee delivery order, so it gets complicated.

I think your timeout idea is a good one. Here's some sample code (untested). The idea is that we write to a channel (non-blocking) if we've seen something, and another goroutine receives from that channel. If it doesn't hear anything after the timeout, it cancels the Receive's context.

func receiveWithTimeout(ctx context.Context, timeout time.Duration) error {
 	seen := make(chan int, 1)
 	ctx2, cancel := context.WithCancel(ctx)
 	defer cancel()
	go func() {
		for {
			select {
			case <-seen:
			case <-time.After(timeout):
				cancel()
				return
			}
		}
	}()
	return mySubscription.Receive(ctx2, func(c context.Context, m *pubsub.Message) {
		select {
		case seen <- 1:
		default:
		}
		// do work
    })
} 

You can call that in a loop.

[mbinette]

That works great. Thank You!

I'm trying to make sure I can handle errors in processing my messages as well. So instead of calling message.Nack() on error I simply return so that the message uses the MaxExtension to timeout later rather than immediately. The purpose is to process all messages, leave any message that caused an error in the subscription, then exit the app at the end of processing.

Your code lets me do that as long as my timeout value is less than my MaxExtension. Messages are only added once a day on a schedule and I'll be able to monitor errors separately. I can't lose the message.

I will most likely need to implement separate logic that moves the message to another queue on error at some point.

[jba]

Glad that works.

But due to a recent change, your strategy of ignoring the message will reduce your throughput. The reason is that until you ack or nack, our flow controller counts the message as active, and you can only have MaxOutstandingMessages/Bytes active at once.

See #870 for a solution.