cmd/bpa-operator/vendor/k8s.io/apimachinery/pkg/watch/watch.go

   1 /*
   2 Copyright 2014 The Kubernetes Authors.
   3
   4 Licensed under the Apache License, Version 2.0 (the "License");
   5 you may not use this file except in compliance with the License.
   6 You may obtain a copy of the License at
   7
   8     http://www.apache.org/licenses/LICENSE-2.0
   9
  10 Unless required by applicable law or agreed to in writing, software
  11 distributed under the License is distributed on an "AS IS" BASIS,
  12 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 See the License for the specific language governing permissions and
  14 limitations under the License.
  15 */
  16
  17 package watch
  18
  19 import (
  20         "fmt"
  21         "sync"
  22
  23         "k8s.io/klog"
  24
  25         "k8s.io/apimachinery/pkg/runtime"
  26 )
  27
  28 // Interface can be implemented by anything that knows how to watch and report changes.
  29 type Interface interface {
  30         // Stops watching. Will close the channel returned by ResultChan(). Releases
  31         // any resources used by the watch.
  32         Stop()
  33
  34         // Returns a chan which will receive all the events. If an error occurs
  35         // or Stop() is called, this channel will be closed, in which case the
  36         // watch should be completely cleaned up.
  37         ResultChan() <-chan Event
  38 }
  39
  40 // EventType defines the possible types of events.
  41 type EventType string
  42
  43 const (
  44         Added    EventType = "ADDED"
  45         Modified EventType = "MODIFIED"
  46         Deleted  EventType = "DELETED"
  47         Error    EventType = "ERROR"
  48
  49         DefaultChanSize int32 = 100
  50 )
  51
  52 // Event represents a single event to a watched resource.
  53 // +k8s:deepcopy-gen=true
  54 type Event struct {
  55         Type EventType
  56
  57         // Object is:
  58         //  * If Type is Added or Modified: the new state of the object.
  59         //  * If Type is Deleted: the state of the object immediately before deletion.
  60         //  * If Type is Error: *api.Status is recommended; other types may make sense
  61         //    depending on context.
  62         Object runtime.Object
  63 }
  64
  65 type emptyWatch chan Event
  66
  67 // NewEmptyWatch returns a watch interface that returns no results and is closed.
  68 // May be used in certain error conditions where no information is available but
  69 // an error is not warranted.
  70 func NewEmptyWatch() Interface {
  71         ch := make(chan Event)
  72         close(ch)
  73         return emptyWatch(ch)
  74 }
  75
  76 // Stop implements Interface
  77 func (w emptyWatch) Stop() {
  78 }
  79
  80 // ResultChan implements Interface
  81 func (w emptyWatch) ResultChan() <-chan Event {
  82         return chan Event(w)
  83 }
  84
  85 // FakeWatcher lets you test anything that consumes a watch.Interface; threadsafe.
  86 type FakeWatcher struct {
  87         result  chan Event
  88         Stopped bool
  89         sync.Mutex
  90 }
  91
  92 func NewFake() *FakeWatcher {
  93         return &FakeWatcher{
  94                 result: make(chan Event),
  95         }
  96 }
  97
  98 func NewFakeWithChanSize(size int, blocking bool) *FakeWatcher {
  99         return &FakeWatcher{
 100                 result: make(chan Event, size),
 101         }
 102 }
 103
 104 // Stop implements Interface.Stop().
 105 func (f *FakeWatcher) Stop() {
 106         f.Lock()
 107         defer f.Unlock()
 108         if !f.Stopped {
 109                 klog.V(4).Infof("Stopping fake watcher.")
 110                 close(f.result)
 111                 f.Stopped = true
 112         }
 113 }
 114
 115 func (f *FakeWatcher) IsStopped() bool {
 116         f.Lock()
 117         defer f.Unlock()
 118         return f.Stopped
 119 }
 120
 121 // Reset prepares the watcher to be reused.
 122 func (f *FakeWatcher) Reset() {
 123         f.Lock()
 124         defer f.Unlock()
 125         f.Stopped = false
 126         f.result = make(chan Event)
 127 }
 128
 129 func (f *FakeWatcher) ResultChan() <-chan Event {
 130         return f.result
 131 }
 132
 133 // Add sends an add event.
 134 func (f *FakeWatcher) Add(obj runtime.Object) {
 135         f.result <- Event{Added, obj}
 136 }
 137
 138 // Modify sends a modify event.
 139 func (f *FakeWatcher) Modify(obj runtime.Object) {
 140         f.result <- Event{Modified, obj}
 141 }
 142
 143 // Delete sends a delete event.
 144 func (f *FakeWatcher) Delete(lastValue runtime.Object) {
 145         f.result <- Event{Deleted, lastValue}
 146 }
 147
 148 // Error sends an Error event.
 149 func (f *FakeWatcher) Error(errValue runtime.Object) {
 150         f.result <- Event{Error, errValue}
 151 }
 152
 153 // Action sends an event of the requested type, for table-based testing.
 154 func (f *FakeWatcher) Action(action EventType, obj runtime.Object) {
 155         f.result <- Event{action, obj}
 156 }
 157
 158 // RaceFreeFakeWatcher lets you test anything that consumes a watch.Interface; threadsafe.
 159 type RaceFreeFakeWatcher struct {
 160         result  chan Event
 161         Stopped bool
 162         sync.Mutex
 163 }
 164
 165 func NewRaceFreeFake() *RaceFreeFakeWatcher {
 166         return &RaceFreeFakeWatcher{
 167                 result: make(chan Event, DefaultChanSize),
 168         }
 169 }
 170
 171 // Stop implements Interface.Stop().
 172 func (f *RaceFreeFakeWatcher) Stop() {
 173         f.Lock()
 174         defer f.Unlock()
 175         if !f.Stopped {
 176                 klog.V(4).Infof("Stopping fake watcher.")
 177                 close(f.result)
 178                 f.Stopped = true
 179         }
 180 }
 181
 182 func (f *RaceFreeFakeWatcher) IsStopped() bool {
 183         f.Lock()
 184         defer f.Unlock()
 185         return f.Stopped
 186 }
 187
 188 // Reset prepares the watcher to be reused.
 189 func (f *RaceFreeFakeWatcher) Reset() {
 190         f.Lock()
 191         defer f.Unlock()
 192         f.Stopped = false
 193         f.result = make(chan Event, DefaultChanSize)
 194 }
 195
 196 func (f *RaceFreeFakeWatcher) ResultChan() <-chan Event {
 197         f.Lock()
 198         defer f.Unlock()
 199         return f.result
 200 }
 201
 202 // Add sends an add event.
 203 func (f *RaceFreeFakeWatcher) Add(obj runtime.Object) {
 204         f.Lock()
 205         defer f.Unlock()
 206         if !f.Stopped {
 207                 select {
 208                 case f.result <- Event{Added, obj}:
 209                         return
 210                 default:
 211                         panic(fmt.Errorf("channel full"))
 212                 }
 213         }
 214 }
 215
 216 // Modify sends a modify event.
 217 func (f *RaceFreeFakeWatcher) Modify(obj runtime.Object) {
 218         f.Lock()
 219         defer f.Unlock()
 220         if !f.Stopped {
 221                 select {
 222                 case f.result <- Event{Modified, obj}:
 223                         return
 224                 default:
 225                         panic(fmt.Errorf("channel full"))
 226                 }
 227         }
 228 }
 229
 230 // Delete sends a delete event.
 231 func (f *RaceFreeFakeWatcher) Delete(lastValue runtime.Object) {
 232         f.Lock()
 233         defer f.Unlock()
 234         if !f.Stopped {
 235                 select {
 236                 case f.result <- Event{Deleted, lastValue}:
 237                         return
 238                 default:
 239                         panic(fmt.Errorf("channel full"))
 240                 }
 241         }
 242 }
 243
 244 // Error sends an Error event.
 245 func (f *RaceFreeFakeWatcher) Error(errValue runtime.Object) {
 246         f.Lock()
 247         defer f.Unlock()
 248         if !f.Stopped {
 249                 select {
 250                 case f.result <- Event{Error, errValue}:
 251                         return
 252                 default:
 253                         panic(fmt.Errorf("channel full"))
 254                 }
 255         }
 256 }
 257
 258 // Action sends an event of the requested type, for table-based testing.
 259 func (f *RaceFreeFakeWatcher) Action(action EventType, obj runtime.Object) {
 260         f.Lock()
 261         defer f.Unlock()
 262         if !f.Stopped {
 263                 select {
 264                 case f.result <- Event{action, obj}:
 265                         return
 266                 default:
 267                         panic(fmt.Errorf("channel full"))
 268                 }
 269         }
 270 }
 271
 272 // ProxyWatcher lets you wrap your channel in watch Interface. Threadsafe.
 273 type ProxyWatcher struct {
 274         result chan Event
 275         stopCh chan struct{}
 276
 277         mutex   sync.Mutex
 278         stopped bool
 279 }
 280
 281 var _ Interface = &ProxyWatcher{}
 282
 283 // NewProxyWatcher creates new ProxyWatcher by wrapping a channel
 284 func NewProxyWatcher(ch chan Event) *ProxyWatcher {
 285         return &ProxyWatcher{
 286                 result:  ch,
 287                 stopCh:  make(chan struct{}),
 288                 stopped: false,
 289         }
 290 }
 291
 292 // Stop implements Interface
 293 func (pw *ProxyWatcher) Stop() {
 294         pw.mutex.Lock()
 295         defer pw.mutex.Unlock()
 296         if !pw.stopped {
 297                 pw.stopped = true
 298                 close(pw.stopCh)
 299         }
 300 }
 301
 302 // Stopping returns true if Stop() has been called
 303 func (pw *ProxyWatcher) Stopping() bool {
 304         pw.mutex.Lock()
 305         defer pw.mutex.Unlock()
 306         return pw.stopped
 307 }
 308
 309 // ResultChan implements Interface
 310 func (pw *ProxyWatcher) ResultChan() <-chan Event {
 311         return pw.result
 312 }
 313
 314 // StopChan returns stop channel
 315 func (pw *ProxyWatcher) StopChan() <-chan struct{} {
 316         return pw.stopCh
 317 }