service

package

Go to main page

Versions in this module

v1

v1.4.0

Dec 8, 2024

Changes in this version

type BatchPolicy

+ Jitter float64

type StreamBuilder

+ func (s *StreamBuilder) BuildStrict() (*Stream, error)

v1.4.0-rc4

Dec 8, 2024

v1.4.0-rc3

Dec 8, 2024

v1.4.0-rc2

Dec 6, 2024

v1.4.0-rc1

Dec 6, 2024

v1.3.0

Oct 31, 2024

v1.3.0-rc1

Nov 4, 2024

v1.2.0

Aug 21, 2024

Changes in this version

type Resources

+ func (r *Resources) GetGeneric(key any) (any, bool)

+ func (r *Resources) GetOrSetGeneric(key, value any) (actual any, loaded bool)

+ func (r *Resources) SetGeneric(key, value any)

v1.1.0

Jul 12, 2024

Changes in this version

type CLIOptFunc

+ func CLIOptSetBinaryName(n string) CLIOptFunc

+ func CLIOptSetDefaultConfigPaths(paths ...string) CLIOptFunc

+ func CLIOptSetDocumentationURL(n string) CLIOptFunc

+ func CLIOptSetProductName(n string) CLIOptFunc

+ func CLIOptSetShowRunCommand(show bool) CLIOptFunc

type ConfigSchema

+ func (s *ConfigSchema) SetFieldDefault(value any, path ...string)

type Message

+ func (m *Message) AddSyncResponse() error

+ func (m *Message) WithSyncResponseStore() (*Message, *SyncResponseStore)

type MessageBatch

+ func (b MessageBatch) BloblangExecutor(blobl *bloblang.Executor) *MessageBatchBloblangExecutor

+ func (b MessageBatch) InterpolationExecutor(i *InterpolatedString) *MessageBatchInterpolationExecutor

+ type MessageBatchBloblangExecutor struct

+ func (b MessageBatchBloblangExecutor) Mutate(index int) (*Message, error)

+ func (b MessageBatchBloblangExecutor) Query(index int) (*Message, error)

+ func (b MessageBatchBloblangExecutor) QueryValue(index int) (any, error)

+ type MessageBatchInterpolationExecutor struct

+ func (b MessageBatchInterpolationExecutor) TryBytes(index int) ([]byte, error)

+ func (b MessageBatchInterpolationExecutor) TryString(index int) (string, error)

type StreamBuilder

+ func (s *StreamBuilder) SetSchema(schema *ConfigSchema)

+ type SyncResponseStore struct

+ func (s *SyncResponseStore) Read() []MessageBatch

v1.0.2

Jun 7, 2024

Changes in this version

+ const AutoRetryNacksToggleFieldName

+ var ErrBlockingWrite = errors.New("a blocking write attempt was aborted")

+ var ErrEndOfBuffer = errors.New("end of buffer")

+ var ErrEndOfInput = errors.New("end of input")

+ var ErrKeyAlreadyExists = errors.New("key already exists")

+ var ErrKeyNotFound = errors.New("key does not exist")

+ var ErrNotConnected = errors.New("not connected")

+ func Globs(f fs.FS, paths ...string) ([]string, error)

+ func GlobsAndSuperPaths(f fs.FS, paths []string, extensions ...string) ([]string, error)

+ func OSFS() fs.FS

+ func OutputPerformanceDocs(benefitsFromMaxInFlight, benefitsFromBatching bool) (content string)

+ func ReadFile(f fs.FS, name string) ([]byte, error)

+ func RegisterBatchBuffer(name string, spec *ConfigSpec, ctor BatchBufferConstructor) error

+ func RegisterBatchInput(name string, spec *ConfigSpec, ctor BatchInputConstructor) error

+ func RegisterBatchOutput(name string, spec *ConfigSpec, ctor BatchOutputConstructor) error

+ func RegisterBatchProcessor(name string, spec *ConfigSpec, ctor BatchProcessorConstructor) error

+ func RegisterBatchScannerCreator(name string, spec *ConfigSpec, ctor BatchScannerCreatorConstructor) error

+ func RegisterCache(name string, spec *ConfigSpec, ctor CacheConstructor) error

+ func RegisterInput(name string, spec *ConfigSpec, ctor InputConstructor) error

+ func RegisterMetricsExporter(name string, spec *ConfigSpec, ctor MetricsExporterConstructor) error

+ func RegisterOtelTracerProvider(name string, spec *ConfigSpec, ctor OtelTracerProviderConstructor) error

+ func RegisterOutput(name string, spec *ConfigSpec, ctor OutputConstructor) error

+ func RegisterProcessor(name string, spec *ConfigSpec, ctor ProcessorConstructor) error

+ func RegisterRateLimit(name string, spec *ConfigSpec, ctor RateLimitConstructor) error

+ func RegisterTemplateYAML(yamlStr string) error

+ func RunCLI(ctx context.Context, optFuncs ...CLIOptFunc)

+ func XFormatConfigJSON() ([]byte, error)

+ type AckFunc func(ctx context.Context, err error) error

+ type BatchBuffer interface

+ EndOfInput func()

+ ReadBatch func(context.Context) (MessageBatch, AckFunc, error)

+ WriteBatch func(context.Context, MessageBatch, AckFunc) error

+ type BatchBufferConstructor func(conf *ParsedConfig, mgr *Resources) (BatchBuffer, error)

