udf

type Diagnostic ¶

type Diagnostic interface {
	Error(msg string, err error, ctx ...keyvalue.T)

	UDFLog(msg string)
}

type Info ¶

type Info struct {
	Wants    agent.EdgeType
	Provides agent.EdgeType
	Options  map[string]*agent.OptionInfo
}

type Interface ¶

type Interface interface {
	Open() error
	Info() (Info, error)
	Init(options []*agent.Option) error
	Abort(err error)
	Close() error

	Snapshot() ([]byte, error)
	Restore(snapshot []byte) error

	In() chan<- edge.Message
	Out() <-chan edge.Message
}

Interface for communicating with a UDF

type Server ¶

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

Server provides an implementation for the core communication with UDFs. The Server provides only a partial implementation of udf.Interface as it is expected that setup and teardown will be necessary to create a Server. As such the Open and Close methods are not implemented.

Once a Server is created and started the owner can send points or batches to the UDF by writing them to the PointIn or BatchIn channels respectively, and according to the type of UDF created.

The Server may be Aborted at anytime for various reasons. It is the owner's responsibility via the abortCallback to stop writing to the *In channels since no more selects on the channels will be performed.

Calling Stop on the Server should only be done once the owner has stopped writing to the *In channel, at which point the remaining data will be processed and the UDF will be allowed to clean up.

Callling Info returns information about available options the UDF has.

Calling Init is required to process data. The behavior is undefined if you send points/batches to the Server without calling Init.

func NewServer(
	taskID, nodeID string,
	in agent.ByteReadReader,
	out io.WriteCloser,
	d Diagnostic,
	timeout time.Duration,
	abortCallback func(),
	killCallback func(),
) *Server

func (s *Server) Abort(err error)

func (s *Server) In() chan<- edge.Message

func (s *Server) Info() (Info, error)

func (s *Server) Init(options []*agent.Option) error

func (s *Server) Out() <-chan edge.Message

func (s *Server) Restore(snapshot []byte) error

func (s *Server) Snapshot() ([]byte, error)

func (s *Server) Start() error

func (s *Server) Stop() error

func (s *Server) WaitIO()