modules

type CHook func(c *model.KuiperConf)

type Connection interface {
	// Provision validates/configures and constructs the candidate. It
	// must return without blocking for remote readiness, and it may
	// start provider-internal background activity.
	//
	// Ownership: a nil return transfers full ownership to the Pool. The
	// Pool may Close the candidate at any time afterwards — including
	// when a later creation step (e.g. persist) fails before any Dial,
	// or when the lifecycle stops before Dial completes. Close must
	// safely release all candidate resources and background activity,
	// even if Dial was never called or never succeeded.
	// A non-nil return means Provision itself failed: the provider
	// owns cleanup of any partial state, the Pool closes nothing.
	Provision(ctx api.StreamContext, conId string, props map[string]any) error
	// Dial establishes or waits for initial usability. It must honor
	// ctx cancellation promptly or otherwise have a finite bound: the
	// Pool stops a lifecycle by canceling its scope and waiting for the
	// worker, so an uncancellable Dial hangs every stop path
	// (Detach/Drop/reset). Returning nil means the candidate is
	// initially usable.
	Dial(ctx api.StreamContext) error
	GetId(ctx api.StreamContext) string
	// Ping performs one on-demand bounded validation attempt.
	//
	// It may perform bounded remote I/O, including using an ephemeral
	// connection when the provider has no persistent transport. It must
	// not retry, reconnect/recover, or create/replace the provider's
	// long-lived runtime transport.
	//
	// It is an explicit-caller API (Test Connection, initial
	// validation) on a caller-bounded scope: providers must honor the
	// caller deadline rather than imposing their own unbounded block.
	// Self-recovering clients (StatefulDialer) may answer from their
	// local lifecycle flag instead of hitting the remote.
	//
	// Periodic monitoring is a separate opt-in capability provided by
	// PeriodicHealthChecker below; implementing Ping alone never
	// enrolls a provider in periodic checks.
	Ping(ctx api.StreamContext) error
	api.Closable
}

type ConnectionProvider func(ctx api.StreamContext) Connection

type ConnectionStatus struct {
	Status string `json:"status"`
	ErrMsg string `json:"errMsg,omitempty"`
}

type ConvertWriterProvider func(ctx api.StreamContext, schemaId string, logicalSchema map[string]*ast.JsonStreamField, props map[string]any) (message.ConvertWriter, error)

type ConverterProvider func(ctx api.StreamContext, schemaId string, logicalSchema map[string]*ast.JsonStreamField, props map[string]any) (message.Converter, error)

type FileStreamDecorator interface {
	// Provision Set up the static properties
	Provision(ctx api.StreamContext, props map[string]any) error
	// ReadMeta Read the metadata from the file source, and save in the decorator itself
	// It will receive lines, when receiving EOF, there will be no more lines.
	ReadMeta(ctx api.StreamContext, line []byte)
	// Decorate the file source
	Decorate(ctx api.StreamContext, data any) any
}

type FileStreamDecoratorProvider func(ctx api.StreamContext) FileStreamDecorator

type FileStreamReader interface {
	// Provision Set up the static properties
	Provision(ctx api.StreamContext, props map[string]any) error
	// Bind set the file stream. Make sure the previous read has done
	Bind(ctx api.StreamContext, fileStream io.Reader, maxSize int) error
	// Read the next record. Returns EOF when the input has reached its end.
	Read(ctx api.StreamContext) (any, error)
	// IsBytesReader If is bytes reader, Read must return []byte, otherwise return map or []map
	IsBytesReader() bool
	api.Closable
}

type FileStreamReaderProvider func(ctx api.StreamContext) FileStreamReader

type Files struct {
	SchemaFile string
	SoFile     string
	Version    string
}

type Merger interface {
	Split(ctx api.StreamContext, b []byte) [][]byte
	// Merging is called when read in a new frame
	Merging(ctx api.StreamContext, b []byte) error
	// Trigger is called when rate limiter is trigger to send out a message
	Trigger(ctx api.StreamContext) ([]any, bool)
}

type MergerProvider func(ctx api.StreamContext, payloadSchema string, logicalSchema map[string]*ast.JsonStreamField, props map[string]any) (Merger, error)

type NewFuncFunc func() api.Function

type NewLookupSourceFunc func() api.Source

type NewSinkFunc func() api.Sink

type NewSourceFunc func() api.Source

type PeriodicHealthChecker interface {
	HealthCheck(ctx api.StreamContext) error
}

type PoolRecoverableConnection interface {
	Connection
	Recover(ctx api.StreamContext) error
}

type RollHook interface {
	Provision(ctx api.StreamContext, props map[string]any) error
	RollDone(ctx api.StreamContext, filePath string) error
	api.Closable
}

type RollHookProvider func() RollHook

type SchemaTypeDef interface {
	Scan(logger api.Logger, path string) (map[string]*Files, error)
	Infer(logger api.Logger, filePath string, messageId string) (ast.StreamFields, error)
}

type SchemaTypeInfo struct {
	Def SchemaTypeDef
	Ext string
}

type StatefulDialer interface {
	SetStatusChangeHandler(ctx api.StreamContext, handler api.StatusChangeHandler)
	Status(ctx api.StreamContext) ConnectionStatus
}