summaryrefslogtreecommitdiff
path: root/vendor/k8s.io/apimachinery/pkg/watch
diff options
context:
space:
mode:
Diffstat (limited to 'vendor/k8s.io/apimachinery/pkg/watch')
-rw-r--r--vendor/k8s.io/apimachinery/pkg/watch/doc.go19
-rw-r--r--vendor/k8s.io/apimachinery/pkg/watch/filter.go105
-rw-r--r--vendor/k8s.io/apimachinery/pkg/watch/mux.go299
-rw-r--r--vendor/k8s.io/apimachinery/pkg/watch/streamwatcher.go136
-rw-r--r--vendor/k8s.io/apimachinery/pkg/watch/watch.go324
-rw-r--r--vendor/k8s.io/apimachinery/pkg/watch/zz_generated.deepcopy.go40
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
-}