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
			}
		}
	}
}
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 ( " " ) } logrus . Warn ( " " ) } ( ) 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 ( " " ) if ! reflect . DeepEqual ( currentConfig , newConfig ) { logrus . Warn ( " " ) // 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.
Run will block listening to all subscriptions and return once the context is cancelled or one of the subscription has a unrecoverable error .