+ type BatchError struct

+ func NewBatchError(b MessageBatch, headline error) *BatchError

+ func (err *BatchError) Error() string

+ func (err *BatchError) Failed(i int, merr error) *BatchError

+ func (err *BatchError) IndexedErrors() int

+ func (err *BatchError) Unwrap() error

+ func (err *BatchError) WalkMessages(fn func(int, *Message, error) bool)

+ func (err *BatchError) WalkMessagesIndexedBy(s *Indexer, fn func(int, *Message, error) bool)

+ type BatchInput interface

+ Connect func(context.Context) error

+ ReadBatch func(context.Context) (MessageBatch, AckFunc, error)

+ func AutoRetryNacksBatched(i BatchInput) BatchInput

+ func AutoRetryNacksBatchedToggled(c *ParsedConfig, i BatchInput) (BatchInput, error)

+ func InputBatchedWithMaxInFlight(n int, i BatchInput) BatchInput

+ type BatchInputConstructor func(conf *ParsedConfig, mgr *Resources) (BatchInput, error)

+ type BatchOutput interface

+ Connect func(context.Context) error

+ WriteBatch func(context.Context, MessageBatch) error

+ type BatchOutputConstructor func(conf *ParsedConfig, mgr *Resources) (out BatchOutput, batchPolicy BatchPolicy, maxInFlight int, err error)

+ type BatchPolicy struct

+ ByteSize int

+ Check string

+ Count int

+ Period string

+ func (b BatchPolicy) IsNoop() bool

+ func (b BatchPolicy) NewBatcher(res *Resources) (*Batcher, error)

+ type BatchProcessor interface

+ ProcessBatch func(context.Context, MessageBatch) ([]MessageBatch, error)

+ type BatchProcessorConstructor func(conf *ParsedConfig, mgr *Resources) (BatchProcessor, error)

+ type BatchScanner interface

+ Close func(context.Context) error

+ NextBatch func(context.Context) (MessageBatch, AckFunc, error)

+ func AutoAggregateBatchScannerAcks(strm SimpleBatchScanner, aFn AckFunc) BatchScanner

+ type BatchScannerCreator interface

+ Close func(context.Context) error

+ Create func(io.ReadCloser, AckFunc, *ScannerSourceDetails) (BatchScanner, error)

+ type BatchScannerCreatorConstructor func(conf *ParsedConfig, mgr *Resources) (BatchScannerCreator, error)

+ type Batcher struct

+ func (b *Batcher) Add(msg *Message) bool

+ func (b *Batcher) Close(ctx context.Context) error

+ func (b *Batcher) Flush(ctx context.Context) (batch MessageBatch, err error)

+ func (b *Batcher) UntilNext() (time.Duration, bool)

+ func (b *Batcher) XUnwrapper() any

+ type CLIOptBuilder struct

+ type CLIOptFunc func(*CLIOptBuilder)

+ func CLIOptAddTeeLogger(l *slog.Logger) CLIOptFunc

+ func CLIOptOnConfigParse(fn func(fn *ParsedConfig) error) CLIOptFunc

+ func CLIOptOnLoggerInit(fn func(*Logger)) CLIOptFunc

+ func CLIOptSetMainSchemaFrom(fn func() *ConfigSchema) CLIOptFunc

+ func CLIOptSetVersion(version, dateBuilt string) CLIOptFunc

+ type Cache interface

+ Add func(ctx context.Context, key string, value []byte, ttl *time.Duration) error

+ Delete func(ctx context.Context, key string) error

+ Get func(ctx context.Context, key string) ([]byte, error)

+ Set func(ctx context.Context, key string, value []byte, ttl *time.Duration) error

+ type CacheConstructor func(conf *ParsedConfig, mgr *Resources) (Cache, error)

+ type CacheItem struct

+ Key string

+ TTL *time.Duration

+ Value []byte

+ type Closer interface

+ Close func(ctx context.Context) error

+ type ConfigField struct

+ func NewAnyField(name string) *ConfigField

+ func NewAnyListField(name string) *ConfigField

+ func NewAnyMapField(name string) *ConfigField

+ func NewAutoRetryNacksToggleField() *ConfigField

+ func NewBackOffField(name string, allowUnbounded bool, defaults *backoff.ExponentialBackOff) *ConfigField

+ func NewBackOffToggledField(name string, allowUnbounded bool, defaults *backoff.ExponentialBackOff) *ConfigField

+ func NewBatchPolicyField(name string) *ConfigField

+ func NewBloblangField(name string) *ConfigField

+ func NewBoolField(name string) *ConfigField

+ func NewDurationField(name string) *ConfigField

+ func NewExtractTracingSpanMappingField() *ConfigField

+ func NewFloatField(name string) *ConfigField

+ func NewFloatListField(name string) *ConfigField

+ func NewFloatMapField(name string) *ConfigField

+ func NewHTTPRequestAuthSignerFields() []*ConfigField

+ func NewInjectTracingSpanMappingField() *ConfigField

+ func NewInputField(name string) *ConfigField

+ func NewInputListField(name string) *ConfigField

+ func NewInputMapField(name string) *ConfigField

+ func NewInputMaxInFlightField() *ConfigField

+ func NewIntField(name string) *ConfigField

+ func NewIntListField(name string) *ConfigField

+ func NewIntMapField(name string) *ConfigField

+ func NewInternalField(ifield docs.FieldSpec) *ConfigField

