Method Shutdown

pubsub/pubsub.go:691–725

Shutdown flushes pending ack sends and disconnects the Subscription.

(ctx context.Context)

// Shutdown flushes pending ack sends and disconnects the Subscription.
func (s *Subscription) Shutdown(ctx context.Context) (err error) {
	ctx, span := s.tracer.Start(ctx, "Subscription.Shutdown")
	defer func() { s.tracer.End(ctx, span, err) }()

	s.mu.Lock()
	if errors.Is(s.err, errSubscriptionShutdown) {
		// Already Shutdown.
		defer s.mu.Unlock()
		return s.err
	}
	s.err = errSubscriptionShutdown
	s.mu.Unlock()
	c := make(chan struct{})
	go func() {
		defer close(c)
		if s.ackBatcher != nil {
			s.ackBatcher.Shutdown()
		}
	}()
	select {
	case <-ctx.Done():
	case <-c:
	}
	s.cancel()
	if err := s.driver.Close(); err != nil {
		return wrapError(s.driver, err)
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if err := s.unreportedAckErr; err != nil {
		s.unreportedAckErr = nil
		return err
	}
	return ctx.Err()
}

// As converts i to driver-specific types.
// See https://gocloud.dev/concepts/as/ for background information, the "As"