Function openSubscription

pubsub/gcppubsubv2/gcppubsub.go:506–525

openSubscription returns a driver.Subscription.

(subscriber *raw.Subscriber, opts *SubscriptionOptions)

// openSubscription returns a driver.Subscription.
func openSubscription(subscriber *raw.Subscriber, opts *SubscriptionOptions) *subscription {
	if opts == nil {
		opts = &SubscriptionOptions{}
	}
	if opts.MaxBatchSize == 0 {
		opts.MaxBatchSize = defaultRecvBatcherOpts.MaxBatchSize
	}
	// Construct a context that's used (repeatedly if necessary) to call Receive.
	// It never expires; the Receive may last forever.
	// The cancel function is used during shutdown.
	ctx, cancel := context.WithCancel(context.Background())
	return &subscription{
		subscriber:    subscriber,
		options:       opts,
		receiving:     false,
		receiveCtx:    ctx,
		receiveCancel: cancel,
		acks:          map[driver.AckID]*ackableMsg{},
	}
}

// ReceiveBatch implements driver.Subscription.ReceiveBatch.
func (s *subscription) ReceiveBatch(ctx context.Context, maxMessages int) ([]*driver.Message, error) {