fs

Documentation

Index

• Variables
• func NewDefaultInstanceFactory() api.FunctionInstanceFactory
• type CtxKey
• type DefaultInstanceFactory
  • func (f *DefaultInstanceFactory) NewFunctionInstance(definition *model.Function, funcCtx api.FunctionContext, index int32, ...) api.FunctionInstance
• type FuncCtxImpl
  • func NewFuncCtxImpl(store api.StateStore) *FuncCtxImpl
  • func (f *FuncCtxImpl) GetState(key string) ([]byte, error)
  • func (f *FuncCtxImpl) PutState(key string, value []byte) error
• type FunctionInstanceImpl
  • func (instance *FunctionInstanceImpl) Context() context.Context
  • func (instance *FunctionInstanceImpl) Definition() *model.Function
  • func (instance *FunctionInstanceImpl) FunctionContext() api.FunctionContext
  • func (instance *FunctionInstanceImpl) Index() int32
  • func (instance *FunctionInstanceImpl) Logger() *slog.Logger
  • func (instance *FunctionInstanceImpl) Run(runtimeFactory api.FunctionRuntimeFactory, sources []<-chan contube.Record, ...)
  • func (instance *FunctionInstanceImpl) Stop()
  • func (instance *FunctionInstanceImpl) WaitForReady() <-chan error
• type FunctionManager
  • func NewFunctionManager(opts ...ManagerOption) (*FunctionManager, error)
  • func (fm *FunctionManager) Close() error
  • func (fm *FunctionManager) ConsumeEvent(name string) (contube.Record, error)
  • func (fm *FunctionManager) DeleteFunction(namespace, name string) error
  • func (fm *FunctionManager) GetStateStore() api.StateStore
  • func (fm *FunctionManager) ListFunctions() (result []string)
  • func (fm *FunctionManager) ProduceEvent(name string, event contube.Record) error
  • func (fm *FunctionManager) StartFunction(f *model.Function) error
• type ManagerOption
  • func WithDefaultRuntimeFactory(factory api.FunctionRuntimeFactory) ManagerOption
  • func WithDefaultTubeFactory(factory contube.TubeFactory) ManagerOption
  • func WithInstanceFactory(factory api.FunctionInstanceFactory) ManagerOption
  • func WithRuntimeFactory(name string, factory api.FunctionRuntimeFactory) ManagerOption
  • func WithStateStore(store api.StateStore) ManagerOption
  • func WithTubeFactory(name string, factory contube.TubeFactory) ManagerOption

Variables

var ErrStateStoreNotLoaded = errors.New("state store not loaded")

Functions

func NewDefaultInstanceFactory

func NewDefaultInstanceFactory() api.FunctionInstanceFactory

Types

type CtxKey

type CtxKey string

const (
	CtxKeyFunctionName  CtxKey = "function-name"
	CtxKeyInstanceIndex CtxKey = "instance-index"
)

type DefaultInstanceFactory

type DefaultInstanceFactory struct {
	api.FunctionInstanceFactory
}

func (*DefaultInstanceFactory) NewFunctionInstance

func (f *DefaultInstanceFactory) NewFunctionInstance(definition *model.Function, funcCtx api.FunctionContext,
	index int32, logger *slog.Logger) api.FunctionInstance

type FuncCtxImpl

type FuncCtxImpl struct {
	api.FunctionContext
	// contains filtered or unexported fields
}

func NewFuncCtxImpl

func NewFuncCtxImpl(store api.StateStore) *FuncCtxImpl

func (*FuncCtxImpl) GetState

func (f *FuncCtxImpl) GetState(key string) ([]byte, error)

func (*FuncCtxImpl) PutState

func (f *FuncCtxImpl) PutState(key string, value []byte) error

type FunctionInstanceImpl

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

func (*FunctionInstanceImpl) Context

func (instance *FunctionInstanceImpl) Context() context.Context

func (*FunctionInstanceImpl) Definition

func (instance *FunctionInstanceImpl) Definition() *model.Function

func (*FunctionInstanceImpl) FunctionContext

func (instance *FunctionInstanceImpl) FunctionContext() api.FunctionContext

func (*FunctionInstanceImpl) Index

func (instance *FunctionInstanceImpl) Index() int32

func (*FunctionInstanceImpl) Logger

func (instance *FunctionInstanceImpl) Logger() *slog.Logger

func (*FunctionInstanceImpl) Run

func (instance *FunctionInstanceImpl) Run(runtimeFactory api.FunctionRuntimeFactory, sources []<-chan contube.Record,
	sink chan<- contube.Record)

func (*FunctionInstanceImpl) Stop

func (instance *FunctionInstanceImpl) Stop()

func (*FunctionInstanceImpl) WaitForReady

func (instance *FunctionInstanceImpl) WaitForReady() <-chan error

type FunctionManager

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

func NewFunctionManager

func NewFunctionManager(opts ...ManagerOption) (*FunctionManager, error)

func (*FunctionManager) Close

func (fm *FunctionManager) Close() error

func (*FunctionManager) ConsumeEvent

func (fm *FunctionManager) ConsumeEvent(name string) (contube.Record, error)

func (*FunctionManager) DeleteFunction

func (fm *FunctionManager) DeleteFunction(namespace, name string) error

func (*FunctionManager) GetStateStore

func (fm *FunctionManager) GetStateStore() api.StateStore

GetStateStore returns the state store used by the function manager Return nil if no state store is configured

func (*FunctionManager) ListFunctions

func (fm *FunctionManager) ListFunctions() (result []string)

func (*FunctionManager) ProduceEvent

func (fm *FunctionManager) ProduceEvent(name string, event contube.Record) error

func (*FunctionManager) StartFunction

func (fm *FunctionManager) StartFunction(f *model.Function) error

type ManagerOption

type ManagerOption interface {
	// contains filtered or unexported methods
}

func WithDefaultRuntimeFactory

func WithDefaultRuntimeFactory(factory api.FunctionRuntimeFactory) ManagerOption

func WithDefaultTubeFactory

func WithDefaultTubeFactory(factory contube.TubeFactory) ManagerOption

func WithInstanceFactory

func WithInstanceFactory(factory api.FunctionInstanceFactory) ManagerOption

func WithRuntimeFactory

func WithRuntimeFactory(name string, factory api.FunctionRuntimeFactory) ManagerOption

func WithStateStore

func WithStateStore(store api.StateStore) ManagerOption

func WithTubeFactory

func WithTubeFactory(name string, factory contube.TubeFactory) ManagerOption