rebase: update to latest github.com/openshift/api version

Also vendor all dependencies.

Signed-off-by: Niels de Vos <ndevos@ibm.com>

api/vendor/k8s.io/apimachinery/pkg/watch/doc.go

@@ -0,0 +1,19 @@
+/*
+Copyright 2014 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package watch contains a generic watchable interface, and a fake for
+// testing code that uses the watch interface.
+package watch // import "k8s.io/apimachinery/pkg/watch"

api/vendor/k8s.io/apimachinery/pkg/watch/filter.go

@@ -0,0 +1,104 @@
+/*
+Copyright 2014 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package watch
+
+import (
+	"sync"
+)
+
+// FilterFunc should take an event, possibly modify it in some way, and return
+// the modified event. If the event should be ignored, then return keep=false.
+type FilterFunc func(in Event) (out Event, keep bool)
+
+// Filter passes all events through f before allowing them to pass on.
+// Putting a filter on a watch, as an unavoidable side-effect due to the way
+// go channels work, effectively causes the watch's event channel to have its
+// queue length increased by one.
+//
+// WARNING: filter has a fatal flaw, in that it can't properly update the
+// Type field (Add/Modified/Deleted) to reflect items beginning to pass the
+// filter when they previously didn't.
+func Filter(w Interface, f FilterFunc) Interface {
+	fw := &filteredWatch{
+		incoming: w,
+		result:   make(chan Event),
+		f:        f,
+	}
+	go fw.loop()
+	return fw
+}
+
+type filteredWatch struct {
+	incoming Interface
+	result   chan Event
+	f        FilterFunc
+}
+
+// ResultChan returns a channel which will receive filtered events.
+func (fw *filteredWatch) ResultChan() <-chan Event {
+	return fw.result
+}
+
+// Stop stops the upstream watch, which will eventually stop this watch.
+func (fw *filteredWatch) Stop() {
+	fw.incoming.Stop()
+}
+
+// loop waits for new values, filters them, and resends them.
+func (fw *filteredWatch) loop() {
+	defer close(fw.result)
+	for event := range fw.incoming.ResultChan() {
+		filtered, keep := fw.f(event)
+		if keep {
+			fw.result <- filtered
+		}
+	}
+}
+
+// Recorder records all events that are sent from the watch until it is closed.
+type Recorder struct {
+	Interface
+
+	lock   sync.Mutex
+	events []Event
+}
+
+var _ Interface = &Recorder{}
+
+// NewRecorder wraps an Interface and records any changes sent across it.
+func NewRecorder(w Interface) *Recorder {
+	r := &Recorder{}
+	r.Interface = Filter(w, r.record)
+	return r
+}
+
+// record is a FilterFunc and tracks each received event.
+func (r *Recorder) record(in Event) (Event, bool) {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+	r.events = append(r.events, in)
+	return in, true
+}
+
+// Events returns a copy of the events sent across this recorder.
+func (r *Recorder) Events() []Event {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+	copied := make([]Event, len(r.events))
+	copy(copied, r.events)
+	return copied
+}

api/vendor/k8s.io/apimachinery/pkg/watch/mux.go

@@ -0,0 +1,320 @@
+/*
+Copyright 2014 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package watch
+
+import (
+	"fmt"
+	"sync"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+// FullChannelBehavior controls how the Broadcaster reacts if a watcher's watch
+// channel is full.
+type FullChannelBehavior int
+
+const (
+	WaitIfChannelFull FullChannelBehavior = iota
+	DropIfChannelFull
+)
+
+// Buffer the incoming queue a little bit even though it should rarely ever accumulate
+// anything, just in case a few events are received in such a short window that
+// Broadcaster can't move them onto the watchers' queues fast enough.
+const incomingQueueLength = 25
+
+// Broadcaster distributes event notifications among any number of watchers. Every event
+// is delivered to every watcher.
+type Broadcaster struct {
+	watchers     map[int64]*broadcasterWatcher
+	nextWatcher  int64
+	distributing sync.WaitGroup
+
+	// incomingBlock allows us to ensure we don't race and end up sending events
+	// to a closed channel following a broadcaster shutdown.
+	incomingBlock sync.Mutex
+	incoming      chan Event
+	stopped       chan struct{}
+
+	// How large to make watcher's channel.
+	watchQueueLength int
+	// If one of the watch channels is full, don't wait for it to become empty.
+	// Instead just deliver it to the watchers that do have space in their
+	// channels and move on to the next event.
+	// It's more fair to do this on a per-watcher basis than to do it on the
+	// "incoming" channel, which would allow one slow watcher to prevent all
+	// other watchers from getting new events.
+	fullChannelBehavior FullChannelBehavior
+}
+
+// NewBroadcaster creates a new Broadcaster. queueLength is the maximum number of events to queue per watcher.
+// It is guaranteed that events will be distributed in the order in which they occur,
+// but the order in which a single event is distributed among all of the watchers is unspecified.
+func NewBroadcaster(queueLength int, fullChannelBehavior FullChannelBehavior) *Broadcaster {
+	m := &Broadcaster{
+		watchers:            map[int64]*broadcasterWatcher{},
+		incoming:            make(chan Event, incomingQueueLength),
+		stopped:             make(chan struct{}),
+		watchQueueLength:    queueLength,
+		fullChannelBehavior: fullChannelBehavior,
+	}
+	m.distributing.Add(1)
+	go m.loop()
+	return m
+}
+
+// NewLongQueueBroadcaster functions nearly identically to NewBroadcaster,
+// except that the incoming queue is the same size as the outgoing queues
+// (specified by queueLength).
+func NewLongQueueBroadcaster(queueLength int, fullChannelBehavior FullChannelBehavior) *Broadcaster {
+	m := &Broadcaster{
+		watchers:            map[int64]*broadcasterWatcher{},
+		incoming:            make(chan Event, queueLength),
+		stopped:             make(chan struct{}),
+		watchQueueLength:    queueLength,
+		fullChannelBehavior: fullChannelBehavior,
+	}
+	m.distributing.Add(1)
+	go m.loop()
+	return m
+}
+
+const internalRunFunctionMarker = "internal-do-function"
+
+// a function type we can shoehorn into the queue.
+type functionFakeRuntimeObject func()
+
+func (obj functionFakeRuntimeObject) GetObjectKind() schema.ObjectKind {
+	return schema.EmptyObjectKind
+}
+func (obj functionFakeRuntimeObject) DeepCopyObject() runtime.Object {
+	if obj == nil {
+		return nil
+	}
+	// funcs are immutable. Hence, just return the original func.
+	return obj
+}
+
+// Execute f, blocking the incoming queue (and waiting for it to drain first).
+// The purpose of this terrible hack is so that watchers added after an event
+// won't ever see that event, and will always see any event after they are
+// added.
+func (m *Broadcaster) blockQueue(f func()) {
+	m.incomingBlock.Lock()
+	defer m.incomingBlock.Unlock()
+	select {
+	case <-m.stopped:
+		return
+	default:
+	}
+	var wg sync.WaitGroup
+	wg.Add(1)
+	m.incoming <- Event{
+		Type: internalRunFunctionMarker,
+		Object: functionFakeRuntimeObject(func() {
+			defer wg.Done()
+			f()
+		}),
+	}
+	wg.Wait()
+}
+
+// Watch adds a new watcher to the list and returns an Interface for it.
+// Note: new watchers will only receive new events. They won't get an entire history
+// of previous events. It will block until the watcher is actually added to the
+// broadcaster.
+func (m *Broadcaster) Watch() (Interface, error) {
+	var w *broadcasterWatcher
+	m.blockQueue(func() {
+		id := m.nextWatcher
+		m.nextWatcher++
+		w = &broadcasterWatcher{
+			result:  make(chan Event, m.watchQueueLength),
+			stopped: make(chan struct{}),
+			id:      id,
+			m:       m,
+		}
+		m.watchers[id] = w
+	})
+	if w == nil {
+		return nil, fmt.Errorf("broadcaster already stopped")
+	}
+	return w, nil
+}
+
+// WatchWithPrefix adds a new watcher to the list and returns an Interface for it. It sends
+// queuedEvents down the new watch before beginning to send ordinary events from Broadcaster.
+// The returned watch will have a queue length that is at least large enough to accommodate
+// all of the items in queuedEvents. It will block until the watcher is actually added to
+// the broadcaster.
+func (m *Broadcaster) WatchWithPrefix(queuedEvents []Event) (Interface, error) {
+	var w *broadcasterWatcher
+	m.blockQueue(func() {
+		id := m.nextWatcher
+		m.nextWatcher++
+		length := m.watchQueueLength
+		if n := len(queuedEvents) + 1; n > length {
+			length = n
+		}
+		w = &broadcasterWatcher{
+			result:  make(chan Event, length),
+			stopped: make(chan struct{}),
+			id:      id,
+			m:       m,
+		}
+		m.watchers[id] = w
+		for _, e := range queuedEvents {
+			w.result <- e
+		}
+	})
+	if w == nil {
+		return nil, fmt.Errorf("broadcaster already stopped")
+	}
+	return w, nil
+}
+
+// stopWatching stops the given watcher and removes it from the list.
+func (m *Broadcaster) stopWatching(id int64) {
+	m.blockQueue(func() {
+		w, ok := m.watchers[id]
+		if !ok {
+			// No need to do anything, it's already been removed from the list.
+			return
+		}
+		delete(m.watchers, id)
+		close(w.result)
+	})
+}
+
+// closeAll disconnects all watchers (presumably in response to a Shutdown call).
+func (m *Broadcaster) closeAll() {
+	for _, w := range m.watchers {
+		close(w.result)
+	}
+	// Delete everything from the map, since presence/absence in the map is used
+	// by stopWatching to avoid double-closing the channel.
+	m.watchers = map[int64]*broadcasterWatcher{}
+}
+
+// Action distributes the given event among all watchers.
+func (m *Broadcaster) Action(action EventType, obj runtime.Object) error {
+	m.incomingBlock.Lock()
+	defer m.incomingBlock.Unlock()
+	select {
+	case <-m.stopped:
+		return fmt.Errorf("broadcaster already stopped")
+	default:
+	}
+
+	m.incoming <- Event{action, obj}
+	return nil
+}
+
+// Action distributes the given event among all watchers, or drops it on the floor
+// if too many incoming actions are queued up.  Returns true if the action was sent,
+// false if dropped.
+func (m *Broadcaster) ActionOrDrop(action EventType, obj runtime.Object) (bool, error) {
+	m.incomingBlock.Lock()
+	defer m.incomingBlock.Unlock()
+
+	// Ensure that if the broadcaster is stopped we do not send events to it.
+	select {
+	case <-m.stopped:
+		return false, fmt.Errorf("broadcaster already stopped")
+	default:
+	}
+
+	select {
+	case m.incoming <- Event{action, obj}:
+		return true, nil
+	default:
+		return false, nil
+	}
+}
+
+// Shutdown disconnects all watchers (but any queued events will still be distributed).
+// You must not call Action or Watch* after calling Shutdown. This call blocks
+// until all events have been distributed through the outbound channels. Note
+// that since they can be buffered, this means that the watchers might not
+// have received the data yet as it can remain sitting in the buffered
+// channel. It will block until the broadcaster stop request is actually executed
+func (m *Broadcaster) Shutdown() {
+	m.blockQueue(func() {
+		close(m.stopped)
+		close(m.incoming)
+	})
+	m.distributing.Wait()
+}
+
+// loop receives from m.incoming and distributes to all watchers.
+func (m *Broadcaster) loop() {
+	// Deliberately not catching crashes here. Yes, bring down the process if there's a
+	// bug in watch.Broadcaster.
+	for event := range m.incoming {
+		if event.Type == internalRunFunctionMarker {
+			event.Object.(functionFakeRuntimeObject)()
+			continue
+		}
+		m.distribute(event)
+	}
+	m.closeAll()
+	m.distributing.Done()
+}
+
+// distribute sends event to all watchers. Blocking.
+func (m *Broadcaster) distribute(event Event) {
+	if m.fullChannelBehavior == DropIfChannelFull {
+		for _, w := range m.watchers {
+			select {
+			case w.result <- event:
+			case <-w.stopped:
+			default: // Don't block if the event can't be queued.
+			}
+		}
+	} else {
+		for _, w := range m.watchers {
+			select {
+			case w.result <- event:
+			case <-w.stopped:
+			}
+		}
+	}
+}
+
+// broadcasterWatcher handles a single watcher of a broadcaster
+type broadcasterWatcher struct {
+	result  chan Event
+	stopped chan struct{}
+	stop    sync.Once
+	id      int64
+	m       *Broadcaster
+}
+
+// ResultChan returns a channel to use for waiting on events.
+func (mw *broadcasterWatcher) ResultChan() <-chan Event {
+	return mw.result
+}
+
+// Stop stops watching and removes mw from its list.
+// It will block until the watcher stop request is actually executed
+func (mw *broadcasterWatcher) Stop() {
+	mw.stop.Do(func() {
+		close(mw.stopped)
+		mw.m.stopWatching(mw.id)
+	})
+}

api/vendor/k8s.io/apimachinery/pkg/watch/streamwatcher.go

@@ -0,0 +1,136 @@
+/*
+Copyright 2014 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package watch
+
+import (
+	"fmt"
+	"io"
+	"sync"
+
+	"k8s.io/klog/v2"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/net"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+)
+
+// Decoder allows StreamWatcher to watch any stream for which a Decoder can be written.
+type Decoder interface {
+	// Decode should return the type of event, the decoded object, or an error.
+	// An error will cause StreamWatcher to call Close(). Decode should block until
+	// it has data or an error occurs.
+	Decode() (action EventType, object runtime.Object, err error)
+
+	// Close should close the underlying io.Reader, signalling to the source of
+	// the stream that it is no longer being watched. Close() must cause any
+	// outstanding call to Decode() to return with an error of some sort.
+	Close()
+}
+
+// Reporter hides the details of how an error is turned into a runtime.Object for
+// reporting on a watch stream since this package may not import a higher level report.
+type Reporter interface {
+	// AsObject must convert err into a valid runtime.Object for the watch stream.
+	AsObject(err error) runtime.Object
+}
+
+// StreamWatcher turns any stream for which you can write a Decoder interface
+// into a watch.Interface.
+type StreamWatcher struct {
+	sync.Mutex
+	source   Decoder
+	reporter Reporter
+	result   chan Event
+	done     chan struct{}
+}
+
+// NewStreamWatcher creates a StreamWatcher from the given decoder.
+func NewStreamWatcher(d Decoder, r Reporter) *StreamWatcher {
+	sw := &StreamWatcher{
+		source:   d,
+		reporter: r,
+		// It's easy for a consumer to add buffering via an extra
+		// goroutine/channel, but impossible for them to remove it,
+		// so nonbuffered is better.
+		result: make(chan Event),
+		// If the watcher is externally stopped there is no receiver anymore
+		// and the send operations on the result channel, especially the
+		// error reporting might block forever.
+		// Therefore a dedicated stop channel is used to resolve this blocking.
+		done: make(chan struct{}),
+	}
+	go sw.receive()
+	return sw
+}
+
+// ResultChan implements Interface.
+func (sw *StreamWatcher) ResultChan() <-chan Event {
+	return sw.result
+}
+
+// Stop implements Interface.
+func (sw *StreamWatcher) Stop() {
+	// Call Close() exactly once by locking and setting a flag.
+	sw.Lock()
+	defer sw.Unlock()
+	// closing a closed channel always panics, therefore check before closing
+	select {
+	case <-sw.done:
+	default:
+		close(sw.done)
+		sw.source.Close()
+	}
+}
+
+// receive reads result from the decoder in a loop and sends down the result channel.
+func (sw *StreamWatcher) receive() {
+	defer utilruntime.HandleCrash()
+	defer close(sw.result)
+	defer sw.Stop()
+	for {
+		action, obj, err := sw.source.Decode()
+		if err != nil {
+			switch err {
+			case io.EOF:
+				// watch closed normally
+			case io.ErrUnexpectedEOF:
+				klog.V(1).Infof("Unexpected EOF during watch stream event decoding: %v", err)
+			default:
+				if net.IsProbableEOF(err) || net.IsTimeout(err) {
+					klog.V(5).Infof("Unable to decode an event from the watch stream: %v", err)
+				} else {
+					select {
+					case <-sw.done:
+					case sw.result <- Event{
+						Type:   Error,
+						Object: sw.reporter.AsObject(fmt.Errorf("unable to decode an event from the watch stream: %v", err)),
+					}:
+					}
+				}
+			}
+			return
+		}
+		select {
+		case <-sw.done:
+			return
+		case sw.result <- Event{
+			Type:   action,
+			Object: obj,
+		}:
+		}
+	}
+}

api/vendor/k8s.io/apimachinery/pkg/watch/watch.go

@@ -0,0 +1,324 @@
+/*
+Copyright 2014 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package watch
+
+import (
+	"fmt"
+	"sync"
+
+	"k8s.io/klog/v2"
+
+	"k8s.io/apimachinery/pkg/runtime"
+)
+
+// Interface can be implemented by anything that knows how to watch and report changes.
+type Interface interface {
+	// Stop stops watching. Will close the channel returned by ResultChan(). Releases
+	// any resources used by the watch.
+	Stop()
+
+	// ResultChan returns a chan which will receive all the events. If an error occurs
+	// or Stop() is called, the implementation will close this channel and
+	// release any resources used by the watch.
+	ResultChan() <-chan Event
+}
+
+// EventType defines the possible types of events.
+type EventType string
+
+const (
+	Added    EventType = "ADDED"
+	Modified EventType = "MODIFIED"
+	Deleted  EventType = "DELETED"
+	Bookmark EventType = "BOOKMARK"
+	Error    EventType = "ERROR"
+)
+
+var (
+	DefaultChanSize int32 = 100
+)
+
+// Event represents a single event to a watched resource.
+// +k8s:deepcopy-gen=true
+type Event struct {
+	Type EventType
+
+	// Object is:
+	//  * If Type is Added or Modified: the new state of the object.
+	//  * If Type is Deleted: the state of the object immediately before deletion.
+	//  * If Type is Bookmark: the object (instance of a type being watched) where
+	//    only ResourceVersion field is set. On successful restart of watch from a
+	//    bookmark resourceVersion, client is guaranteed to not get repeat event
+	//    nor miss any events.
+	//  * If Type is Error: *api.Status is recommended; other types may make sense
+	//    depending on context.
+	Object runtime.Object
+}
+
+type emptyWatch chan Event
+
+// NewEmptyWatch returns a watch interface that returns no results and is closed.
+// May be used in certain error conditions where no information is available but
+// an error is not warranted.
+func NewEmptyWatch() Interface {
+	ch := make(chan Event)
+	close(ch)
+	return emptyWatch(ch)
+}
+
+// Stop implements Interface
+func (w emptyWatch) Stop() {
+}
+
+// ResultChan implements Interface
+func (w emptyWatch) ResultChan() <-chan Event {
+	return chan Event(w)
+}
+
+// FakeWatcher lets you test anything that consumes a watch.Interface; threadsafe.
+type FakeWatcher struct {
+	result  chan Event
+	stopped bool
+	sync.Mutex
+}
+
+func NewFake() *FakeWatcher {
+	return &FakeWatcher{
+		result: make(chan Event),
+	}
+}
+
+func NewFakeWithChanSize(size int, blocking bool) *FakeWatcher {
+	return &FakeWatcher{
+		result: make(chan Event, size),
+	}
+}
+
+// Stop implements Interface.Stop().
+func (f *FakeWatcher) Stop() {
+	f.Lock()
+	defer f.Unlock()
+	if !f.stopped {
+		klog.V(4).Infof("Stopping fake watcher.")
+		close(f.result)
+		f.stopped = true
+	}
+}
+
+func (f *FakeWatcher) IsStopped() bool {
+	f.Lock()
+	defer f.Unlock()
+	return f.stopped
+}
+
+// Reset prepares the watcher to be reused.
+func (f *FakeWatcher) Reset() {
+	f.Lock()
+	defer f.Unlock()
+	f.stopped = false
+	f.result = make(chan Event)
+}
+
+func (f *FakeWatcher) ResultChan() <-chan Event {
+	return f.result
+}
+
+// Add sends an add event.
+func (f *FakeWatcher) Add(obj runtime.Object) {
+	f.result <- Event{Added, obj}
+}
+
+// Modify sends a modify event.
+func (f *FakeWatcher) Modify(obj runtime.Object) {
+	f.result <- Event{Modified, obj}
+}
+
+// Delete sends a delete event.
+func (f *FakeWatcher) Delete(lastValue runtime.Object) {
+	f.result <- Event{Deleted, lastValue}
+}
+
+// Error sends an Error event.
+func (f *FakeWatcher) Error(errValue runtime.Object) {
+	f.result <- Event{Error, errValue}
+}
+
+// Action sends an event of the requested type, for table-based testing.
+func (f *FakeWatcher) Action(action EventType, obj runtime.Object) {
+	f.result <- Event{action, obj}
+}
+
+// RaceFreeFakeWatcher lets you test anything that consumes a watch.Interface; threadsafe.
+type RaceFreeFakeWatcher struct {
+	result  chan Event
+	Stopped bool
+	sync.Mutex
+}
+
+func NewRaceFreeFake() *RaceFreeFakeWatcher {
+	return &RaceFreeFakeWatcher{
+		result: make(chan Event, DefaultChanSize),
+	}
+}
+
+// Stop implements Interface.Stop().
+func (f *RaceFreeFakeWatcher) Stop() {
+	f.Lock()
+	defer f.Unlock()
+	if !f.Stopped {
+		klog.V(4).Infof("Stopping fake watcher.")
+		close(f.result)
+		f.Stopped = true
+	}
+}
+
+func (f *RaceFreeFakeWatcher) IsStopped() bool {
+	f.Lock()
+	defer f.Unlock()
+	return f.Stopped
+}
+
+// Reset prepares the watcher to be reused.
+func (f *RaceFreeFakeWatcher) Reset() {
+	f.Lock()
+	defer f.Unlock()
+	f.Stopped = false
+	f.result = make(chan Event, DefaultChanSize)
+}
+
+func (f *RaceFreeFakeWatcher) ResultChan() <-chan Event {
+	f.Lock()
+	defer f.Unlock()
+	return f.result
+}
+
+// Add sends an add event.
+func (f *RaceFreeFakeWatcher) Add(obj runtime.Object) {
+	f.Lock()
+	defer f.Unlock()
+	if !f.Stopped {
+		select {
+		case f.result <- Event{Added, obj}:
+			return
+		default:
+			panic(fmt.Errorf("channel full"))
+		}
+	}
+}
+
+// Modify sends a modify event.
+func (f *RaceFreeFakeWatcher) Modify(obj runtime.Object) {
+	f.Lock()
+	defer f.Unlock()
+	if !f.Stopped {
+		select {
+		case f.result <- Event{Modified, obj}:
+			return
+		default:
+			panic(fmt.Errorf("channel full"))
+		}
+	}
+}
+
+// Delete sends a delete event.
+func (f *RaceFreeFakeWatcher) Delete(lastValue runtime.Object) {
+	f.Lock()
+	defer f.Unlock()
+	if !f.Stopped {
+		select {
+		case f.result <- Event{Deleted, lastValue}:
+			return
+		default:
+			panic(fmt.Errorf("channel full"))
+		}
+	}
+}
+
+// Error sends an Error event.
+func (f *RaceFreeFakeWatcher) Error(errValue runtime.Object) {
+	f.Lock()
+	defer f.Unlock()
+	if !f.Stopped {
+		select {
+		case f.result <- Event{Error, errValue}:
+			return
+		default:
+			panic(fmt.Errorf("channel full"))
+		}
+	}
+}
+
+// Action sends an event of the requested type, for table-based testing.
+func (f *RaceFreeFakeWatcher) Action(action EventType, obj runtime.Object) {
+	f.Lock()
+	defer f.Unlock()
+	if !f.Stopped {
+		select {
+		case f.result <- Event{action, obj}:
+			return
+		default:
+			panic(fmt.Errorf("channel full"))
+		}
+	}
+}
+
+// ProxyWatcher lets you wrap your channel in watch Interface. threadsafe.
+type ProxyWatcher struct {
+	result chan Event
+	stopCh chan struct{}
+
+	mutex   sync.Mutex
+	stopped bool
+}
+
+var _ Interface = &ProxyWatcher{}
+
+// NewProxyWatcher creates new ProxyWatcher by wrapping a channel
+func NewProxyWatcher(ch chan Event) *ProxyWatcher {
+	return &ProxyWatcher{
+		result:  ch,
+		stopCh:  make(chan struct{}),
+		stopped: false,
+	}
+}
+
+// Stop implements Interface
+func (pw *ProxyWatcher) Stop() {
+	pw.mutex.Lock()
+	defer pw.mutex.Unlock()
+	if !pw.stopped {
+		pw.stopped = true
+		close(pw.stopCh)
+	}
+}
+
+// Stopping returns true if Stop() has been called
+func (pw *ProxyWatcher) Stopping() bool {
+	pw.mutex.Lock()
+	defer pw.mutex.Unlock()
+	return pw.stopped
+}
+
+// ResultChan implements Interface
+func (pw *ProxyWatcher) ResultChan() <-chan Event {
+	return pw.result
+}
+
+// StopChan returns stop channel
+func (pw *ProxyWatcher) StopChan() <-chan struct{} {
+	return pw.stopCh
+}

api/vendor/k8s.io/apimachinery/pkg/watch/zz_generated.deepcopy.go

@@ -0,0 +1,41 @@
+//go:build !ignore_autogenerated
+// +build !ignore_autogenerated
+
+/*
+Copyright The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by deepcopy-gen. DO NOT EDIT.
+
+package watch
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *Event) DeepCopyInto(out *Event) {
+	*out = *in
+	if in.Object != nil {
+		out.Object = in.Object.DeepCopyObject()
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Event.
+func (in *Event) DeepCopy() *Event {
+	if in == nil {
+		return nil
+	}
+	out := new(Event)
+	in.DeepCopyInto(out)
+	return out
+}