source

Documentation

Index

• Variables
• type ActivateWatchableDataSource
• type DataSource
• type DataSources
  • func NewDataSources() *DataSources
  • func (dss *DataSources) Add(ds DataSource) error
  • func (dss *DataSources) Get(name string) DataSource
  • func (dss *DataSources) Info() map[string]Info
  • func (dss *DataSources) Remove(name string) DataSource
  • func (dss *DataSources) SetObserver(obs DataSourcesObserver)
• type DataSourcesObserver
• type DrainableSource
• type GenericDataFormat
• type GenericDataFormatFunc
  • func (fn GenericDataFormatFunc) FrameItem(buf []byte) ([]byte, error)
  • func (fn GenericDataFormatFunc) MarshalGet(item interface{}) ([]byte, error)
  • func (fn GenericDataFormatFunc) MarshalInit(item interface{}) ([]byte, error)
  • func (fn GenericDataFormatFunc) MarshalItem(item interface{}) ([]byte, error)
• type GenericDataSource
• type GenericDataSourceFormats
• type GenericDataWatcher
• type GetableDataSource
• type Info
  • func GetInfo(ds DataSource) Info
• type ItemDataSource
• type ItemWatcher
• type ItemWatcherBatchFunc
  • func (batchFunc ItemWatcherBatchFunc) HandleItem(item []byte) error
  • func (batchFunc ItemWatcherBatchFunc) HandleItems(items [][]byte) error
• type ItemWatcherFunc
  • func (itemFunc ItemWatcherFunc) HandleItem(item []byte) error
  • func (itemFunc ItemWatcherFunc) HandleItems(items [][]byte) error
• type TextTemplatedSource
• type WatchInitableDataSource
• type WatchableDataSource

Variables

var (
	// ErrUnsupportedFormat should be returned by DataSource.Get and
	// DataSource.Watch if the requested format is not supported.
	ErrUnsupportedFormat = errors.New("unsupported format")

	// ErrNotGetable should be returned by DataSource.Get if the data source
	// does not support get.
	ErrNotGetable = errors.New("get not supported, data source is watch-only")

	// ErrNotWatchable should be returned by DataSource.Get if the data source
	// does not support watch.
	ErrNotWatchable = errors.New("watch not supported, data source is get-only")
)

var ErrSourceAlreadyDefined = errors.New("data source already defined")

Types

type ActivateWatchableDataSource

type ActivateWatchableDataSource interface {
	WatchableDataSource

	// Activate gets called when the GenericDataWatcher transitions from
	// inactive to active.  It may be used by implementations to start or
	// trigger any resources needed to generate items to pass to the set
	// GenericDataWatcher.
	Activate()
}

ActivateWatchableDataSource is an optional interface that WatchableDataSources may implement to get notified about source activation.

type DataSource

type DataSource interface {
	// Name returns the unique identifier (GWR noun path) for this source.
	Name() string

	// Formats returns a list of supported format name strings.  All
	// implemented formats must be listed.  At least "json" must be supported.
	Formats() []string

	// Attrs returnts arbitrary descriptive data about the data source.  This
	// data is exposed be the /meta/nouns data source.
	//
	// TODO: standardize and document common fields; current ideas include:
	// - affording get-only or watch-only
	// - affording sampling config (%-age, N-per-t, etc)
	// - whether this data source is lossy (likely the default) or whether
	//   attempts should be taken to drop no item (opt-in edge case); this
	//   could be used internally at least to switch between an implementation
	//   that drops data when buffers fill, or blocks and provides back
	//   pressure.
	Attrs() map[string]interface{}

	// Get implementations:
	// - may return ErrNotGetable if get is not supported by the data source
	// - if the format is not support then ErrUnsupportedFormat must be returned
	// - must format and write any available data to the supplied io.Writer
	// - should return any write error
	Get(format string, w io.Writer) error

	// Watch implementations:
	// - may return ErrNotWatchable if watch is not supported by the data
	//   source
	// - if the format is not support then ErrUnsupportedFormat must be returned
	// - may format and write initial data to the supplied io.Writer; any
	//   initial write error must be returned
	// - may retain and write to the supplied io.Writer indefinately until it
	//   returns a write error
	//
	// Note that at this level, data sources are responsible for both item
	// marshalling and stream framing.
	//
	// Framing for the required "json" format is as follows:
	// - JSON must be encoded in compact (no intermediate whitespace) form
	// - each JSON record must be separated by a newline "\n"
	//
	// Framing for the required "text" format is as follows:
	// - any initial stream data should be followed by a blank line (double new
	//   line "\n\n")
	// - items should be separated by newlines
	// - if an item's text form takes up multiple lines, it should either use
	//   indentation or a double blank line to separate itself from siblings
	Watch(format string, w io.Writer) error
}

