stream

Documentation

Index

• func InputStream() (result *inputStream)
• type Context
• type DataStream
  • func Stream(from *DataStream, handler FilterHandler) (result *DataStream)
  • func (s *DataStream) BindFault(f PushHandler)
  • func (s *DataStream) BindOut(f PushHandler)
  • func (s *DataStream) Context() *Context
  • func (s *DataStream) Fault() *FaultStream
  • func (s *DataStream) Filter(f func(value interface{}) bool) *DataStream
  • func (s *DataStream) FilterByField(fieldName string, fieldValue interface{}) *DataStream
  • func (s *DataStream) ID(id string) *DataStream
  • func (s *DataStream) KeyByFieldSync(fields ...string) *DataStream
  • func (s *DataStream) Map(f func(value interface{}) (interface{}, error)) *DataStream
  • func (s *DataStream) Name(name string) *DataStream
  • func (s *DataStream) Print()
  • func (s *DataStream) Watermark(f func(value interface{}) time.Time) *DataStream
• type Event
• type FaultStream
• type FilterHandler
• type IInputStream
• type IStreamSource
• type KeyedEvent
• type KeyedStream
• type PushHandler

Functions

func InputStream

func InputStream() (result *inputStream)

Types

type Context

type Context struct {
}

type DataStream

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

func Stream

func Stream(from *DataStream, handler FilterHandler) (result *DataStream)

func (*DataStream) BindFault

func (s *DataStream) BindFault(f PushHandler)

func (*DataStream) BindOut

func (s *DataStream) BindOut(f PushHandler)

func (*DataStream) Context

func (s *DataStream) Context() *Context

func (*DataStream) Fault

func (s *DataStream) Fault() *FaultStream

func (*DataStream) Filter

func (s *DataStream) Filter(f func(value interface{}) bool) *DataStream

func (*DataStream) FilterByField

func (s *DataStream) FilterByField(fieldName string, fieldValue interface{}) *DataStream

func (*DataStream) ID

func (s *DataStream) ID(id string) *DataStream

func (*DataStream) KeyByFieldSync

func (s *DataStream) KeyByFieldSync(fields ...string) *DataStream

func (*DataStream) Map

func (s *DataStream) Map(f func(value interface{}) (interface{}, error)) *DataStream

func (*DataStream) Name

func (s *DataStream) Name(name string) *DataStream

func (*DataStream) Print

func (s *DataStream) Print()

func (*DataStream) Watermark

func (s *DataStream) Watermark(f func(value interface{}) time.Time) *DataStream

type Event

type Event struct {
	Timestamp time.Time `json:"tm"`
	Payload   interface{}
}

type FaultStream

type FaultStream struct {
	DataStream
}

type FilterHandler

type FilterHandler func(event *Event) (*Event, error)

type IInputStream

type IInputStream interface {
	Push(event interface{})
}

type IStreamSource

type IStreamSource interface {
	Out(f PushHandler)
	FaultOut(f PushHandler)
}

type KeyedEvent

type KeyedEvent struct {
	Key   interface{}
	Event Event
}

type KeyedStream

type KeyedStream struct {
	Sync chan chan KeyedEvent
}

type PushHandler

type PushHandler func(event *Event)