+ func NewInterpolatedStringEnumField(name string, options ...string) *ConfigField

+ func NewInterpolatedStringField(name string) *ConfigField

+ func NewInterpolatedStringListField(name string) *ConfigField

+ func NewInterpolatedStringMapField(name string) *ConfigField

+ func NewMetadataExcludeFilterField(name string) *ConfigField

+ func NewMetadataFilterField(name string) *ConfigField

+ func NewObjectField(name string, fields ...*ConfigField) *ConfigField

+ func NewObjectListField(name string, fields ...*ConfigField) *ConfigField

+ func NewObjectMapField(name string, fields ...*ConfigField) *ConfigField

+ func NewOutputField(name string) *ConfigField

+ func NewOutputListField(name string) *ConfigField

+ func NewOutputMapField(name string) *ConfigField

+ func NewOutputMaxInFlightField() *ConfigField

+ func NewProcessorField(name string) *ConfigField

+ func NewProcessorListField(name string) *ConfigField

+ func NewScannerField(name string) *ConfigField

+ func NewStringAnnotatedEnumField(name string, options map[string]string) *ConfigField

+ func NewStringEnumField(name string, options ...string) *ConfigField

+ func NewStringField(name string) *ConfigField

+ func NewStringListField(name string) *ConfigField

+ func NewStringListOfListsField(name string) *ConfigField

+ func NewStringMapField(name string) *ConfigField

+ func NewTLSField(name string) *ConfigField

+ func NewTLSToggledField(name string) *ConfigField

+ func NewURLField(name string) *ConfigField

+ func NewURLListField(name string) *ConfigField

+ func (c *ConfigField) Advanced() *ConfigField

+ func (c *ConfigField) Default(v any) *ConfigField

+ func (c *ConfigField) Deprecated() *ConfigField

+ func (c *ConfigField) Description(d string) *ConfigField

+ func (c *ConfigField) Example(e any) *ConfigField

+ func (c *ConfigField) Examples(e ...any) *ConfigField

+ func (c *ConfigField) LintRule(blobl string) *ConfigField

+ func (c *ConfigField) Optional() *ConfigField

+ func (c *ConfigField) Secret() *ConfigField

+ func (c *ConfigField) Version(v string) *ConfigField

+ func (c *ConfigField) XUnwrapper() any

+ type ConfigSchema struct

+ func ConfigSchemaFromJSONV0(jBytes []byte) (*ConfigSchema, error)

+ func (s *ConfigSchema) Environment() *Environment

+ func (s *ConfigSchema) Field(f *ConfigField) *ConfigSchema

+ func (s *ConfigSchema) Fields(fs ...*ConfigField) *ConfigSchema

+ func (s *ConfigSchema) MarshalJSONSchema() ([]byte, error)

+ func (s *ConfigSchema) MarshalJSONV0() ([]byte, error)

+ func (s *ConfigSchema) NewStreamConfigLinter() *StreamConfigLinter

+ func (s *ConfigSchema) NewStreamConfigMarshaller() *StreamConfigMarshaller

+ func (s *ConfigSchema) SetVersion(version, dateBuilt string) *ConfigSchema

+ type ConfigSpec struct

+ func NewConfigSpec() *ConfigSpec

+ func (c *ConfigSpec) Beta() *ConfigSpec

+ func (c *ConfigSpec) Categories(categories ...string) *ConfigSpec

+ func (c *ConfigSpec) Deprecated() *ConfigSpec

+ func (c *ConfigSpec) Description(description string) *ConfigSpec

+ func (c *ConfigSpec) EncodeJSON(v []byte) error

+ func (c *ConfigSpec) Example(title, summary, config string) *ConfigSpec

+ func (c *ConfigSpec) Field(f *ConfigField) *ConfigSpec

+ func (c *ConfigSpec) Fields(fs ...*ConfigField) *ConfigSpec

+ func (c *ConfigSpec) Footnotes(description string) *ConfigSpec

+ func (c *ConfigSpec) LintRule(blobl string) *ConfigSpec

+ func (c *ConfigSpec) ParseYAML(yamlStr string, env *Environment) (*ParsedConfig, error)

+ func (c *ConfigSpec) Stable() *ConfigSpec

+ func (c *ConfigSpec) Summary(summary string) *ConfigSpec

+ func (c *ConfigSpec) Version(v string) *ConfigSpec

+ type ConfigView struct

+ func (c *ConfigView) Description() string

+ func (c *ConfigView) FormatJSON() ([]byte, error)

+ func (c *ConfigView) IsDeprecated() bool

+ func (c *ConfigView) RenderDocs() ([]byte, error)

+ func (c *ConfigView) Summary() string

+ type Environment struct

+ func GlobalEnvironment() *Environment

+ func NewEmptyEnvironment() *Environment

+ func NewEnvironment() *Environment

+ func (e *Environment) Clone() *Environment

+ func (e *Environment) CoreConfigSchema(version, dateBuilt string) *ConfigSchema

+ func (e *Environment) FullConfigSchema(version, dateBuilt string) *ConfigSchema

+ func (e *Environment) GenerateSchema(version, dateBuilt string) *EnvironmentSchema

+ func (e *Environment) GetBufferConfig(name string) (*ConfigView, bool)

+ func (e *Environment) GetCacheConfig(name string) (*ConfigView, bool)

+ func (e *Environment) GetInputConfig(name string) (*ConfigView, bool)

