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() *common.Logger
  • func (instance *FunctionInstanceImpl) Run(runtime api.FunctionRuntime, sources []<-chan contube.Record, ...)
  • func (instance *FunctionInstanceImpl) Stop()
• type FunctionManager
  • func NewFunctionManager(opts ...ManagerOption) (FunctionManager, error)
• type ManagerOption
  • func WithInstanceFactory(factory api.FunctionInstanceFactory) ManagerOption
  • func WithLogger(log *logr.Logger) ManagerOption
  • func WithQueueFactory(factory contube.TubeFactory) ManagerOption
  • func WithRuntimeFactory(name string, factory api.FunctionRuntimeFactory) ManagerOption
  • func WithStateStore(store api.StateStore) ManagerOption
  • func WithTubeFactory(name string, factory contube.TubeFactory) ManagerOption
• type NamespacedName
  • func GetNamespacedName(namespace, name string) NamespacedName
  • func (n NamespacedName) String() string

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 *common.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() *common.Logger

func (*FunctionInstanceImpl) Run

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

func (*FunctionInstanceImpl) Stop

func (instance *FunctionInstanceImpl) Stop()

type FunctionManager

type FunctionManager interface {
	StartFunction(f *model.Function) error
	DeleteFunction(namespace, name string) error
	ListFunctions() []string
	ProduceEvent(name string, event contube.Record) error
	ConsumeEvent(name string) (contube.Record, error)
	GetStateStore() api.StateStore
	Close() error
}

func NewFunctionManager

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

type ManagerOption

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

func WithInstanceFactory

func WithInstanceFactory(factory api.FunctionInstanceFactory) ManagerOption

func WithLogger

func WithLogger(log *logr.Logger) ManagerOption

func WithQueueFactory

func WithQueueFactory(factory contube.TubeFactory) 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

type NamespacedName

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

func GetNamespacedName

func GetNamespacedName(namespace, name string) NamespacedName

func (NamespacedName) String

func (n NamespacedName) String() string