watcher

Documentation

Index

• Constants
• type BaseWatcher
  • func New(s store.Store, ctx context.Context, key string, convert ConvertFunc, ...) (*BaseWatcher, error)
  • func (w *BaseWatcher) Get(prefix string) (map[string]interface{}, error)
  • func (w *BaseWatcher) Status() Status
  • func (w *BaseWatcher) Watch(prefix string, ctx context.Context) (<-chan *Event, error)
• type Cacher
  • func NewCacher(data map[string]interface{}) *Cacher
  • func (c *Cacher) Count() int
  • func (c *Cacher) Delete(key string, recursive bool) []string
  • func (c *Cacher) Get(key string) map[string]interface{}
  • func (c *Cacher) GetAll() map[string]interface{}
  • func (c *Cacher) GetAllKeys() []string
  • func (c *Cacher) Put(key string, value interface{})
  • func (c *Cacher) Reset(data map[string]interface{})
• type ConvertFunc
• type Event
• type Receiver
• type Sender
  • func NewSender(data map[string]interface{}) *Sender
  • func (s *Sender) Broadcast(keys []string, action store.Action)
  • func (s *Sender) Watch(key string, ctx context.Context) <-chan *Event
• type Status
• type Watcher

Constants

const (
	// NIL represents a nil watcher, which will watch nothing
	NIL = "nil"
	// CONFIG represents a config watcher, watch /lain/config
	CONFIG = "config"
	// PODGROUP represents a podgroup watcher, watch /lain/deployd/pod_groups
	PODGROUP = "podgroup"
	// DEPENDS represents a depends watcher, watch /lain/deployd/depends
	DEPENDS = "depends"
	// NODES represents a node watcher, it's based on podgroup watcher
	NODES = "nodes"
	// CONTAINER represents a container watcher, it's based on podgroup watcher
	CONTAINER = "container"
)

Types

type BaseWatcher

type BaseWatcher struct {
	Store store.Store
	Ctx   context.Context
	*Sender
	// contains filtered or unexported fields
}

BaseWatcher having a sender, store, and convert function. It read and watch data from store, and use convert() to convert, then use sender to cache data and broadcast the event.

func New

func New(s store.Store, ctx context.Context, key string, convert ConvertFunc, ckey2skey func(string) string) (*BaseWatcher, error)

New create a new watcher

func (*BaseWatcher) Get

func (w *BaseWatcher) Get(prefix string) (map[string]interface{}, error)

Get function get the newest data for keys having `prefix` prefix. it returned error only when key is empty. it return all the data when prefix is '*'

func (*BaseWatcher) Status

func (w *BaseWatcher) Status() Status

Status return the watcher stats

func (*BaseWatcher) Watch

func (w *BaseWatcher) Watch(prefix string, ctx context.Context) (<-chan *Event, error)

Watch function watch the keys having `prefix` prefix, it watch all the keys when prefix is '*'. it return a event channel. error wil be returned only when key is empty. the return channel have the newest data with init action in it.

type Cacher

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

Cacher is a simple cache, used to cache converted data from store

func NewCacher

func NewCacher(data map[string]interface{}) *Cacher

NewCacher create a new cache and initialize it by the given data

func (*Cacher) Count

func (c *Cacher) Count() int

func (*Cacher) Delete

func (c *Cacher) Delete(key string, recursive bool) []string

Delete delete the key in cache, if recursive is true, all the keys having `key` prefix will be deleted it returns the key list which is deleted

func (*Cacher) Get

func (c *Cacher) Get(key string) map[string]interface{}

Get find the values in cache, if key was found, return the map with only one key; or it will return all the KV data which key has `key` prefix

func (*Cacher) GetAll

func (c *Cacher) GetAll() map[string]interface{}

GetAll return all the data in cache

func (*Cacher) GetAllKeys

func (c *Cacher) GetAllKeys() []string

func (*Cacher) Put

func (c *Cacher) Put(key string, value interface{})

Put set the key in cache, if value is nil, it will delete key in cache

func (*Cacher) Reset

func (c *Cacher) Reset(data map[string]interface{})

Reset will the data in cache, all the old data will be deleted, replaced by given new data

type ConvertFunc

type ConvertFunc func([]*store.KVPair) (map[string]interface{}, error)

ConvertFunc convert the data from store into a general type

type Event

type Event struct {
	// ID is a event id, it is cumulative when you get a new event
	ID uint64
	// Action is the event type, init, update, delete or error
	Action store.Action
	// Data is event data, it always return the newest data, no matter what the action is
	Data map[string]interface{}
}

Event represents a watcher event

type Receiver

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

Receiver represents a receiver of sender, which can receive data from sender

type Sender

type Sender struct {
	sync.Mutex
	*Cacher
	// contains filtered or unexported fields
}

Sender having a cache, when the data in cache changed, call Broadcast() sending new data to receivers.

func NewSender

func NewSender(data map[string]interface{}) *Sender

NewSender create a sender and initialze it's cacher by the given data

func (*Sender) Broadcast

func (s *Sender) Broadcast(keys []string, action store.Action)

Broadcast the change event keys: the changed keys in cache action: the store action

func (*Sender) Watch

func (s *Sender) Watch(key string, ctx context.Context) <-chan *Event

Watch a key in sender, this will add a new receiver in sender; the return channel will be closed when context was canceled.

type Status

type Status struct {
	NumReceivers int
	UpdateTime   time.Time
	LastEvent    store.Event
	TotalKeys    int
}

Status of a watcher

type Watcher

type Watcher interface {
	Get(prefix string) (map[string]interface{}, error)
	Watch(prefix string, ctx context.Context) (<-chan *Event, error)
	Status() Status
}