+ func (e *Environment) GetMetricsConfig(name string) (*ConfigView, bool)

+ func (e *Environment) GetOutputConfig(name string) (*ConfigView, bool)

+ func (e *Environment) GetProcessorConfig(name string) (*ConfigView, bool)

+ func (e *Environment) GetRateLimitConfig(name string) (*ConfigView, bool)

+ func (e *Environment) GetScannerConfig(name string) (*ConfigView, bool)

+ func (e *Environment) GetTracerConfig(name string) (*ConfigView, bool)

+ func (e *Environment) NewStreamBuilder() *StreamBuilder

+ func (e *Environment) NewStreamTemplateTester() *StreamTemplateTester

+ func (e *Environment) RegisterBatchBuffer(name string, spec *ConfigSpec, ctor BatchBufferConstructor) error

+ func (e *Environment) RegisterBatchInput(name string, spec *ConfigSpec, ctor BatchInputConstructor) error

+ func (e *Environment) RegisterBatchOutput(name string, spec *ConfigSpec, ctor BatchOutputConstructor) error

+ func (e *Environment) RegisterBatchProcessor(name string, spec *ConfigSpec, ctor BatchProcessorConstructor) error

+ func (e *Environment) RegisterBatchScannerCreator(name string, spec *ConfigSpec, ctor BatchScannerCreatorConstructor) error

+ func (e *Environment) RegisterCache(name string, spec *ConfigSpec, ctor CacheConstructor) error

+ func (e *Environment) RegisterInput(name string, spec *ConfigSpec, ctor InputConstructor) error

+ func (e *Environment) RegisterMetricsExporter(name string, spec *ConfigSpec, ctor MetricsExporterConstructor) error

+ func (e *Environment) RegisterOtelTracerProvider(name string, spec *ConfigSpec, ctor OtelTracerProviderConstructor) error

+ func (e *Environment) RegisterOutput(name string, spec *ConfigSpec, ctor OutputConstructor) error

+ func (e *Environment) RegisterProcessor(name string, spec *ConfigSpec, ctor ProcessorConstructor) error

+ func (e *Environment) RegisterRateLimit(name string, spec *ConfigSpec, ctor RateLimitConstructor) error

+ func (e *Environment) RegisterTemplateYAML(yamlStr string) error

+ func (e *Environment) UseBloblangEnvironment(bEnv *bloblang.Environment)

+ func (e *Environment) UseFS(fs *FS)

+ func (e *Environment) WalkBuffers(fn func(name string, config *ConfigView))

+ func (e *Environment) WalkCaches(fn func(name string, config *ConfigView))

+ func (e *Environment) WalkInputs(fn func(name string, config *ConfigView))

+ func (e *Environment) WalkMetrics(fn func(name string, config *ConfigView))

+ func (e *Environment) WalkOutputs(fn func(name string, config *ConfigView))

+ func (e *Environment) WalkProcessors(fn func(name string, config *ConfigView))

+ func (e *Environment) WalkRateLimits(fn func(name string, config *ConfigView))

+ func (e *Environment) WalkScanners(fn func(name string, config *ConfigView))

+ func (e *Environment) WalkTracers(fn func(name string, config *ConfigView))

+ type EnvironmentSchema struct

+ func (e *EnvironmentSchema) Minimise() *EnvironmentSchema

+ func (e *EnvironmentSchema) ReduceToStatus(status string) *EnvironmentSchema

+ func (e *EnvironmentSchema) ToCUE() ([]byte, error)

+ func (e *EnvironmentSchema) XFlattened() map[string][]string

+ type ErrBackOff struct

+ Err error

+ Wait time.Duration

+ func NewErrBackOff(err error, wait time.Duration) *ErrBackOff

+ func (e *ErrBackOff) Error() string

+ type FS struct

+ func NewFS(filesystem fs.FS) *FS

+ func (f *FS) MkdirAll(path string, perm fs.FileMode) error

+ func (f *FS) Open(name string) (fs.File, error)

+ func (f *FS) OpenFile(name string, flag int, perm fs.FileMode) (fs.File, error)

+ func (f *FS) Remove(name string) error

+ func (f *FS) Stat(name string) (fs.FileInfo, error)

+ type FieldView struct

+ func (f *FieldView) IsAdvanced() bool

+ type HTTPMultiplexer interface

+ HandleFunc func(pattern string, handler func(http.ResponseWriter, *http.Request))

+ type Indexer struct

+ func (s *Indexer) IndexOf(m *Message) int

+ type Input interface

+ Connect func(context.Context) error

+ Read func(context.Context) (*Message, AckFunc, error)

+ func AutoRetryNacks(i Input) Input

+ func AutoRetryNacksToggled(c *ParsedConfig, i Input) (Input, error)

+ func InputWithMaxInFlight(n int, i Input) Input

+ type InputConstructor func(conf *ParsedConfig, mgr *Resources) (Input, error)

+ type InterpolatedString struct

+ func NewInterpolatedString(expr string) (*InterpolatedString, error)

+ func (i *InterpolatedString) Bytes(m *Message) []byte

+ func (i *InterpolatedString) Static() (string, bool)

+ func (i *InterpolatedString) String(m *Message) string

+ func (i *InterpolatedString) TryBytes(m *Message) ([]byte, error)

+ func (i *InterpolatedString) TryString(m *Message) (string, error)

+ type Lint struct

+ Column int

+ Line int

+ Type LintType

