diff options
author | Paul Holzinger <pholzing@redhat.com> | 2022-03-11 15:02:04 +0100 |
---|---|---|
committer | Paul Holzinger <pholzing@redhat.com> | 2022-03-15 14:46:29 +0100 |
commit | 918fc88a983ea1c1ee7e0abb0fc1dce3c0a35be3 (patch) | |
tree | c967a39935b452f4cc61f425188bdbf42111c10a /vendor/k8s.io/apimachinery/pkg/watch | |
parent | 68ce83fe919f2d37762b8b746a73495f45e550f3 (diff) | |
download | podman-918fc88a983ea1c1ee7e0abb0fc1dce3c0a35be3.tar.gz podman-918fc88a983ea1c1ee7e0abb0fc1dce3c0a35be3.tar.bz2 podman-918fc88a983ea1c1ee7e0abb0fc1dce3c0a35be3.zip |
move k8s deps into podman
We only need a small part of the k8s dependencies but they are the
biggest dependencies in podman by far. Moving them into podman allows us
to remove the unnecessary parts.
Signed-off-by: Paul Holzinger <pholzing@redhat.com>
Diffstat (limited to 'vendor/k8s.io/apimachinery/pkg/watch')
-rw-r--r-- | vendor/k8s.io/apimachinery/pkg/watch/doc.go | 19 | ||||
-rw-r--r-- | vendor/k8s.io/apimachinery/pkg/watch/filter.go | 105 | ||||
-rw-r--r-- | vendor/k8s.io/apimachinery/pkg/watch/mux.go | 299 | ||||
-rw-r--r-- | vendor/k8s.io/apimachinery/pkg/watch/streamwatcher.go | 136 | ||||
-rw-r--r-- | vendor/k8s.io/apimachinery/pkg/watch/watch.go | 324 | ||||
-rw-r--r-- | vendor/k8s.io/apimachinery/pkg/watch/zz_generated.deepcopy.go | 40 |
6 files changed, 0 insertions, 923 deletions
diff --git a/vendor/k8s.io/apimachinery/pkg/watch/doc.go b/vendor/k8s.io/apimachinery/pkg/watch/doc.go deleted file mode 100644 index 7e6bf3fb9..000000000 --- a/vendor/k8s.io/apimachinery/pkg/watch/doc.go +++ /dev/null @@ -1,19 +0,0 @@ -/* -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" diff --git a/vendor/k8s.io/apimachinery/pkg/watch/filter.go b/vendor/k8s.io/apimachinery/pkg/watch/filter.go deleted file mode 100644 index 22c9449f5..000000000 --- a/vendor/k8s.io/apimachinery/pkg/watch/filter.go +++ /dev/null @@ -1,105 +0,0 @@ -/* -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 -} diff --git a/vendor/k8s.io/apimachinery/pkg/watch/mux.go b/vendor/k8s.io/apimachinery/pkg/watch/mux.go deleted file mode 100644 index e01d51906..000000000 --- a/vendor/k8s.io/apimachinery/pkg/watch/mux.go +++ /dev/null @@ -1,299 +0,0 @@ -/* -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" - - "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 - - 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()) { - 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 { - 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 { - // The panic here is to be consistent with the previous interface behavior - // we are willing to re-evaluate in the future. - panic("broadcaster already stopped") - } - return w -} - -// 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 { - 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 { - // The panic here is to be consistent with the previous interface behavior - // we are willing to re-evaluate in the future. - panic("broadcaster already stopped") - } - return w -} - -// 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) { - m.incoming <- Event{action, obj} -} - -// 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 { - select { - case m.incoming <- Event{action, obj}: - return true - default: - return false - } -} - -// 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) - }) -} diff --git a/vendor/k8s.io/apimachinery/pkg/watch/streamwatcher.go b/vendor/k8s.io/apimachinery/pkg/watch/streamwatcher.go deleted file mode 100644 index 42dcac2b9..000000000 --- a/vendor/k8s.io/apimachinery/pkg/watch/streamwatcher.go +++ /dev/null @@ -1,136 +0,0 @@ -/* -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, - }: - } - } -} diff --git a/vendor/k8s.io/apimachinery/pkg/watch/watch.go b/vendor/k8s.io/apimachinery/pkg/watch/watch.go deleted file mode 100644 index fd0550e4a..000000000 --- a/vendor/k8s.io/apimachinery/pkg/watch/watch.go +++ /dev/null @@ -1,324 +0,0 @@ -/* -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 { - // Stops watching. Will close the channel returned by ResultChan(). Releases - // any resources used by the watch. - Stop() - - // 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 -} diff --git a/vendor/k8s.io/apimachinery/pkg/watch/zz_generated.deepcopy.go b/vendor/k8s.io/apimachinery/pkg/watch/zz_generated.deepcopy.go deleted file mode 100644 index 71ef4da33..000000000 --- a/vendor/k8s.io/apimachinery/pkg/watch/zz_generated.deepcopy.go +++ /dev/null @@ -1,40 +0,0 @@ -// +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 -} |