DataSource is the low-level interface implemented by all data sources.

On formats, implementanions: - must implement format == "json" - should implement format == "text" - may implement any other formats that make sense for them

Further implementation requirements are listed within the interface functions' documentation.

type DataSources

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

DataSources is a flat collection of DataSources with a meta introspection data source.

func NewDataSources

func NewDataSources() *DataSources

NewDataSources creates a DataSources structure an sets up its "/meta/nouns" data source.

func (*DataSources) Add

func (dss *DataSources) Add(ds DataSource) error

Add a DataSource, if none is already defined for the given name.

func (*DataSources) Get

func (dss *DataSources) Get(name string) DataSource

Get returns the named data source or nil if none is defined.

func (*DataSources) Info

func (dss *DataSources) Info() map[string]Info

Info returns a map of info about all sources.

func (*DataSources) Remove

func (dss *DataSources) Remove(name string) DataSource

Remove a DataSource by name, if any exsits. Returns the source removed, nil if none was defined.

func (*DataSources) SetObserver

func (dss *DataSources) SetObserver(obs DataSourcesObserver)

SetObserver sets the (single!) observer of data source changes; if nil is passed, observation is disabled.

type DataSourcesObserver

type DataSourcesObserver interface {
	SourceAdded(ds DataSource)
	SourceRemoved(ds DataSource)
}

DataSourcesObserver is an interface to observe data sources changes.

Observation happens after after the source has been added (resp. removed).

type DrainableSource

type DrainableSource interface {
	DataSource
	Drain()
}

DrainableSource is a DataSource that can be drained. Draining a source should flush any unsent data, and then close any remaining Watch writers.

type GenericDataFormat

type GenericDataFormat interface {
	// MarshalGet serializes the passed data from GenericDataSource.Get.
	MarshalGet(interface{}) ([]byte, error)

	// MarshalInit serializes the passed data from GenericDataSource.GetInit.
	MarshalInit(interface{}) ([]byte, error)

	// MarshalItem serializes data passed to a GenericDataWatcher.
	MarshalItem(interface{}) ([]byte, error)

	// FrameItem wraps a MarshalItem-ed byte buffer for a watch stream.
	FrameItem([]byte) ([]byte, error)
}

GenericDataFormat provides both a data marshaling protocol and a framing protocol for the watch stream. Any marshaling or framing error should cause a break in any watch streams subscribed to this format.

type GenericDataFormatFunc

type GenericDataFormatFunc func(interface{}) ([]byte, error)

GenericDataFormatFunc is a convenience for implement simple single-function formats with newline framing.

func (GenericDataFormatFunc) FrameItem

func (fn GenericDataFormatFunc) FrameItem(buf []byte) ([]byte, error)

FrameItem wraps a MarshalItem-ed byte buffer for a watch stream.

func (GenericDataFormatFunc) MarshalGet

func (fn GenericDataFormatFunc) MarshalGet(item interface{}) ([]byte, error)

MarshalGet calls the wrapped function.

func (GenericDataFormatFunc) MarshalInit

func (fn GenericDataFormatFunc) MarshalInit(item interface{}) ([]byte, error)

MarshalInit calls the wrapped function.

func (GenericDataFormatFunc) MarshalItem

func (fn GenericDataFormatFunc) MarshalItem(item interface{}) ([]byte, error)

MarshalItem calls the wrapped function.

type GenericDataSource

type GenericDataSource interface {
	// Name must return the name of the data source; see DataSource.Name.
	Name() string
}

GenericDataSource is a format-agnostic data source

type GenericDataSourceFormats

type GenericDataSourceFormats interface {
	Formats() map[string]GenericDataFormat
}

GenericDataSourceFormats is implemented by generic data sources to define additional formats beyond the default json and templated text ones.

type GenericDataWatcher

type GenericDataWatcher interface {
	Active() bool

	// HandleItem is called with a single item of generic unmarshaled data.
	HandleItem(item interface{}) bool

	// HandleItem is called with a batch of generic unmarshaled data.
	HandleItems(items []interface{}) bool
}