+ What string

+ func (l Lint) Error() string

+ type LintError []Lint

+ func (e LintError) Error() string

+ type LintType int

+ const LintBadBloblang

+ const LintBadLabel

+ const LintComponentMissing

+ const LintComponentNotFound

+ const LintCustom

+ const LintDeprecated

+ const LintDuplicateLabel

+ const LintExpectedArray

+ const LintExpectedObject

+ const LintExpectedScalar

+ const LintFailedRead

+ const LintInvalidOption

+ const LintMissing

+ const LintMissingEnvVar

+ const LintMissingLabel

+ const LintShouldOmit

+ const LintUnknown

+ type Logger struct

+ func (l *Logger) Debug(message string)

+ func (l *Logger) Debugf(template string, args ...any)

+ func (l *Logger) Error(message string)

+ func (l *Logger) Errorf(template string, args ...any)

+ func (l *Logger) Info(message string)

+ func (l *Logger) Infof(template string, args ...any)

+ func (l *Logger) Trace(message string)

+ func (l *Logger) Tracef(template string, args ...any)

+ func (l *Logger) Warn(message string)

+ func (l *Logger) Warnf(template string, args ...any)

+ func (l *Logger) With(keyValuePairs ...any) *Logger

+ type Message struct

+ func NewInternalMessage(imsg *message.Part) *Message

+ func NewMessage(content []byte) *Message

+ func (m *Message) AsBytes() ([]byte, error)

+ func (m *Message) AsStructured() (any, error)

+ func (m *Message) AsStructuredMut() (any, error)

+ func (m *Message) BloblangMutate(blobl *bloblang.Executor) (*Message, error)

+ func (m *Message) BloblangMutateFrom(blobl *bloblang.Executor, from *Message) (*Message, error)

+ func (m *Message) BloblangQuery(blobl *bloblang.Executor) (*Message, error)

+ func (m *Message) BloblangQueryValue(blobl *bloblang.Executor) (any, error)

+ func (m *Message) Context() context.Context

+ func (m *Message) Copy() *Message

+ func (m *Message) DeepCopy() *Message

+ func (m *Message) GetError() error

+ func (m *Message) MetaDelete(key string)

+ func (m *Message) MetaGet(key string) (string, bool)

+ func (m *Message) MetaGetMut(key string) (any, bool)

+ func (m *Message) MetaSet(key, value string)

+ func (m *Message) MetaSetMut(key string, value any)

+ func (m *Message) MetaWalk(fn func(string, string) error) error

+ func (m *Message) MetaWalkMut(fn func(key string, value any) error) error

+ func (m *Message) SetBytes(b []byte)

+ func (m *Message) SetError(err error)

+ func (m *Message) SetStructured(i any)

+ func (m *Message) SetStructuredMut(i any)

+ func (m *Message) WithContext(ctx context.Context) *Message

+ type MessageBatch []*Message

+ func ExecuteProcessors(ctx context.Context, processors []*OwnedProcessor, inbatches ...MessageBatch) ([]MessageBatch, error)

+ func (b MessageBatch) AddSyncResponse() error

+ func (b MessageBatch) BloblangMutate(index int, blobl *bloblang.Executor) (*Message, error)

+ func (b MessageBatch) BloblangQuery(index int, blobl *bloblang.Executor) (*Message, error)

+ func (b MessageBatch) BloblangQueryValue(index int, blobl *bloblang.Executor) (any, error)

+ func (b MessageBatch) Copy() MessageBatch

+ func (b MessageBatch) DeepCopy() MessageBatch

+ func (b MessageBatch) Index() *Indexer

+ func (b MessageBatch) InterpolatedBytes(index int, i *InterpolatedString) []byte

+ func (b MessageBatch) InterpolatedString(index int, i *InterpolatedString) string

+ func (b MessageBatch) TryInterpolatedBytes(index int, i *InterpolatedString) ([]byte, error)

+ func (b MessageBatch) TryInterpolatedString(index int, i *InterpolatedString) (string, error)

+ func (b MessageBatch) WalkWithBatchedErrors(fn func(int, *Message) error) error

+ type MessageBatchHandlerFunc func(context.Context, MessageBatch) error

+ type MessageHandlerFunc func(context.Context, *Message) error

+ type MetadataExcludeFilter struct

+ func (m *MetadataExcludeFilter) Walk(msg *Message, fn func(key, value string) error) error

+ func (m *MetadataExcludeFilter) WalkMut(msg *Message, fn func(key string, value any) error) error

+ type MetadataFilter struct

+ func (m *MetadataFilter) IsEmpty() bool

+ func (m *MetadataFilter) Match(k string) bool

+ func (m *MetadataFilter) Walk(msg *Message, fn func(key, value string) error) error

+ func (m *MetadataFilter) WalkMut(msg *Message, fn func(key string, value any) error) error

+ type MetricCounter struct

+ func (c *MetricCounter) Incr(count int64, labelValues ...string)

+ func (c *MetricCounter) IncrFloat64(count float64, labelValues ...string)

+ type MetricGauge struct

+ func (g *MetricGauge) Set(value int64, labelValues ...string)

+ func (g *MetricGauge) SetFloat64(value float64, labelValues ...string)

+ type MetricTimer struct

+ func (t *MetricTimer) Timing(delta int64, labelValues ...string)

+ type Metrics struct

+ func (m *Metrics) NewCounter(name string, labelKeys ...string) *MetricCounter

