prow/pubsub/subscriber/server.go
func (s *PullServer) handlePulls(ctx context.Context, projectSubscriptions config.PubsubSubscriptions) (*errgroup.Group, context.Context, error) {
	// Since config might change we need be able to cancel the current run
	errGroup, derivedCtx := errgroup.WithContext(ctx)
	for project, subscriptions := range projectSubscriptions {
		client, err := s.Client.new(ctx, project)
		if err != nil {
			return errGroup, derivedCtx, err
		}
		for _, subName := range subscriptions {
			sub := client.subscription(subName)
			errGroup.Go(func() error {
				logrus.Infof("Listening for subscription %s on project %s", sub.string(), project)
				defer logrus.Warnf("Stopped Listening for subscription %s on project %s", sub.string(), project)
				err := sub.receive(derivedCtx, func(ctx context.Context, msg messageInterface) {
					if err = s.Subscriber.handleMessage(msg, sub.string()); err != nil {
						s.Subscriber.Metrics.ACKMessageCounter.With(prometheus.Labels{subscriptionLabel: sub.string()}).Inc()
					} else {
						s.Subscriber.Metrics.NACKMessageCounter.With(prometheus.Labels{subscriptionLabel: sub.string()}).Inc()
					}
					msg.ack()
				})
				if err != nil {
					logrus.WithError(err).Errorf("failed to listen for subscription %s on project %s", sub.string(), project)
					return err
				}
				return nil
			})
		}
	}
	return errGroup, derivedCtx, nil
}
func (s *PullServer) handlePulls(ctx context.Context, projectSubscriptions config.PubsubSubscriptions) (*errgroup.Group, context.Context, error) {
	// Since config might change we need be able to cancel the current run
	errGroup, derivedCtx := errgroup.WithContext(ctx)
	for project, subscriptions := range projectSubscriptions {
		client, err := s.Client.new(ctx, project)
		if err != nil {
			return errGroup, derivedCtx, err
		}
		for _, subName := range subscriptions {
			sub := client.subscription(subName)
			errGroup.Go(func() error {
				logrus.Infof("Listening for subscription %s on project %s", sub.string(), project)
				defer logrus.Warnf("Stopped Listening for subscription %s on project %s", sub.string(), project)
				err := sub.receive(derivedCtx, func(ctx context.Context, msg messageInterface) {
					if err = s.Subscriber.handleMessage(msg, sub.string()); err != nil {
						s.Subscriber.Metrics.ACKMessageCounter.With(prometheus.Labels{subscriptionLabel: sub.string()}).Inc()
					} else {
						s.Subscriber.Metrics.NACKMessageCounter.With(prometheus.Labels{subscriptionLabel: sub.string()}).Inc()
					}
					msg.ack()
				})
				if err != nil {
					logrus.WithError(err).Errorf("failed to listen for subscription %s on project %s", sub.string(), project)
					return err
				}
				return nil
			})
		}
	}
	return errGroup, derivedCtx, nil
}
// handlePulls pull for Pub/Sub subscriptions and handle them.