prow/pubsub/subscriber/server.go
func (s *PullServer) Run(ctx context.Context) error {
	configEvent := make(chan config.Delta, 2)
	s.Subscriber.ConfigAgent.Subscribe(configEvent)

	var err error
	defer func() {
		if err != nil {
			logrus.WithError(ctx.Err()).Error("Pull server shutting down")
		}
		logrus.Warn("Pull server shutting down")
	}()
	currentConfig := s.Subscriber.ConfigAgent.Config().PubSubSubscriptions
	errGroup, derivedCtx, err := s.handlePulls(ctx, currentConfig)
	if err != nil {
		return err
	}

	for {
		select {
		// Parent context. Shutdown
		case <-ctx.Done():
			return ctx.Err()
		// Current thread context, it may be failing already
		case <-derivedCtx.Done():
			err = errGroup.Wait()
			return err
		// Checking for update config
		case event := <-configEvent:
			newConfig := event.After.PubSubSubscriptions
			logrus.Info("Received new config")
			if !reflect.DeepEqual(currentConfig, newConfig) {
				logrus.Warn("New config found, reloading pull Server")
				// Making sure the current thread finishes before starting a new one.
				errGroup.Wait()
				// Starting a new thread with new config
				errGroup, derivedCtx, err = s.handlePulls(ctx, newConfig)
				if err != nil {
					return err
				}
				currentConfig = newConfig
			}
		}
	}
}
func (s *PullServer) Run(ctx context.Context) error {
	configEvent := make(chan config.Delta, 2)
	s.Subscriber.ConfigAgent.Subscribe(configEvent)

	var err error
	defer func() {
		if err != nil {
			logrus.WithError(ctx.Err()).Error("Pull server shutting down")
		}
		logrus.Warn("Pull server shutting down")
	}()
	currentConfig := s.Subscriber.ConfigAgent.Config().PubSubSubscriptions
	errGroup, derivedCtx, err := s.handlePulls(ctx, currentConfig)
	if err != nil {
		return err
	}

	for {
		select {
		// Parent context. Shutdown
		case <-ctx.Done():
			return ctx.Err()
		// Current thread context, it may be failing already
		case <-derivedCtx.Done():
			err = errGroup.Wait()
			return err
		// Checking for update config
		case event := <-configEvent:
			newConfig := event.After.PubSubSubscriptions
			logrus.Info("Received new config")
			if !reflect.DeepEqual(currentConfig, newConfig) {
				logrus.Warn("New config found, reloading pull Server")
				// Making sure the current thread finishes before starting a new one.
				errGroup.Wait()
				// Starting a new thread with new config
				errGroup, derivedCtx, err = s.handlePulls(ctx, newConfig)
				if err != nil {
					return err
				}
				currentConfig = newConfig
			}
		}
	}
}
// Run will block listening to all subscriptions and return once the context is cancelled
// or one of the subscription has a unrecoverable error.