+ func (m *Metrics) NewGauge(name string, labelKeys ...string) *MetricGauge

+ func (m *Metrics) NewTimer(name string, labelKeys ...string) *MetricTimer

+ type MetricsExporter interface

+ Close func(ctx context.Context) error

+ NewCounterCtor func(name string, labelKeys ...string) MetricsExporterCounterCtor

+ NewGaugeCtor func(name string, labelKeys ...string) MetricsExporterGaugeCtor

+ NewTimerCtor func(name string, labelKeys ...string) MetricsExporterTimerCtor

+ type MetricsExporterConstructor func(conf *ParsedConfig, log *Logger) (MetricsExporter, error)

+ type MetricsExporterCounter interface

+ Incr func(count int64)

+ type MetricsExporterCounterCtor func(labelValues ...string) MetricsExporterCounter

+ type MetricsExporterGauge interface

+ Set func(value int64)

+ type MetricsExporterGaugeCtor func(labelValues ...string) MetricsExporterGauge

+ type MetricsExporterTimer interface

+ Timing func(delta int64)

+ type MetricsExporterTimerCtor func(labelValues ...string) MetricsExporterTimer

+ type MockResourcesOptFn func(*mock.Manager)

+ func MockResourcesOptAddCache(name string) MockResourcesOptFn

+ func MockResourcesOptAddRateLimit(name string, fn func(context.Context) (time.Duration, error)) MockResourcesOptFn

+ func MockResourcesOptUseLogger(l *Logger) MockResourcesOptFn

+ type OtelTracerProviderConstructor func(conf *ParsedConfig) (trace.TracerProvider, error)

+ type Output interface

+ Connect func(context.Context) error

+ Write func(context.Context, *Message) error

+ type OutputConstructor func(conf *ParsedConfig, mgr *Resources) (out Output, maxInFlight int, err error)

+ type OwnedInput struct

+ func (o *OwnedInput) BatchedWith(b *Batcher) *OwnedInput

+ func (o *OwnedInput) Close(ctx context.Context) error

+ func (o *OwnedInput) ReadBatch(ctx context.Context) (MessageBatch, AckFunc, error)

+ func (o *OwnedInput) XUnwrapper() any

+ type OwnedOutput struct

+ func (o *OwnedOutput) BatchedWith(b *Batcher) *OwnedOutput

+ func (o *OwnedOutput) Close(ctx context.Context) error

+ func (o *OwnedOutput) Prime() error

+ func (o *OwnedOutput) PrimeBuffered(n int) error

+ func (o *OwnedOutput) Write(ctx context.Context, m *Message) error

+ func (o *OwnedOutput) WriteBatch(ctx context.Context, b MessageBatch) error

+ func (o *OwnedOutput) WriteBatchNonBlocking(b MessageBatch, aFn AckFunc) error

+ func (o *OwnedOutput) XUnwrapper() any

+ type OwnedProcessor struct

+ func (o *OwnedProcessor) Close(ctx context.Context) error

+ func (o *OwnedProcessor) Process(ctx context.Context, msg *Message) (MessageBatch, error)

+ func (o *OwnedProcessor) ProcessBatch(ctx context.Context, batch MessageBatch) ([]MessageBatch, error)

+ func (o *OwnedProcessor) XUnwrapper() any

+ type OwnedScanner struct

+ func (s *OwnedScanner) Close(ctx context.Context) error

+ func (s *OwnedScanner) NextBatch(ctx context.Context) (MessageBatch, AckFunc, error)

+ type OwnedScannerCreator struct

+ func (s *OwnedScannerCreator) Close(ctx context.Context) error

+ func (s *OwnedScannerCreator) Create(rdr io.ReadCloser, aFn AckFunc, details *ScannerSourceDetails) (*OwnedScanner, error)

+ type ParsedConfig struct

+ func (p *ParsedConfig) Contains(path ...string) bool

+ func (p *ParsedConfig) EngineVersion() string

+ func (p *ParsedConfig) FieldAny(path ...string) (any, error)

+ func (p *ParsedConfig) FieldAnyList(path ...string) ([]*ParsedConfig, error)

+ func (p *ParsedConfig) FieldAnyMap(path ...string) (map[string]*ParsedConfig, error)

+ func (p *ParsedConfig) FieldBackOff(path ...string) (*backoff.ExponentialBackOff, error)

+ func (p *ParsedConfig) FieldBackOffToggled(path ...string) (boff *backoff.ExponentialBackOff, enabled bool, err error)

+ func (p *ParsedConfig) FieldBatchPolicy(path ...string) (conf BatchPolicy, err error)

+ func (p *ParsedConfig) FieldBloblang(path ...string) (*bloblang.Executor, error)

+ func (p *ParsedConfig) FieldBool(path ...string) (bool, error)

+ func (p *ParsedConfig) FieldDuration(path ...string) (time.Duration, error)

+ func (p *ParsedConfig) FieldFloat(path ...string) (float64, error)

+ func (p *ParsedConfig) FieldFloatList(path ...string) ([]float64, error)

+ func (p *ParsedConfig) FieldFloatMap(path ...string) (map[string]float64, error)

+ func (p *ParsedConfig) FieldInput(path ...string) (*OwnedInput, error)

+ func (p *ParsedConfig) FieldInputList(path ...string) ([]*OwnedInput, error)

