manager

Mar 1, 2024

Changes in this version

+ type InputWrapper struct

+ func WrapInput(i input.Streamed) *InputWrapper

+ func (w *InputWrapper) CloseExistingInput(ctx context.Context, forSwap bool) error

+ func (w *InputWrapper) Connected() bool

+ func (w *InputWrapper) SwapInput(i input.Streamed)

+ func (w *InputWrapper) TransactionChan() <-chan message.Transaction

+ func (w *InputWrapper) TriggerCloseNow()

+ func (w *InputWrapper) TriggerStopConsuming()

+ func (w *InputWrapper) WaitForClose(ctx context.Context) error

type ResourceConfig

+ func FromAny(prov docs.Provider, v any) (conf ResourceConfig, err error)

+ func FromParsed(prov docs.Provider, pConf *docs.ParsedConfig) (conf ResourceConfig, err error)

type Type

+ func (t *Type) NewScanner(conf scanner.Config) (scanner.Creator, error)

Apr 25, 2023

Changes in this version

type Type

+ func (t *Type) CloseObservability(ctx context.Context) error

Feb 20, 2023

Changes in this version

type Type

+ func (t *Type) RemoveCache(ctx context.Context, name string) error

+ func (t *Type) RemoveInput(ctx context.Context, name string) error

+ func (t *Type) RemoveOutput(ctx context.Context, name string) error

+ func (t *Type) RemoveProcessor(ctx context.Context, name string) error

+ func (t *Type) RemoveRateLimit(ctx context.Context, name string) error

Dec 21, 2022

Changes in this version

type OptFunc

+ func OptSetFS(fs ifs.FS) OptFunc

Oct 26, 2022

Changes in this version

type Type

+ func (t *Type) FS() ifs.FS

Aug 31, 2022

Changes in this version

type Type

+ func (t *Type) TriggerCloseNow()

+ func (t *Type) TriggerStopConsuming()

Jul 18, 2022

Changes in this version

type OptFunc

+ func OptSetStreamHTTPNamespacing(enabled bool) OptFunc

Jun 23, 2022

Changes in this version

type OptFunc

+ func OptSetAPIReg(r APIReg) OptFunc

+ func OptSetLogger(logger log.Modular) OptFunc

+ func OptSetMetrics(stats *metrics.Namespaced) OptFunc

+ func OptSetTracer(tracer trace.TracerProvider) OptFunc

type Type

+ func (t *Type) Tracer() trace.TracerProvider

May 11, 2022

Changes in this version

type OptFunc

+ func OptSetStreamsMode(b bool) OptFunc

type Type

+ func New(conf ResourceConfig, apiReg APIReg, log log.Modular, stats *metrics.Namespaced, ...) (*Type, error)

Apr 20, 2022

Changes in this version

+ func Spec() docs.FieldSpecs

+ type APIReg interface

+ RegisterEndpoint func(path, desc string, h http.HandlerFunc)

+ type ErrResourceNotFound string

+ func (e ErrResourceNotFound) Error() string

+ type OptFunc func(*Type)

+ func OptSetBloblangEnvironment(env *bloblang.Environment) OptFunc

+ func OptSetEnvironment(e *bundle.Environment) OptFunc

+ type ResourceConfig struct

+ ResourceCaches []cache.Config

+ ResourceInputs []input.Config

+ ResourceOutputs []output.Config

+ ResourceProcessors []processor.Config

+ ResourceRateLimits []ratelimit.Config

+ func NewResourceConfig() ResourceConfig

+ func (r *ResourceConfig) AddFrom(extra *ResourceConfig) error

+ type Type struct

+ func NewV2(conf ResourceConfig, apiReg APIReg, log log.Modular, stats *metrics.Namespaced, ...) (*Type, error)

+ func (t *Type) AccessCache(ctx context.Context, name string, fn func(cache.V1)) error

+ func (t *Type) AccessInput(ctx context.Context, name string, fn func(iinput.Streamed)) error

+ func (t *Type) AccessOutput(ctx context.Context, name string, fn func(ioutput.Sync)) error

+ func (t *Type) AccessProcessor(ctx context.Context, name string, fn func(iprocessor.V1)) error

+ func (t *Type) AccessRateLimit(ctx context.Context, name string, fn func(ratelimit.V1)) error

+ func (t *Type) BloblEnvironment() *bloblang.Environment

+ func (t *Type) CloseAsync()

+ func (t *Type) Environment() *bundle.Environment

+ func (t *Type) ForStream(id string) interop.Manager

+ func (t *Type) GetDocs(name string, ctype docs.Type) (docs.ComponentSpec, bool)

+ func (t *Type) GetPipe(name string) (<-chan message.Transaction, error)

+ func (t *Type) IntoPath(segments ...string) interop.Manager

+ func (t *Type) Label() string

+ func (t *Type) Logger() log.Modular

+ func (t *Type) Metrics() metrics.Type

+ func (t *Type) NewBuffer(conf buffer.Config) (buffer.Streamed, error)

+ func (t *Type) NewCache(conf cache.Config) (cache.V1, error)

+ func (t *Type) NewInput(conf input.Config, pipelines ...iprocessor.PipelineConstructorFunc) (iinput.Streamed, error)

+ func (t *Type) NewOutput(conf output.Config, pipelines ...iprocessor.PipelineConstructorFunc) (ioutput.Streamed, error)

+ func (t *Type) NewProcessor(conf processor.Config) (iprocessor.V1, error)

+ func (t *Type) NewRateLimit(conf ratelimit.Config) (ratelimit.V1, error)

+ func (t *Type) Path() []string

+ func (t *Type) ProbeCache(name string) bool

+ func (t *Type) ProbeInput(name string) bool

+ func (t *Type) ProbeOutput(name string) bool

+ func (t *Type) ProbeProcessor(name string) bool

+ func (t *Type) ProbeRateLimit(name string) bool

+ func (t *Type) RegisterEndpoint(apiPath, desc string, h http.HandlerFunc)

+ func (t *Type) SetPipe(name string, tran <-chan message.Transaction)

+ func (t *Type) StoreCache(ctx context.Context, name string, conf cache.Config) error

+ func (t *Type) StoreInput(ctx context.Context, name string, conf input.Config) error

+ func (t *Type) StoreOutput(ctx context.Context, name string, conf output.Config) error

+ func (t *Type) StoreProcessor(ctx context.Context, name string, conf processor.Config) error

+ func (t *Type) StoreRateLimit(ctx context.Context, name string, conf ratelimit.Config) error

+ func (t *Type) UnsetPipe(name string, tran <-chan message.Transaction)

+ func (t *Type) WaitForClose(timeout time.Duration) error

+ func (t *Type) WithAddedMetrics(m metrics.Type) interop.Manager

+ func (t *Type) WithMetricsMapping(m *metrics.Mapping) *Type