xstream

Documentation

Index

• func FilterFunc(f interface{}) (nodes.UnFunc, error)
• func FlatMapFunc(f interface{}) (nodes.UnFunc, error)
• func GetConfAsBool(file, key string) (bool, error)
• func GetConfAsFloat(file, key string) (float64, error)
• func GetConfAsInt(file, key string) (int, error)
• func GetConfAsString(file, key string) (string, error)
• func MapFunc(f interface{}) (nodes.UnFunc, error)
• func ProcessFunc(f interface{}) (nodes.UnFunc, error)
• func Transform(op nodes.UnOperation, name string, bufferLength int) *nodes.UnaryOperator
• type Conf
• type PrintableTopo
• type TopologyNew
  • func NewWithNameAndQos(name string, qos api.Qos, checkpointInterval int) (*TopologyNew, error)
  • func (s *TopologyNew) AddOperator(inputs []api.Emitter, operator nodes.OperatorNode) *TopologyNew
  • func (s *TopologyNew) AddSink(inputs []api.Emitter, snk *nodes.SinkNode) *TopologyNew
  • func (s *TopologyNew) AddSrc(src *nodes.SourceNode) *TopologyNew
  • func (s *TopologyNew) Cancel()
  • func (s *TopologyNew) GetContext() api.StreamContext
  • func (s *TopologyNew) GetCoordinator() *checkpoints.Coordinator
  • func (s *TopologyNew) GetMetrics() (keys []string, values []interface{})
  • func (s *TopologyNew) GetTopo() *PrintableTopo
  • func (s *TopologyNew) Open() <-chan error

Functions

func FilterFunc

func FilterFunc(f interface{}) (nodes.UnFunc, error)

FilterFunc returns a unary function (api.UnFunc) which applies the user-defined filtering to apply predicates that filters out data items from being included in the downstream. The provided user-defined function must be of type:

func(T)bool - where T is the type of incoming data item, bool is the value of the predicate

When the user-defined function returns false, the current processed data item will not be placed in the downstream processing.

func FlatMapFunc

func FlatMapFunc(f interface{}) (nodes.UnFunc, error)

FlatMapFunc returns an unary function which applies a user-defined function which takes incoming comsite items and deconstruct them into individual items which can then be re-streamed. The type for the user-defined function is:

func (T) R - where R is the original item, R is a slice of decostructed items

The slice returned should be restreamed by placing each item onto the stream for downstream processing.

func GetConfAsBool

func GetConfAsBool(file, key string) (bool, error)

func GetConfAsFloat

func GetConfAsFloat(file, key string) (float64, error)

func GetConfAsInt

func GetConfAsInt(file, key string) (int, error)

func GetConfAsString

func GetConfAsString(file, key string) (string, error)

func MapFunc

func MapFunc(f interface{}) (nodes.UnFunc, error)

MapFunc returns an unary function which applies the user-defined function which maps, one-to-one, the incomfing value to a new value. The user-defined function must be of type:

func(T) R - where T is the incoming item, R is the type of the returned mapped item

func ProcessFunc

func ProcessFunc(f interface{}) (nodes.UnFunc, error)

ProcessFunc returns a unary function which applies the specified user-defined function that processes data items from upstream and returns a result value. The provided function must be of type:

func(T) R
where T is the type of incoming item
R the type of returned processed item

func Transform

func Transform(op nodes.UnOperation, name string, bufferLength int) *nodes.UnaryOperator

Types

type Conf

type Conf map[string]interface{}

type PrintableTopo

type PrintableTopo struct {
	Sources []string            `json:"sources"`
	Edges   map[string][]string `json:"edges"`
}

type TopologyNew

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

func NewWithNameAndQos

func NewWithNameAndQos(name string, qos api.Qos, checkpointInterval int) (*TopologyNew, error)

func (*TopologyNew) AddOperator

func (s *TopologyNew) AddOperator(inputs []api.Emitter, operator nodes.OperatorNode) *TopologyNew

func (*TopologyNew) AddSink

func (s *TopologyNew) AddSink(inputs []api.Emitter, snk *nodes.SinkNode) *TopologyNew

func (*TopologyNew) AddSrc

func (s *TopologyNew) AddSrc(src *nodes.SourceNode) *TopologyNew

func (*TopologyNew) Cancel

func (s *TopologyNew) Cancel()

may be called multiple times so must be idempotent

func (*TopologyNew) GetContext

func (s *TopologyNew) GetContext() api.StreamContext

func (*TopologyNew) GetCoordinator

func (s *TopologyNew) GetCoordinator() *checkpoints.Coordinator

func (*TopologyNew) GetMetrics

func (s *TopologyNew) GetMetrics() (keys []string, values []interface{})

func (*TopologyNew) GetTopo

func (s *TopologyNew) GetTopo() *PrintableTopo

func (*TopologyNew) Open

func (s *TopologyNew) Open() <-chan error