+ func (p *ParsedConfig) FieldInputMap(path ...string) (map[string]*OwnedInput, error)

+ func (p *ParsedConfig) FieldInt(path ...string) (int, error)

+ func (p *ParsedConfig) FieldIntList(path ...string) ([]int, error)

+ func (p *ParsedConfig) FieldIntMap(path ...string) (map[string]int, error)

+ func (p *ParsedConfig) FieldInterpolatedString(path ...string) (*InterpolatedString, error)

+ func (p *ParsedConfig) FieldInterpolatedStringList(path ...string) ([]*InterpolatedString, error)

+ func (p *ParsedConfig) FieldInterpolatedStringMap(path ...string) (map[string]*InterpolatedString, error)

+ func (p *ParsedConfig) FieldMaxInFlight() (int, error)

+ func (p *ParsedConfig) FieldMetadataExcludeFilter(path ...string) (f *MetadataExcludeFilter, err error)

+ func (p *ParsedConfig) FieldMetadataFilter(path ...string) (f *MetadataFilter, err error)

+ func (p *ParsedConfig) FieldObjectList(path ...string) ([]*ParsedConfig, error)

+ func (p *ParsedConfig) FieldObjectMap(path ...string) (map[string]*ParsedConfig, error)

+ func (p *ParsedConfig) FieldOutput(path ...string) (*OwnedOutput, error)

+ func (p *ParsedConfig) FieldOutputList(path ...string) ([]*OwnedOutput, error)

+ func (p *ParsedConfig) FieldOutputMap(path ...string) (map[string]*OwnedOutput, error)

+ func (p *ParsedConfig) FieldProcessor(path ...string) (*OwnedProcessor, error)

+ func (p *ParsedConfig) FieldProcessorList(path ...string) ([]*OwnedProcessor, error)

+ func (p *ParsedConfig) FieldScanner(path ...string) (*OwnedScannerCreator, error)

+ func (p *ParsedConfig) FieldString(path ...string) (string, error)

+ func (p *ParsedConfig) FieldStringList(path ...string) ([]string, error)

+ func (p *ParsedConfig) FieldStringListOfLists(path ...string) ([][]string, error)

+ func (p *ParsedConfig) FieldStringMap(path ...string) (map[string]string, error)

+ func (p *ParsedConfig) FieldTLS(path ...string) (*tls.Config, error)

+ func (p *ParsedConfig) FieldTLSToggled(path ...string) (tconf *tls.Config, enabled bool, err error)

+ func (p *ParsedConfig) FieldURL(path ...string) (*url.URL, error)

+ func (p *ParsedConfig) FieldURLList(path ...string) ([]*url.URL, error)

+ func (p *ParsedConfig) HTTPRequestAuthSignerFromParsed() (fn func(fs.FS, *http.Request) error, err error)

+ func (p *ParsedConfig) Namespace(path ...string) *ParsedConfig

+ func (p *ParsedConfig) WrapBatchInputExtractTracingSpanMapping(inputName string, i BatchInput) (BatchInput, error)

+ func (p *ParsedConfig) WrapBatchOutputExtractTracingSpanMapping(outputName string, o BatchOutput) (BatchOutput, error)

+ func (p *ParsedConfig) WrapInputExtractTracingSpanMapping(inputName string, i Input) (Input, error)

+ func (p *ParsedConfig) WrapOutputExtractTracingSpanMapping(outputName string, o Output) (Output, error)

+ type PrintLogger interface

+ Printf func(format string, v ...any)

+ Println func(v ...any)

+ type Processor interface

+ Process func(context.Context, *Message) (MessageBatch, error)

+ type ProcessorConstructor func(conf *ParsedConfig, mgr *Resources) (Processor, error)

+ type RateLimit interface

+ Access func(context.Context) (time.Duration, error)

+ type RateLimitConstructor func(conf *ParsedConfig, mgr *Resources) (RateLimit, error)

+ type ResourceInput struct

+ func (r *ResourceInput) ReadBatch(ctx context.Context) (MessageBatch, AckFunc, error)

+ type ResourceOutput struct

+ func (o *ResourceOutput) Write(ctx context.Context, m *Message) error

+ func (o *ResourceOutput) WriteBatch(ctx context.Context, b MessageBatch) error

+ type Resources struct

+ func MockResources(opts ...MockResourcesOptFn) *Resources

+ func (r *Resources) AccessCache(ctx context.Context, name string, fn func(c Cache)) error

+ func (r *Resources) AccessInput(ctx context.Context, name string, fn func(i *ResourceInput)) error

+ func (r *Resources) AccessOutput(ctx context.Context, name string, fn func(o *ResourceOutput)) error

+ func (r *Resources) AccessRateLimit(ctx context.Context, name string, fn func(r RateLimit)) error

+ func (r *Resources) EngineVersion() string

+ func (r *Resources) FS() *FS

+ func (r *Resources) HasCache(name string) bool

+ func (r *Resources) HasInput(name string) bool

+ func (r *Resources) HasOutput(name string) bool

+ func (r *Resources) HasRateLimit(name string) bool

+ func (r *Resources) Label() string

+ func (r *Resources) Logger() *Logger

+ func (r *Resources) ManagedBatchOutput(typeName string, maxInFlight int, b BatchOutput) (*OwnedOutput, error)

+ func (r *Resources) Metrics() *Metrics

+ func (r *Resources) OtelTracer() trace.TracerProvider

+ func (r *Resources) XUnwrapper() any

