Documentation
¶
Overview ¶
Package sinker implements the server code for user defined sink.
Examples: https://github.com/numaproj/numaflow-go/tree/main/examples/sinker/
Index ¶
- Constants
- Variables
- func NewServer(h Sinker, inputOptions ...Option) numaflow.Server
- type Datum
- type Message
- type Messages
- type NackOptions
- type Option
- type Response
- func ResponseFailure(id, errMsg string) Response
- func ResponseFallback(id string) Response
- func ResponseNack(id string, opts *NackOptions) Response
- func ResponseOK(id string) Response
- func ResponseOnSuccess(id string, onSuccessMessage Message) Response
- func ResponseServe(id string, result []byte) Response
- type Responses
- type Service
- type Sinker
- type SinkerFunc
- type SystemMetadata
- type UserMetadata
Constants ¶
const ( UDContainerFallbackSink = "fb-udsink" UDContainerOnSuccessSink = "ons-udsink" )
Variables ¶
var NewSystemMetadata = metadata.NewSystemMetadata
NewSystemMetadata creates a new SystemMetadata. This is for internal and testing purposes only.
var NewUserMetadata = metadata.NewUserMetadata
NewUserMetadata creates a new UserMetadata.
Functions ¶
Types ¶
type Datum ¶
type Datum interface {
// Keys returns the keys of the message.
Keys() []string
// 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
// ID returns the ID of the message.
ID() string
// Headers returns the headers of the message.
Headers() map[string]string
// UserMetadata returns the user metadata of the message.
UserMetadata() *UserMetadata
// SystemMetadata returns the system metadata of the message.
SystemMetadata() *SystemMetadata
}
Datum is the interface of incoming message payload for sink function.
type Message ¶ added in v0.12.0
type Message struct {
// contains filtered or unexported fields
}
Message is used to wrap the data returned by Sink
func NewMessage ¶ added in v0.12.0
NewMessage creates a Message with value
func (Message) UserMetadata ¶ added in v0.12.0
func (m Message) UserMetadata() *UserMetadata
UserMetadata returns message user metadata
func (Message) WithUserMetadata ¶ added in v0.12.0
func (m Message) WithUserMetadata(userMetadata *UserMetadata) Message
WithUserMetadata is used to assign the user metadata to the message
type Messages ¶ added in v0.12.0
type Messages []Message
func MessagesBuilder ¶ added in v0.12.0
func MessagesBuilder() Messages
MessagesBuilder returns an empty instance of Messages
type NackOptions ¶ added in v0.12.0
type NackOptions = nackoptions.NackOptions
NackOptions carries per-message redelivery options for a nacked message. See ResponseNack.
type Option ¶
type Option func(*options)
Option is the interface to apply options.
func WithMaxMessageSize ¶
WithMaxMessageSize sets the sinkServer max receive message size and the sinkServer max send message size to the given size.
func WithServerInfoFilePath ¶
WithServerInfoFilePath sets the sinkServer info file path.
func WithSockAddr ¶
WithSockAddr start the sinkServer with the given sock addr. This is mainly used for testing purpose.
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"`
// Fallback is true if the message to be sent to the fallback sink.
Fallback bool `json:"fallback,omitempty"`
// OnSuccess is true if the message to be sent to the onSuccess sink.
OnSuccess bool `json:"on_success,omitempty"`
// Serve is true if the message to be sent to serving store.
Serve bool `json:"serve,omitempty"`
// ServeResponse is the response that will be sent to the serving store.
ServeResponse []byte `json:"serve_reponse,omitempty"`
// OnSuccessMessage is the message to be sent to the onSuccess sink.
OnSuccessMessage Message `json:"on_success_message,omitempty"`
// Nack is true if the message should be negatively acknowledged (redelivered).
Nack bool `json:"nack,omitempty"`
// NackOptions carries optional redelivery options when Nack is true.
NackOptions *NackOptions `json:"nack_options,omitempty"`
}
Response is the processing result of each message
func ResponseFailure ¶
ResponseFailure creates a failed Response with the given id and error message. The Success field is set to false and the Err field is set to the provided error message.
func ResponseFallback ¶ added in v0.7.0
ResponseFallback creates a Response with the Fallback field set to true. This indicates that the message should be sent to the fallback sink.
func ResponseNack ¶ added in v0.12.0
func ResponseNack(id string, opts *NackOptions) Response
ResponseNack creates a Response with the Nack field set to true. This indicates the message should be negatively acknowledged and redelivered. opts may be nil; when set it carries redelivery options.
func ResponseOK ¶
ResponseOK creates a successful Response with the given id. The Success field is set to true.
func ResponseOnSuccess ¶ added in v0.12.0
ResponseOnSuccess creates a Response with the OnSuccess field set to true. This indicates that the message should be sent to the onSuccess sink.
func ResponseServe ¶ added in v0.10.0
ResponseServe creates a Response with the Serve field set to true. This indicates that the message should be sent to the serving store.
type Responses ¶
type Responses []Response
func ResponsesBuilder ¶
func ResponsesBuilder() Responses
ResponsesBuilder returns an empty instance of Responses
type Service ¶
type Service struct {
sinkpb.UnimplementedSinkServer
Sinker Sinker
// contains filtered or unexported fields
}
Service implements the proto gen server interface and contains the sinkfn operation handler.
type Sinker ¶
type Sinker interface {
// Sink is the function to process a list of incoming messages
Sink(ctx context.Context, datumStreamCh <-chan Datum) Responses
}
Sinker is the interface of sink function implementation.
type SinkerFunc ¶
SinkerFunc is utility type used to convert a Sink function to a Sinker.
type SystemMetadata ¶ added in v0.11.0
type SystemMetadata = metadata.SystemMetadata
SystemMetadata wraps system-generated metadata groups per message. It is read-only to UDFs.
type UserMetadata ¶ added in v0.11.0
type UserMetadata = metadata.UserMetadata
UserMetadata wraps user-defined metadata groups per message.
Source Files