metathings_deviced_flow

Documentation

Index

• Variables
• type Flow
• type FlowFactory
  • func NewFlowFactory(name string, args ...interface{}) (FlowFactory, error)
• type FlowFilter
• type FlowOption
• type FlowSet
• type FlowSetFactory
  • func NewFlowSetFactory(name string, args ...interface{}) (FlowSetFactory, error)
• type FlowSetFrame
• type FlowSetOption
• type RedisStreamFlowSet
  • func (rsfs *RedisStreamFlowSet) Close() (err error)
  • func (rsfs *RedisStreamFlowSet) Err() error
  • func (rsfs *RedisStreamFlowSet) Id() string
  • func (rsfs *RedisStreamFlowSet) PullFrame() <-chan *FlowSetFrame
  • func (rsfs *RedisStreamFlowSet) PushFrame(flwst_frm *FlowSetFrame) error
• type RedisStreamFlowSetFactory
  • func (fty *RedisStreamFlowSetFactory) New(opt *FlowSetOption) (FlowSet, error)
• type RedisStreamFlowSetFactoryOption
• type RedisStreamFlowSetOption
  • func NewRedisStreamFlowSetOption(id string) *RedisStreamFlowSetOption

Variables

var (
	ErrUnknownFlowFactory          = errors.New("unknown flow factory")
	ErrUnknownFlowSetFactory       = errors.New("unknown flow set factory")
	ErrGetAliveRedisClientMaxRetry = errors.New("get alive redis client max retry")
)

Types

type Flow

type Flow interface {
	Id() string
	Device() string
	PushFrame(*pb.Frame) error
	PullFrame() <-chan *pb.Frame
	QueryFrame(...*FlowFilter) ([]*pb.Frame, error)
	Err() error
	Close() error
}

type FlowFactory

type FlowFactory interface {
	New(*FlowOption) (Flow, error)
}

func NewFlowFactory

func NewFlowFactory(name string, args ...interface{}) (FlowFactory, error)

type FlowFilter

type FlowFilter struct {
	BeginAt time.Time
	EndAt   time.Time
}

type FlowOption

type FlowOption struct {
	FlowId   string
	DeviceId string
}

type FlowSet

type FlowSet interface {
	Id() string
	PushFrame(*FlowSetFrame) error
	PullFrame() <-chan *FlowSetFrame
	Err() error
	Close() error
}

type FlowSetFactory

type FlowSetFactory interface {
	New(*FlowSetOption) (FlowSet, error)
}

func NewFlowSetFactory

func NewFlowSetFactory(name string, args ...interface{}) (FlowSetFactory, error)

type FlowSetFrame

type FlowSetFrame struct {
	Device *pb.Device `json:"device"`
	Frame  *pb.Frame  `json:"frame"`
}

type FlowSetOption

type FlowSetOption struct {
	FlowSetId string
}

type RedisStreamFlowSet

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

func (*RedisStreamFlowSet) Close

func (rsfs *RedisStreamFlowSet) Close() (err error)

func (*RedisStreamFlowSet) Err

func (rsfs *RedisStreamFlowSet) Err() error

func (*RedisStreamFlowSet) Id

func (rsfs *RedisStreamFlowSet) Id() string

func (*RedisStreamFlowSet) PullFrame

func (rsfs *RedisStreamFlowSet) PullFrame() <-chan *FlowSetFrame

func (*RedisStreamFlowSet) PushFrame

func (rsfs *RedisStreamFlowSet) PushFrame(flwst_frm *FlowSetFrame) error

type RedisStreamFlowSetFactory

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

func (*RedisStreamFlowSetFactory) New

func (fty *RedisStreamFlowSetFactory) New(opt *FlowSetOption) (FlowSet, error)

type RedisStreamFlowSetFactoryOption

type RedisStreamFlowSetFactoryOption struct {
	RedisStreamAddr        string
	RedisStreamDB          int
	RedisStreamPassword    string
	RedisStreamPoolInitial int
	RedisStreamPoolMax     int
}

type RedisStreamFlowSetOption

type RedisStreamFlowSetOption struct {
	Id string

	ReadStreamGroupBlockTime time.Duration
	StreamExpireTime         time.Duration
	StreamTrimLimit          int64
	StreamTrimProb           float32
}

func NewRedisStreamFlowSetOption

func NewRedisStreamFlowSetOption(id string) *RedisStreamFlowSetOption