flow

type AggregationOp interface {
	// Add puts a slice of elements as the input
	Add([]StreamRecord)
	// Snapshot takes a snapshot of the current state of the AggregationOp
	// Taking a snapshot will restore the dirty flag
	Snapshot() interface{}
	// Dirty flag means if any new item is added after the last snapshot
	Dirty() bool
}

type AggregationOpFactory func() AggregationOp

type Component interface {
	// Setup is the lifecycle hook for resource preparation, e.g. start background job for listening input channel.
	// It must be called before the flow starts to process elements.
	Setup(context.Context) error
	// Teardown is the lifecycle hook for shutting down the Component
	// Implementation should ENSURE that all resource has been correctly recycled before this method returns.
	Teardown(context.Context) error
}

type ComponentState struct {
	sync.WaitGroup
}

type Data []any

type DedupPriorityQueue struct {
	Items []Element
	// contains filtered or unexported fields
}

type Element interface {
	GetIndex() int
	SetIndex(int)
}

type Flow interface {
	io.Closer
	// Filter is used to filter data.
	// The parameter f can be either predicate function for streaming,
	// or conditions for batch query.
	Filter(UnaryOperation[bool]) Flow
	// Map is used to transform data
	Map(UnaryOperation[any]) Flow
	// Window is used to split infinite data into "buckets" of finite size.
	// Currently, it is only applicable to streaming context.
	Window(WindowAssigner) WindowedFlow
	// To pipes data to the given sink
	To(sink Sink) Flow
	// Open opens the flow in the async mode for streaming scenario.
	// The first error is the error combination while opening all components,
	// while the second is a channel for receiving async errors.
	Open() <-chan error
}

type Inlet interface {
	In() chan<- StreamRecord
}

type Operator interface {
	Inlet
	Outlet
	Component
	Exec(downstream Inlet)
}

type Outlet interface {
	Out() <-chan StreamRecord
}

type Sink interface {
	Inlet
	Component
}

type Source interface {
	Outlet
	Component
	Exec(downstream Inlet)
}

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

type UnaryFunc[R any] func(context.Context, interface{}) R

type UnaryOperation[R any] interface {
	Apply(ctx context.Context, data interface{}) R
}

type Window interface {
	// MaxTimestamp returns the upper bound of the Window.
	// Unit: Millisecond
	MaxTimestamp() int64
}

type WindowAssigner interface {
	// AssignWindows assigns a slice of Window according to the given timestamp, e.g. eventTime.
	// The unit of the timestamp here is MilliSecond.
	AssignWindows(timestamp int64) ([]Window, error)
}

type WindowedFlow interface {
	AllowedMaxWindows(windowCnt int) WindowedFlow
	// TopN applies a TopNAggregation to each Window.
	TopN(topNum int, opts ...any) Flow
}