Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
var (
DROP = fmt.Sprintf("%U__DROP__", '\\') // U+005C__DROP__
)
Functions ¶
func NewServer ¶
func NewServer(r SessionReducerCreator, inputOptions ...Option) numaflow.Server
NewServer creates a new session reduce server.
Types ¶
type Datum ¶
type Datum interface {
// Value returns the payload of the message.
Value() []byte
// EventTime returns the event time of the message.
EventTime() time.Time
// Watermark returns the watermark of the message.
Watermark() time.Time
// Headers returns the headers of the message.
Headers() map[string]string
}
Datum contains methods to get the payload information.
type Message ¶
type Message struct {
// contains filtered or unexported fields
}
Message is used to wrap the data return by SessionReduce functions
type Option ¶
type Option func(*options)
Option is the interface to apply options.
func WithMaxMessageSize ¶
WithMaxMessageSize sets the server max receive message size and the server max send message size to the given size.
func WithServerInfoFilePath ¶
WithServerInfoFilePath sets the server info file path to the given path.
func WithSockAddr ¶
WithSockAddr start the server with the given sock addr. This is mainly used for testing purposes.
type Service ¶
type Service struct {
sessionreducepb.UnimplementedSessionReduceServer
// contains filtered or unexported fields
}
Service implements the proto gen server interface and contains the sesionreduce operation handler.
func (*Service) IsReady ¶
func (fs *Service) IsReady(context.Context, *emptypb.Empty) (*sessionreducepb.ReadyResponse, error)
IsReady returns true to indicate the gRPC connection is ready.
func (*Service) SessionReduceFn ¶
func (fs *Service) SessionReduceFn(stream sessionreducepb.SessionReduce_SessionReduceFnServer) error
SessionReduceFn applies a session reduce function to a request stream and streams the results.
type SessionReducer ¶
type SessionReducer interface {
// SessionReduce applies a session reduce function to a request stream and streams the results.
SessionReduce(ctx context.Context, keys []string, inputCh <-chan Datum, outputCh chan<- Message)
// Accumulator returns the accumulator for the session reducer, will be invoked when this session is merged
// with another session.
Accumulator(ctx context.Context) []byte
// MergeAccumulator merges the accumulator for the session reducer, will be invoked when another session is merged
// with this session.
MergeAccumulator(ctx context.Context, accumulator []byte)
}
SessionReducer is the interface which can be used to implement a session reduce operation.
type SessionReducerCreator ¶
type SessionReducerCreator interface {
// Create creates a session reducer, will be invoked once for every keyed window.
Create() SessionReducer
}
SessionReducerCreator is the interface which can be used to create a session reducer.
Source Files