saw

type Datum ¶

type Datum struct {
	Key   DatumKey
	Value interface{}
}

Datum is the data item passed between saws. Datum is a key-value pair where key must be string and value can by anything. The optional SortOrder specifies optimal order when datums with same key get aggregated.

type DatumKey ¶

type DatumKey string

type DatumSeqFunc ¶

type DatumSeqFunc func(datum Datum) SeqID

type ExportSaw ¶

type ExportSaw interface {
	Export() (interface{}, error)
}

Saw can optionally provide Export() interface, it provides a snapshot of its current state, which can be later merged to another saw

type Hub ¶

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

Hub is a simple pubsub to allow loosely coupled communication between saws Saws can Register with topic(s) it subscribes to, or Publish datum to a topic.

It's a simple local, sync implementation only for better pipeline program structure, and it should keep it as it is. parallel, async computing, should be addressed by Queues and Pars, implemented by each individual Saw.

func NewHub(varPrefix string) *Hub

func (hub *Hub) Publish(id TopicID, datum Datum)

func (hub *Hub) Register(saw Saw, subscribes ...TopicID)

type JSONDecoder ¶

type JSONDecoder struct {
	ValueType reflect.Type
}

func NewJSONDecoder(example interface{}) JSONDecoder

func (jd JSONDecoder) DecodeValue(buf []byte) (interface{}, error)

type JSONEncoder ¶

type JSONEncoder struct{}

func (je JSONEncoder) EncodeValue(value interface{}, buf []byte) ([]byte, error)

type MergeSaw ¶

type MergeSaw interface {
	MergeFrom(other interface{}) error
}

Saw can merge Export() result of other saws, so that Saws can be aggregated or migrated between instances

type ProtoDecoder ¶

type ProtoDecoder struct {
	ValueType reflect.Type
}

func NewProtoDecoder(example interface{}) ProtoDecoder

func (pd ProtoDecoder) DecodeValue(buf []byte) (interface{}, error)

type ProtoEncoder ¶

type ProtoEncoder struct{}

func (pe ProtoEncoder) EncodeValue(value interface{}, buf []byte) ([]byte, error)

type Saw ¶

type Saw interface {
	// Feeds a new data point into Saw.
	Emit(v Datum) error

	// Release all resources and returns final computation result.
	Result(ctx context.Context) (interface{}, error)
}

Saw is the basic computation unit, it's largely a state machine.

In general, implementation should allow concurrent call to Emit() for good parallelism, it can be archive by having a stateless saw (see Transfrom), or make it managed by concurrent tables like table.MemTable (see table subpackage).

type SawNoResult ¶

type SawNoResult struct{}

func (snr SawNoResult) Result(ctx context.Context) (interface{}, error)

type SeqID ¶

type SeqID int64

func (seq SeqID) Advance(x int) SeqID

func (seq SeqID) DistanceFrom(other SeqID) int

type TopicID ¶

type TopicID string

type TransformFunc ¶

type TransformFunc func(input Datum) (output Datum, err error)

type TransformSpec ¶

type TransformSpec struct {
	Name      string
	Transfrom TransformFunc
	Inputs    []TopicID
	Outputs   []TopicID
}

TransformSpec configures a Transform.

Transform is a stateless Saw, that for each Datum input --- normally from Hub, it calls a pure TransformFunc to get a Datum output, then publish the output to a set of Hub topics.

Use RegisterTransform() to create a Transform saw and register it to a Hub.

type ValueDecoder ¶

type ValueDecoder interface {
	DecodeValue(buf []byte) (interface{}, error)
}

type ValueEncoder ¶

type ValueEncoder interface {
	EncodeValue(value interface{}, buf []byte) ([]byte, error)
}

type VarFloat ¶

type VarFloat interface {
	Add(delta float64)
	Set(value float64)
}

func ReportFloat(ns, name string) VarFloat

type VarInt ¶

type VarInt interface {
	Add(delta int64)
	Set(value int64)
}

func ReportInt(ns, name string) VarInt

type Window ¶

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

Window implements a sliding window of saws. Window keeps finite set of saws, called frames, each corresponded with a SeqID. Window keeps WindowSpec.WindowSize of latest frames with largest, contintued SeqID.

For every input, it calls WindowSpec.SeqFunc to get SeqID of a datum and route it to the frame if it's still in sliding window. Window assumes datum's SeqID are dense, rougly incremental overtime, or you will get too many datums dropped.

When window needed to be slided, Result() of old frames will be called in their seperate gorountines. In Window.Result(), all frames' Result() will be called in this maner as they all been slide away. Window doesn't care frame's Result() return.

func NewWindow(spec WindowSpec) *Window

func (win *Window) AllFrames() map[SeqID]Saw

func (win *Window) Emit(datum Datum) error

func (win *Window) LatestFrame() (seq SeqID, frame Saw)

func (win *Window) Result(ctx context.Context) (result interface{}, err error)

type WindowFrameFactory ¶

type WindowFrameFactory func(name string, seq SeqID) (Saw, error)

type WindowSpec ¶

type WindowSpec struct {
	Name          string
	FrameFactory  WindowFrameFactory
	SeqFunc       DatumSeqFunc
	WindowSize    int
	MaxSeqAdvance int
}