// watchHandler watches w and keeps *resourceVersion up to date.
func (gc *Reflector) watchHandler(w watch.Interface, resourceVersion *uint64) {
	for {
		event, ok := <-w.ResultChan()
		if !ok {
			glog.Errorf("unexpected watch close")
			return
		}
		if e, a := gc.expectedType, reflect.TypeOf(event.Object); e != a {
			glog.Errorf("expected type %v, but watch event object had type %v", e, a)
			continue
		}
		jsonBase, err := api.FindJSONBase(event.Object)
		if err != nil {
			glog.Errorf("unable to understand watch event %#v", event)
			continue
		}
		switch event.Type {
		case watch.Added:
			gc.store.Add(jsonBase.ID(), event.Object)
		case watch.Modified:
			gc.store.Update(jsonBase.ID(), event.Object)
		case watch.Deleted:
			// TODO: Will any consumers need access to the "last known
			// state", which is passed in event.Object? If so, may need
			// to change this.
			gc.store.Delete(jsonBase.ID())
		default:
			glog.Errorf("unable to understand watch event %#v", event)
		}
		*resourceVersion = jsonBase.ResourceVersion() + 1
	}
}

func (gc *Reflector) watchHandler(w watch.Interface) {
	for {
		event, ok := <-w.ResultChan()
		if !ok {
			glog.Errorf("unexpected watch close")
			return
		}
		if e, a := gc.expectedType, reflect.TypeOf(event.Object); e != a {
			glog.Errorf("expected type %v, but watch event object had type %v", e, a)
			continue
		}
		jsonBase, err := api.FindJSONBase(event.Object)
		if err != nil {
			glog.Errorf("unable to understand watch event %#v", event)
			continue
		}
		switch event.Type {
		case watch.Added:
			gc.store.Add(jsonBase.ID(), event.Object)
		case watch.Modified:
			gc.store.Update(jsonBase.ID(), event.Object)
		case watch.Deleted:
			gc.store.Delete(jsonBase.ID(), event.Object)
		default:
			glog.Errorf("unable to understand watch event %#v", event)
		}
	}
}

// watchHandler watches w and keeps *resourceVersion up to date.
func (r *Reflector) watchHandler(w watch.Interface, resourceVersion *string, resyncCh <-chan time.Time, stopCh <-chan struct{}) error {
	start := time.Now()
	eventCount := 0

	// Stopping the watcher should be idempotent and if we return from this function there's no way
	// we're coming back in with the same watch interface.
	defer w.Stop()

loop:
	for {
		select {
		case <-stopCh:
			return errorStopRequested
		case <-resyncCh:
			return errorResyncRequested
		case event, ok := <-w.ResultChan():
			if !ok {
				break loop
			}
			if event.Type == watch.Error {
				return apierrs.FromObject(event.Object)
			}
			if e, a := r.expectedType, reflect.TypeOf(event.Object); e != a {
				util.HandleError(fmt.Errorf("%s: expected type %v, but watch event object had type %v", r.name, e, a))
				continue
			}
			meta, err := meta.Accessor(event.Object)
			if err != nil {
				util.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", r.name, event))
				continue
			}
			switch event.Type {
			case watch.Added:
				r.store.Add(event.Object)
			case watch.Modified:
				r.store.Update(event.Object)
			case watch.Deleted:
				// TODO: Will any consumers need access to the "last known
				// state", which is passed in event.Object? If so, may need
				// to change this.
				r.store.Delete(event.Object)
			default:
				util.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", r.name, event))
			}
			*resourceVersion = meta.ResourceVersion()
			r.setLastSyncResourceVersion(*resourceVersion)
			eventCount++
		}
	}

	watchDuration := time.Now().Sub(start)
	if watchDuration < 1*time.Second && eventCount == 0 {
		glog.V(4).Infof("%s: Unexpected watch close - watch lasted less than a second and no items received", r.name)
		return errors.New("very short watch")
	}
	glog.V(4).Infof("%s: Watch close - %v total %v items received", r.name, r.expectedType, eventCount)
	return nil
}