GenericDataWatcher is the interface for the watcher passed to GenericDataSource.SetWatcher. Both single-item and batch methods are provided.

type GetableDataSource

type GetableDataSource interface {
	GenericDataSource

	// Get should return any data available for the data source.
	Get() interface{}
}

GetableDataSource is the interface implemented by GenericDataSources that support Get. If a GenericDataSource does not implement GetableDataSource, then any gets for it return source.ErrNotGetable.

type Info

type Info struct {
	Formats []string               `json:"formats"`
	Attrs   map[string]interface{} `json:"attrs"`
}

Info is a convenience info descriptor about a data source.

func GetInfo

func GetInfo(ds DataSource) Info

GetInfo returns a structure that contains format and other information about a given data source.

type ItemDataSource

type ItemDataSource interface {
	// WatchItems implementations have all of the semantics of
	// DataSource.Watch, just over an ItemWatcher instead of an io.Writer.
	//
	// If the data source naturally has batches of items on hand it may call
	// watcher.HandleItems.
	//
	// Passed watchers must be discarded after either HandleItem or
	// HandleItems returns a non-nil error.
	WatchItems(format string, watcher ItemWatcher) error
}

ItemDataSource is an interface implemented by a data source to provide marshaled but unframed streams of Watch items. When implemented protocol libraries can add protocol-specific builtin framing.

type ItemWatcher

type ItemWatcher interface {
	// HandleItem gets a single marshaled item, and should return any framing
	// or write error.
	HandleItem(item []byte) error

	// HandleItems gets a batch of marshaled items, and should return any
	// framing or write error.
	HandleItems(items [][]byte) error
}

ItemWatcher is the interface passed to ItemSource.WatchItems. Any error returned by either HandleItem or HandleItems indicates that this watcher should not be called with more items.

type ItemWatcherBatchFunc

type ItemWatcherBatchFunc func([][]byte) error

ItemWatcherBatchFunc is a convenience type for watching with a simple batch function. The batch function should return any framing or write error.

func (ItemWatcherBatchFunc) HandleItem

func (batchFunc ItemWatcherBatchFunc) HandleItem(item []byte) error

HandleItem calls the wrapped function with a singleton batch.

func (ItemWatcherBatchFunc) HandleItems

func (batchFunc ItemWatcherBatchFunc) HandleItems(items [][]byte) error

HandleItems just calls the wrapped function.

type ItemWatcherFunc

type ItemWatcherFunc func([]byte) error

ItemWatcherFunc is a convenience type for watching with a simple per-item function. The item function should return any framing or write error.

func (ItemWatcherFunc) HandleItem

func (itemFunc ItemWatcherFunc) HandleItem(item []byte) error

HandleItem just calls the wrapped function.

func (ItemWatcherFunc) HandleItems

func (itemFunc ItemWatcherFunc) HandleItems(items [][]byte) error

HandleItems calls the wrapped function for each item in the batch, stopping on and returning the first error.

type TextTemplatedSource

type TextTemplatedSource interface {
	// TextTemplate returns the text/template that is used to construct a
	// TemplatedMarshal to implement the "text" format for this data source.
	TextTemplate() *template.Template
}

TextTemplatedSource is implemented by generic data sources to provide a convenience template for the "text" format.

type WatchInitableDataSource

type WatchInitableDataSource interface {
	WatchableDataSource

	// GetInit should returns initial data to send to new watch streams.
	WatchInit() interface{}
}

WatchInitableDataSource is the interface that a WatchableDataSource should implement if it wants to provide an initial data item to all new watch streams.

type WatchableDataSource

type WatchableDataSource interface {
	GenericDataSource

	// SetWatcher sets the watcher.
	//
	// Implementations should retain a reference to the last passed watcher,
	// and need not retain multiple; in the usual case this method will only be
	// called once per data source lifecycle.
	//
	// Implementations should pass items to watcher.HandleItem and/or
	// watcher.HandleItems methods.
	//
	// Implementations may use watcher.Active to avoid building items which
	// would just be thrown out by a call to HandleItem(s).
	SetWatcher(watcher GenericDataWatcher)
}

WatchableDataSource is the interface implemented by GenericDataSources that support Watch. If a GenericDataSource does not implement WatchableDataSource, then any watches for it return source.ErrNotWatchable.