sideinput

Documentation

Index

• Constants
• func NewSideInputServer(r SideInputRetriever, inputOptions ...Option) numaflow.Server
• type Message
  • func BroadcastMessage(value []byte) Message
  • func NoBroadcastMessage() Message
• type Option
  • func WithMaxMessageSize(size int) Option
  • func WithServerInfoFilePath(f string) Option
  • func WithSockAddr(addr string) Option
• type RetrieveFunc
  • func (mf RetrieveFunc) RetrieveSideInput(ctx context.Context) Message
• type Service
  • func (fs *Service) IsReady(context.Context, *emptypb.Empty) (*sideinputpb.ReadyResponse, error)
  • func (fs *Service) RetrieveSideInput(ctx context.Context, _ *emptypb.Empty) (*sideinputpb.SideInputResponse, error)
• type SideInputRetriever

Constants

const (
	DirPath = "/var/numaflow/side-inputs"
)

Functions

func NewSideInputServer

func NewSideInputServer(r SideInputRetriever, inputOptions ...Option) numaflow.Server

NewSideInputServer creates a new server object.

Types

type Message

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

Message is used to wrap the data return by UserSideInput function. It contains the data value for the given side input parameter requested.

func BroadcastMessage

func BroadcastMessage(value []byte) Message

BroadcastMessage creates a new Message with the given value This is used to broadcast the message to other side input vertices.

func NoBroadcastMessage

func NoBroadcastMessage() Message

NoBroadcastMessage creates a new Message with noBroadcast flag set to true This is used to drop the message and not to broadcast it to other side input vertices.

type Option

type Option func(*options)

Option is the interface to apply options.

func WithMaxMessageSize

func WithMaxMessageSize(size int) Option

WithMaxMessageSize sets the server max receive message size and the server max send message size to the given size.

func WithServerInfoFilePath

func WithServerInfoFilePath(f string) Option

WithServerInfoFilePath sets the server info file path to the given path.

func WithSockAddr

func WithSockAddr(addr string) Option

WithSockAddr start the server with the given sock addr. This is mainly used for testing purpose.

type RetrieveFunc

type RetrieveFunc func(ctx context.Context) Message

RetrieveFunc is a utility type used to convert a RetrieveSideInput function to a SideInputRetriever.

func (RetrieveFunc) RetrieveSideInput

func (mf RetrieveFunc) RetrieveSideInput(ctx context.Context) Message

RetrieveSideInput implements the function of RetrieveSideInput function.

type Service

type Service struct {
	sideinputpb.UnimplementedSideInputServer
	Retriever SideInputRetriever
	// contains filtered or unexported fields
}

Service implements the proto gen server interface and contains the retrieve operation handler

func (*Service) IsReady

func (fs *Service) IsReady(context.Context, *emptypb.Empty) (*sideinputpb.ReadyResponse, error)

IsReady returns true to indicate the gRPC connection is ready.

func (*Service) RetrieveSideInput

func (fs *Service) RetrieveSideInput(ctx context.Context, _ *emptypb.Empty) (*sideinputpb.SideInputResponse, error)

RetrieveSideInput applies the function for each side input retrieval request.

type SideInputRetriever

type SideInputRetriever interface {
	// RetrieveSideInput is the function to process each side-input request.
	RetrieveSideInput(ctx context.Context) Message
}

SideInputRetriever is the interface for side input retrieval implementation.