Scott Baker | e7144bc | 2019-10-01 14:16:47 -0700 | [diff] [blame] | 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 ( |
girishk | e7ca43b | 2019-10-10 12:30:03 +0000 | [diff] [blame^] | 20 | "fmt" |
Scott Baker | e7144bc | 2019-10-01 14:16:47 -0700 | [diff] [blame] | 21 | "io" |
| 22 | "sync" |
| 23 | |
girishk | e7ca43b | 2019-10-10 12:30:03 +0000 | [diff] [blame^] | 24 | "k8s.io/klog" |
| 25 | |
Scott Baker | e7144bc | 2019-10-01 14:16:47 -0700 | [diff] [blame] | 26 | "k8s.io/apimachinery/pkg/runtime" |
| 27 | "k8s.io/apimachinery/pkg/util/net" |
| 28 | utilruntime "k8s.io/apimachinery/pkg/util/runtime" |
Scott Baker | e7144bc | 2019-10-01 14:16:47 -0700 | [diff] [blame] | 29 | ) |
| 30 | |
| 31 | // Decoder allows StreamWatcher to watch any stream for which a Decoder can be written. |
| 32 | type Decoder interface { |
| 33 | // Decode should return the type of event, the decoded object, or an error. |
| 34 | // An error will cause StreamWatcher to call Close(). Decode should block until |
| 35 | // it has data or an error occurs. |
| 36 | Decode() (action EventType, object runtime.Object, err error) |
| 37 | |
| 38 | // Close should close the underlying io.Reader, signalling to the source of |
| 39 | // the stream that it is no longer being watched. Close() must cause any |
| 40 | // outstanding call to Decode() to return with an error of some sort. |
| 41 | Close() |
| 42 | } |
| 43 | |
girishk | e7ca43b | 2019-10-10 12:30:03 +0000 | [diff] [blame^] | 44 | // Reporter hides the details of how an error is turned into a runtime.Object for |
| 45 | // reporting on a watch stream since this package may not import a higher level report. |
| 46 | type Reporter interface { |
| 47 | // AsObject must convert err into a valid runtime.Object for the watch stream. |
| 48 | AsObject(err error) runtime.Object |
| 49 | } |
| 50 | |
Scott Baker | e7144bc | 2019-10-01 14:16:47 -0700 | [diff] [blame] | 51 | // StreamWatcher turns any stream for which you can write a Decoder interface |
| 52 | // into a watch.Interface. |
| 53 | type StreamWatcher struct { |
| 54 | sync.Mutex |
girishk | e7ca43b | 2019-10-10 12:30:03 +0000 | [diff] [blame^] | 55 | source Decoder |
| 56 | reporter Reporter |
| 57 | result chan Event |
| 58 | stopped bool |
Scott Baker | e7144bc | 2019-10-01 14:16:47 -0700 | [diff] [blame] | 59 | } |
| 60 | |
| 61 | // NewStreamWatcher creates a StreamWatcher from the given decoder. |
girishk | e7ca43b | 2019-10-10 12:30:03 +0000 | [diff] [blame^] | 62 | func NewStreamWatcher(d Decoder, r Reporter) *StreamWatcher { |
Scott Baker | e7144bc | 2019-10-01 14:16:47 -0700 | [diff] [blame] | 63 | sw := &StreamWatcher{ |
girishk | e7ca43b | 2019-10-10 12:30:03 +0000 | [diff] [blame^] | 64 | source: d, |
| 65 | reporter: r, |
Scott Baker | e7144bc | 2019-10-01 14:16:47 -0700 | [diff] [blame] | 66 | // It's easy for a consumer to add buffering via an extra |
| 67 | // goroutine/channel, but impossible for them to remove it, |
| 68 | // so nonbuffered is better. |
| 69 | result: make(chan Event), |
| 70 | } |
| 71 | go sw.receive() |
| 72 | return sw |
| 73 | } |
| 74 | |
| 75 | // ResultChan implements Interface. |
| 76 | func (sw *StreamWatcher) ResultChan() <-chan Event { |
| 77 | return sw.result |
| 78 | } |
| 79 | |
| 80 | // Stop implements Interface. |
| 81 | func (sw *StreamWatcher) Stop() { |
| 82 | // Call Close() exactly once by locking and setting a flag. |
| 83 | sw.Lock() |
| 84 | defer sw.Unlock() |
| 85 | if !sw.stopped { |
| 86 | sw.stopped = true |
| 87 | sw.source.Close() |
| 88 | } |
| 89 | } |
| 90 | |
| 91 | // stopping returns true if Stop() was called previously. |
| 92 | func (sw *StreamWatcher) stopping() bool { |
| 93 | sw.Lock() |
| 94 | defer sw.Unlock() |
| 95 | return sw.stopped |
| 96 | } |
| 97 | |
| 98 | // receive reads result from the decoder in a loop and sends down the result channel. |
| 99 | func (sw *StreamWatcher) receive() { |
| 100 | defer close(sw.result) |
| 101 | defer sw.Stop() |
| 102 | defer utilruntime.HandleCrash() |
| 103 | for { |
| 104 | action, obj, err := sw.source.Decode() |
| 105 | if err != nil { |
| 106 | // Ignore expected error. |
| 107 | if sw.stopping() { |
| 108 | return |
| 109 | } |
| 110 | switch err { |
| 111 | case io.EOF: |
| 112 | // watch closed normally |
| 113 | case io.ErrUnexpectedEOF: |
| 114 | klog.V(1).Infof("Unexpected EOF during watch stream event decoding: %v", err) |
| 115 | default: |
Scott Baker | e7144bc | 2019-10-01 14:16:47 -0700 | [diff] [blame] | 116 | if net.IsProbableEOF(err) { |
girishk | e7ca43b | 2019-10-10 12:30:03 +0000 | [diff] [blame^] | 117 | klog.V(5).Infof("Unable to decode an event from the watch stream: %v", err) |
Scott Baker | e7144bc | 2019-10-01 14:16:47 -0700 | [diff] [blame] | 118 | } else { |
girishk | e7ca43b | 2019-10-10 12:30:03 +0000 | [diff] [blame^] | 119 | sw.result <- Event{ |
| 120 | Type: Error, |
| 121 | Object: sw.reporter.AsObject(fmt.Errorf("unable to decode an event from the watch stream: %v", err)), |
| 122 | } |
Scott Baker | e7144bc | 2019-10-01 14:16:47 -0700 | [diff] [blame] | 123 | } |
| 124 | } |
| 125 | return |
| 126 | } |
| 127 | sw.result <- Event{ |
| 128 | Type: action, |
| 129 | Object: obj, |
| 130 | } |
| 131 | } |
| 132 | } |