Method getNextBatch

pubsub/pubsub.go:652–686

getNextBatch gets the next batch of messages from the server. It will return a channel that will itself return the messages as they come from each independent batch, or an operation error

(nMessages int)

// getNextBatch gets the next batch of messages from the server. It will return a channel that will itself return the
// messages as they come from each independent batch, or an operation error
func (s *Subscription) getNextBatch(nMessages int) chan msgsOrError {
	// Split nMessages into batches based on recvBatchOpts; we'll make a
	// separate ReceiveBatch call for each batch, and aggregate the results in
	// msgs.
	batches := batcher.Split(nMessages, s.recvBatchOpts)
	result := make(chan msgsOrError, len(batches))
	g, ctx := errgroup.WithContext(s.backgroundCtx)
	for _, maxMessagesInBatch := range batches {
		// Make a copy of the loop variable since it will be used by a goroutine.
		curMaxMessagesInBatch := maxMessagesInBatch
		g.Go(func() error {
			var msgs []*driver.Message
			err := retry.Call(ctx, gax.Backoff{}, s.driver.IsRetryable, func() error {
				var err error
				spanCtx, span := s.tracer.Start(ctx, "driver.Subscription.ReceiveBatch")
				defer func() { s.tracer.End(spanCtx, span, err) }()
				msgs, err = s.driver.ReceiveBatch(spanCtx, curMaxMessagesInBatch)
				return err
			})
			if err != nil {
				return wrapError(s.driver, err)
			}
			result <- msgsOrError{msgs: msgs}
			return nil
		})
	}
	go func() {
		// wait on group completion on the background and proper channel closing
		if err := g.Wait(); err != nil {
			result <- msgsOrError{err: err}
		}
		close(result)
	}()
	return result
}

var errSubscriptionShutdown = gcerr.Newf(gcerr.FailedPrecondition, nil, "pubsub: Subscription has been Shutdown")