cmd/bpa-operator/vendor/k8s.io/apimachinery/pkg/watch/filter.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         "sync"
  21 )
  22
  23 // FilterFunc should take an event, possibly modify it in some way, and return
  24 // the modified event. If the event should be ignored, then return keep=false.
  25 type FilterFunc func(in Event) (out Event, keep bool)
  26
  27 // Filter passes all events through f before allowing them to pass on.
  28 // Putting a filter on a watch, as an unavoidable side-effect due to the way
  29 // go channels work, effectively causes the watch's event channel to have its
  30 // queue length increased by one.
  31 //
  32 // WARNING: filter has a fatal flaw, in that it can't properly update the
  33 // Type field (Add/Modified/Deleted) to reflect items beginning to pass the
  34 // filter when they previously didn't.
  35 //
  36 func Filter(w Interface, f FilterFunc) Interface {
  37         fw := &filteredWatch{
  38                 incoming: w,
  39                 result:   make(chan Event),
  40                 f:        f,
  41         }
  42         go fw.loop()
  43         return fw
  44 }
  45
  46 type filteredWatch struct {
  47         incoming Interface
  48         result   chan Event
  49         f        FilterFunc
  50 }
  51
  52 // ResultChan returns a channel which will receive filtered events.
  53 func (fw *filteredWatch) ResultChan() <-chan Event {
  54         return fw.result
  55 }
  56
  57 // Stop stops the upstream watch, which will eventually stop this watch.
  58 func (fw *filteredWatch) Stop() {
  59         fw.incoming.Stop()
  60 }
  61
  62 // loop waits for new values, filters them, and resends them.
  63 func (fw *filteredWatch) loop() {
  64         defer close(fw.result)
  65         for event := range fw.incoming.ResultChan() {
  66                 filtered, keep := fw.f(event)
  67                 if keep {
  68                         fw.result <- filtered
  69                 }
  70         }
  71 }
  72
  73 // Recorder records all events that are sent from the watch until it is closed.
  74 type Recorder struct {
  75         Interface
  76
  77         lock   sync.Mutex
  78         events []Event
  79 }
  80
  81 var _ Interface = &Recorder{}
  82
  83 // NewRecorder wraps an Interface and records any changes sent across it.
  84 func NewRecorder(w Interface) *Recorder {
  85         r := &Recorder{}
  86         r.Interface = Filter(w, r.record)
  87         return r
  88 }
  89
  90 // record is a FilterFunc and tracks each received event.
  91 func (r *Recorder) record(in Event) (Event, bool) {
  92         r.lock.Lock()
  93         defer r.lock.Unlock()
  94         r.events = append(r.events, in)
  95         return in, true
  96 }
  97
  98 // Events returns a copy of the events sent across this recorder.
  99 func (r *Recorder) Events() []Event {
 100         r.lock.Lock()
 101         defer r.lock.Unlock()
 102         copied := make([]Event, len(r.events))
 103         copy(copied, r.events)
 104         return copied
 105 }