watch

Documentation

Overview

Package watch contains a generic watchable interface, and a fake for testing code that uses the watch interface.

Index

• type Decoder
• type Event
• type EventType
• type FakeWatcher
  • func NewFake() *FakeWatcher
  • func (f *FakeWatcher) Action(action EventType, obj runtime.Object)
  • func (f *FakeWatcher) Add(obj runtime.Object)
  • func (f *FakeWatcher) Delete(lastValue runtime.Object)
  • func (f *FakeWatcher) Error(errValue runtime.Object)
  • func (f *FakeWatcher) Modify(obj runtime.Object)
  • func (f *FakeWatcher) ResultChan() <-chan Event
  • func (f *FakeWatcher) Stop()
• type FilterFunc
• type Interface
  • func Filter(w Interface, f FilterFunc) Interface
• type Mux
  • func NewMux(queueLength int) *Mux
  • func (m *Mux) Action(action EventType, obj runtime.Object)
  • func (m *Mux) Shutdown()
  • func (m *Mux) Watch() Interface
• type StreamWatcher
  • func NewStreamWatcher(d Decoder) *StreamWatcher
  • func (sw *StreamWatcher) ResultChan() <-chan Event
  • func (sw *StreamWatcher) Stop()

Types

type Decoder

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()
}

Decoder allows StreamWatcher to watch any stream for which a Decoder can be written.

type Event

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 Error: *api.Status is recommended; other types may make sense
	//    depending on context.
	Object runtime.Object
}

Event represents a single event to a watched resource.

type EventType

type EventType string

EventType defines the possible types of events.

const (
	Added    EventType = "ADDED"
	Modified EventType = "MODIFIED"
	Deleted  EventType = "DELETED"
	Error    EventType = "ERROR"
)

type FakeWatcher

type FakeWatcher struct {
	Stopped bool
	sync.Mutex
	// contains filtered or unexported fields
}

FakeWatcher lets you test anything that consumes a watch.Interface; threadsafe.

func NewFake

func NewFake() *FakeWatcher

func (*FakeWatcher) Action

func (f *FakeWatcher) Action(action EventType, obj runtime.Object)

Action sends an event of the requested type, for table-based testing.

func (*FakeWatcher) Add

func (f *FakeWatcher) Add(obj runtime.Object)

Add sends an add event.

func (*FakeWatcher) Delete

func (f *FakeWatcher) Delete(lastValue runtime.Object)

Delete sends a delete event.

func (*FakeWatcher) Error

func (f *FakeWatcher) Error(errValue runtime.Object)

Error sends an Error event.

func (*FakeWatcher) Modify

func (f *FakeWatcher) Modify(obj runtime.Object)

Modify sends a modify event.

func (*FakeWatcher) ResultChan

func (f *FakeWatcher) ResultChan() <-chan Event

func (*FakeWatcher) Stop

func (f *FakeWatcher) Stop()

Stop implements Interface.Stop().

type FilterFunc

type FilterFunc func(in Event) (out Event, keep bool)

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 Interface

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, this channel will be closed, in which case the
	// watch should be completely cleaned up.
	ResultChan() <-chan Event
}

Interface can be implemented by anything that knows how to watch and report changes.

func Filter

func Filter(w Interface, f FilterFunc) Interface

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.

type Mux

type Mux struct {
	// contains filtered or unexported fields
}

Mux distributes event notifications among any number of watchers. Every event is delivered to every watcher.

func NewMux

func NewMux(queueLength int) *Mux

NewMux creates a new Mux. queueLength is the maximum number of events to queue. When queueLength is 0, Action will block until any prior event has been completely distributed. It is guaranteed that events will be distibuted in the order in which they ocurr, but the order in which a single event is distributed among all of the watchers is unspecified.

func (*Mux) Action

func (m *Mux) Action(action EventType, obj runtime.Object)

Action distributes the given event among all watchers.

func (*Mux) Shutdown

func (m *Mux) Shutdown()

Shutdown disconnects all watchers (but any queued events will still be distributed). You must not call Action after calling Shutdown.

func (*Mux) Watch

func (m *Mux) Watch() Interface

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.

type StreamWatcher

type StreamWatcher struct {
	sync.Mutex
	// contains filtered or unexported fields
}

StreamWatcher turns any stream for which you can write a Decoder interface into a watch.Interface.

func NewStreamWatcher

func NewStreamWatcher(d Decoder) *StreamWatcher

NewStreamWatcher creates a StreamWatcher from the given decoder.

func (*StreamWatcher) ResultChan

func (sw *StreamWatcher) ResultChan() <-chan Event

ResultChan implements Interface.

func (*StreamWatcher) Stop

func (sw *StreamWatcher) Stop()

Stop implements Interface.