mapstreamer

Documentation

Index

• Variables
• func DefaultOptions() *options
• func NewServer(ms MapStreamer, inputOptions ...Option) numaflow.Server
• type Datum
  • func NewHandlerDatum(value []byte, eventTime time.Time, watermark time.Time) Datum
• type MapStreamer
• type MapStreamerFunc
  • func (msf MapStreamerFunc) MapStream(ctx context.Context, keys []string, datum Datum, messageCh chan<- Message)
• type Message
  • func MessageToDrop() Message
  • func NewMessage(value []byte) Message
  • func (m Message) Keys() []string
  • func (m Message) Tags() []string
  • func (m Message) Value() []byte
  • func (m Message) WithKeys(keys []string) Message
  • func (m Message) WithTags(tags []string) Message
• type Messages
  • func MessagesBuilder() Messages
  • func (m Messages) Append(msg Message) Messages
  • func (m Messages) Items() []Message
• type Option
  • func WithMaxMessageSize(size int) Option
  • func WithServerInfoFilePath(f string) Option
  • func WithSockAddr(addr string) Option
• type Service
  • func (fs *Service) IsReady(context.Context, *emptypb.Empty) (*mapstreampb.ReadyResponse, error)
  • func (fs *Service) MapStreamFn(d *mapstreampb.MapStreamRequest, ...) error

Variables

var (
	DROP = fmt.Sprintf("%U__DROP__", '\\') // U+005C__DROP__
)

Functions

func DefaultOptions

func DefaultOptions() *options

func NewServer

func NewServer(ms MapStreamer, inputOptions ...Option) numaflow.Server

NewServer creates a new map streaming server.

Types

type Datum

type Datum interface {
	Value() []byte
	EventTime() time.Time
	Watermark() time.Time
}

Datum contains methods to get the payload information.

func NewHandlerDatum

func NewHandlerDatum(value []byte, eventTime time.Time, watermark time.Time) Datum

type MapStreamer

type MapStreamer interface {
	// MapStream is the function to process each coming message and streams
	// the result back using a channel.
	MapStream(ctx context.Context, keys []string, datum Datum, messageCh chan<- Message)
}

MapStreamer is the interface of map stream function implementation.

type MapStreamerFunc

type MapStreamerFunc func(ctx context.Context, keys []string, datum Datum, messageCh chan<- Message)

MapStreamerFunc is a utility type used to convert a function to a MapStreamer.

func (MapStreamerFunc) MapStream

func (msf MapStreamerFunc) MapStream(ctx context.Context, keys []string, datum Datum, messageCh chan<- Message)

MapStream implements the function of map stream function.

type Message

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

Message is used to wrap the data return by MapStream functions

func MessageToDrop

func MessageToDrop() Message

MessageToDrop creates a Message to be dropped

func NewMessage

func NewMessage(value []byte) Message

NewMessage creates a Message with value

func (Message) Keys

func (m Message) Keys() []string

Keys returns message keys

func (Message) Tags

func (m Message) Tags() []string

Tags returns message tags

func (Message) Value

func (m Message) Value() []byte

Value returns message value

func (Message) WithKeys

func (m Message) WithKeys(keys []string) Message

WithKeys is used to assign the keys to the message

func (Message) WithTags

func (m Message) WithTags(tags []string) Message

WithTags is used to assign the tags to the message tags will be used for conditional forwarding

type Messages

type Messages []Message

func MessagesBuilder

func MessagesBuilder() Messages

MessagesBuilder returns an empty instance of Messages

func (Messages) Append

func (m Messages) Append(msg Message) Messages

Append appends a Message

func (Messages) Items

func (m Messages) Items() []Message

Items returns the message list

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 purposes.

type Service

type Service struct {
	mapstreampb.UnimplementedMapStreamServer

	MapperStream MapStreamer
}

Service implements the proto gen server interface and contains the map streaming function.

func (*Service) IsReady

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

IsReady returns true to indicate the gRPC connection is ready.

func (*Service) MapStreamFn

func (fs *Service) MapStreamFn(d *mapstreampb.MapStreamRequest, stream mapstreampb.MapStream_MapStreamFnServer) error

MapStreamFn applies a function to each request element and streams the results back.