sink

Documentation

Index

• Constants
• type Client
• type Datum
• type Response
  • func ResponseFailure(id, errMsg string) Response
  • func ResponseOK(id string) Response
• type Responses
  • func ResponsesBuilder() Responses
  • func (r Responses) Append(rep Response) Responses
  • func (r Responses) Items() []Response
• type Service
  • func (fs *Service) IsReady(context.Context, *emptypb.Empty) (*sinkpb.ReadyResponse, error)
  • func (fs *Service) SinkFn(ctx context.Context, datumList *sinkpb.DatumList) (*sinkpb.ResponseList, error)
• type SinkFunc
  • func (sf SinkFunc) HandleDo(ctx context.Context, datumList []Datum) Responses
• type SinkHandler

Constants

const (
	Protocol = "unix"
	Addr     = "/var/run/numaflow/udsink.sock"
)

Types

type Client

type Client interface {
	CloseConn(ctx context.Context) error
	IsReady(ctx context.Context, in *emptypb.Empty) (bool, error)
	SinkFn(ctx context.Context, datumList []*sinkpb.Datum) ([]*sinkpb.Response, error)
}

Client contains methods to call a gRPC client.

type Datum

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

type Response

type Response struct {
	// ID corresponds the ID in the message.
	ID string `json:"id"`
	// Successful or not. If it's false, "err" is expected to be present.
	Success bool `json:"success"`
	// Err represents the error message when "success" is false.
	Err string `json:"err,omitempty"`
}

Response is the processing result of each message

func ResponseFailure

func ResponseFailure(id, errMsg string) Response

func ResponseOK

func ResponseOK(id string) Response

type Responses

type Responses []Response

func ResponsesBuilder

func ResponsesBuilder() Responses

ResponsesBuilder returns an empty instance of Responses

func (Responses) Append

func (r Responses) Append(rep Response) Responses

Append appends a response

func (Responses) Items

func (r Responses) Items() []Response

Items returns the response list

type Service

type Service struct {
	sinkpb.UnimplementedUserDefinedSinkServer

	Sinker SinkHandler
}

Service implements the proto gen server interface and contains the sink operation handler.

func (*Service) IsReady

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

IsReady returns true to indicate the gRPC connection is ready.

func (*Service) SinkFn

func (fs *Service) SinkFn(ctx context.Context, datumList *sinkpb.DatumList) (*sinkpb.ResponseList, error)

SinkFn applies a function to a list of datum element.

type SinkFunc

type SinkFunc func(ctx context.Context, datumList []Datum) Responses

SinkFunc is utility type used to convert a HandleDo function to a SinkHandler.

func (SinkFunc) HandleDo

func (sf SinkFunc) HandleDo(ctx context.Context, datumList []Datum) Responses

HandleDo implements the function of sink function.

type SinkHandler

type SinkHandler interface {
	// HandleDo is the function to process a list of incoming messages
	HandleDo(ctx context.Context, datumList []Datum) Responses
}

SinkHandler is the interface of sink function implementation.