func waitForWatch(t *testing.T, name string, w watchapi.Interface) *watchapi.Event {
	select {
	case e := <-w.ResultChan():
		return &e
	case <-time.After(BuildControllersWatchTimeout):
		t.Fatalf("Timed out waiting for watch: %s", name)
		return nil
	}
	return nil
}

// watchHandler watches w and keeps *resourceVersion up to date.
func (r *Reflector) watchHandler(w watch.Interface, resourceVersion *string, exitWatch <-chan time.Time) error {
	start := time.Now()
	eventCount := 0
loop:
	for {
		select {
		case <-exitWatch:
			w.Stop()
			return errorResyncRequested
		case event, ok := <-w.ResultChan():
			if !ok {
				break loop
			}
			if event.Type == watch.Error {
				return apierrs.FromObject(event.Object)
			}
			if e, a := r.expectedType, reflect.TypeOf(event.Object); e != a {
				glog.Errorf("expected type %v, but watch event object had type %v", e, a)
				continue
			}
			meta, err := meta.Accessor(event.Object)
			if err != nil {
				glog.Errorf("unable to understand watch event %#v", event)
				continue
			}
			switch event.Type {
			case watch.Added:
				r.store.Add(event.Object)
			case watch.Modified:
				r.store.Update(event.Object)
			case watch.Deleted:
				// TODO: Will any consumers need access to the "last known
				// state", which is passed in event.Object? If so, may need
				// to change this.
				r.store.Delete(event.Object)
			default:
				glog.Errorf("unable to understand watch event %#v", event)
			}
			*resourceVersion = meta.ResourceVersion()
			eventCount++
		}
	}

	watchDuration := time.Now().Sub(start)
	if watchDuration < 1*time.Second && eventCount == 0 {
		glog.V(4).Infof("Unexpected watch close - watch lasted less than a second and no items received")
		return errors.New("very short watch")
	}
	glog.V(4).Infof("Watch close - %v total %v items received", r.expectedType, eventCount)
	return nil
}

// waitForComplete waits for the Build to finish
func waitForComplete(build *buildapi.Build, w watch.Interface, t *testing.T) {
	for event := range w.ResultChan() {
		eventBuild, ok := event.Object.(*buildapi.Build)
		if !ok {
			t.Fatalf("Cannot convert input to Build")
		}
		if build.Name != eventBuild.Name {
			continue
		}
		switch eventBuild.Status {
		case buildapi.BuildStatusFailed, buildapi.BuildStatusError:
			t.Fatalf("Unexpected status for Build %s: %v", eventBuild.Name, buildapi.BuildStatusFailed)
		case buildapi.BuildStatusComplete:
			return
		default:
			fmt.Printf("Build %s updated: %v\n", eventBuild.Name, eventBuild.Status)
		}
	}
}

// WatchLoop loops, passing events in w to fn.
// If user sends interrupt signal, shut down cleanly. Otherwise, never return.
func WatchLoop(w watch.Interface, fn func(watch.Event) error) {
	signals := make(chan os.Signal, 1)
	signal.Notify(signals, os.Interrupt)
	defer signal.Stop(signals)
	for {
		select {
		case event, ok := <-w.ResultChan():
			if !ok {
				return
			}
			if err := fn(event); err != nil {
				w.Stop()
			}
		case <-signals:
			w.Stop()
		}
	}
}

// WatchLoop loops, writing objects in the events from w to printer.
// If user sends interrupt signal, shut down cleanly. Otherwise, never return.
func WatchLoop(w watch.Interface, printer ResourcePrinter, out io.Writer) {
	signals := make(chan os.Signal, 1)
	signal.Notify(signals, os.Interrupt)
	defer signal.Stop(signals)
	for {
		select {
		case event, ok := <-w.ResultChan():
			if !ok {
				return
			}
			// TODO: need to print out added/modified/deleted!
			if err := printer.PrintObj(event.Object, out); err != nil {
				w.Stop()
			}
		case <-signals:
			w.Stop()
		}
	}
}