+ type ScannerSourceDetails struct

+ func NewScannerSourceDetails() *ScannerSourceDetails

+ func (r *ScannerSourceDetails) Name() string

+ func (r *ScannerSourceDetails) SetName(name string)

+ type SimpleBatchScanner interface

+ Close func(context.Context) error

+ NextBatch func(context.Context) (MessageBatch, error)

+ type Stream struct

+ func (s *Stream) Run(ctx context.Context) (err error)

+ func (s *Stream) Stop(ctx context.Context) (err error)

+ func (s *Stream) StopWithin(timeout time.Duration) error

+ type StreamBuilder struct

+ func NewStreamBuilder() *StreamBuilder

+ func (s *StreamBuilder) AddBatchConsumerFunc(fn MessageBatchHandlerFunc) error

+ func (s *StreamBuilder) AddBatchProducerFunc() (MessageBatchHandlerFunc, error)

+ func (s *StreamBuilder) AddCacheYAML(conf string) error

+ func (s *StreamBuilder) AddConsumerFunc(fn MessageHandlerFunc) error

+ func (s *StreamBuilder) AddInputYAML(conf string) error

+ func (s *StreamBuilder) AddOutputYAML(conf string) error

+ func (s *StreamBuilder) AddProcessorYAML(conf string) error

+ func (s *StreamBuilder) AddProducerFunc() (MessageHandlerFunc, error)

+ func (s *StreamBuilder) AddRateLimitYAML(conf string) error

+ func (s *StreamBuilder) AddResourcesYAML(conf string) error

+ func (s *StreamBuilder) AsYAML() (string, error)

+ func (s *StreamBuilder) Build() (*Stream, error)

+ func (s *StreamBuilder) BuildTraced() (*Stream, *TracingSummary, error)

+ func (s *StreamBuilder) DisableLinting()

+ func (s *StreamBuilder) SetBufferYAML(conf string) error

+ func (s *StreamBuilder) SetEngineVersion(ev string)

+ func (s *StreamBuilder) SetEnvVarLookupFunc(fn func(string) (string, bool))

+ func (s *StreamBuilder) SetFields(pathValues ...any) error

+ func (s *StreamBuilder) SetHTTPMux(m HTTPMultiplexer)

+ func (s *StreamBuilder) SetLogger(l *slog.Logger)

+ func (s *StreamBuilder) SetLoggerYAML(conf string) error

+ func (s *StreamBuilder) SetMetricsYAML(conf string) error

+ func (s *StreamBuilder) SetPrintLogger(l PrintLogger)

+ func (s *StreamBuilder) SetThreads(n int)

+ func (s *StreamBuilder) SetTracerYAML(conf string) error

+ func (s *StreamBuilder) SetYAML(conf string) error

+ func (s *StreamBuilder) WalkComponents(fn func(w *WalkedComponent) error) error

+ type StreamConfigLinter struct

+ func (s *StreamConfigLinter) LintYAML(yamlBytes []byte) (lints []Lint, err error)

+ func (s *StreamConfigLinter) SetEnvVarLookupFunc(fn func(context.Context, string) (string, bool)) *StreamConfigLinter

+ func (s *StreamConfigLinter) SetRejectDeprecated(v bool) *StreamConfigLinter

+ func (s *StreamConfigLinter) SetRequireLabels(v bool) *StreamConfigLinter

+ func (s *StreamConfigLinter) SetSkipEnvVarCheck(v bool) *StreamConfigLinter

+ type StreamConfigMarshaller struct

+ func (s *StreamConfigMarshaller) AnyToYAML(v any) (yamlStr string, err error)

+ func (s *StreamConfigMarshaller) SetFieldFilter(fn func(view *FieldView, value any) bool) *StreamConfigMarshaller

+ func (s *StreamConfigMarshaller) SetHydrateExamples(v bool) *StreamConfigMarshaller

+ func (s *StreamConfigMarshaller) SetOmitDeprecated(v bool) *StreamConfigMarshaller

+ func (s *StreamConfigMarshaller) SetScrubSecrets(v bool) *StreamConfigMarshaller

+ type StreamTemplateTester struct

+ func (s *StreamTemplateTester) LintYAML(yamlBytes []byte) (lints []Lint, err error)

+ func (s *StreamTemplateTester) RunYAML(yamlBytes []byte) (lints []Lint, err error)

+ type TracingEvent struct

+ Content string

+ Meta map[string]any

+ Type TracingEventType

+ type TracingEventType string

+ var TracingEventConsume TracingEventType = "CONSUME"

+ var TracingEventDelete TracingEventType = "DELETE"

+ var TracingEventError TracingEventType = "ERROR"

+ var TracingEventProduce TracingEventType = "PRODUCE"

+ var TracingEventUnknown TracingEventType = "UNKNOWN"

+ type TracingSummary struct

+ func (s *TracingSummary) InputEvents() map[string][]TracingEvent

+ func (s *TracingSummary) OutputEvents() map[string][]TracingEvent

+ func (s *TracingSummary) ProcessorEvents() map[string][]TracingEvent

+ func (s *TracingSummary) TotalInput() uint64

+ func (s *TracingSummary) TotalOutput() uint64

+ func (s *TracingSummary) TotalProcessorErrors() uint64

+ type WalkedComponent struct

+ ComponentType string

+ Label string

+ Name string

+ func (w *WalkedComponent) ConfigYAML() string

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL