service

package

Go to main page

Versions in this module

v3

v3.65.0

Mar 7, 2022

v3.64.0

Feb 23, 2022

v3.63.0

Feb 8, 2022

Changes in this version

type BatchPolicy

+ func (b BatchPolicy) NewBatcher(res *Resources) (*Batcher, 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)

+ type CacheItem struct

+ Key string

+ TTL *time.Duration

+ Value []byte

type ConfigField

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

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

+ func NewDurationField(name string) *ConfigField

type ConfigSpec

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

type ParsedConfig

+ 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) FieldDuration(path ...string) (time.Duration, error)

v3.62.0

Jan 21, 2022

Changes in this version

type ConfigField

+ func NewInputField(name string) *ConfigField

+ func NewInputListField(name string) *ConfigField

+ func NewOutputField(name string) *ConfigField

+ func NewOutputListField(name string) *ConfigField

+ func NewProcessorField(name string) *ConfigField

+ func NewProcessorListField(name string) *ConfigField

+ type OwnedInput struct

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

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

+ type OwnedOutput struct

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

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

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

+ 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)

type ParsedConfig

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

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

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

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

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

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

v3.61.0

Dec 28, 2021

Changes in this version

type ConfigField

+ func NewMetadataFilterField(name string) *ConfigField

type MessageBatch

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

+ type MetadataFilter struct

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

type ParsedConfig

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

type StreamBuilder

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

+ func (s *StreamBuilder) DisableLinting()

+ type TracingEvent struct

+ Content string

+ 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

v3.60.1

Dec 3, 2021

v3.60.0

Dec 1, 2021

Changes in this version

type ConfigField

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

type Environment

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

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

type ParsedConfig

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

v3.59.0

Nov 22, 2021

Changes in this version

type ConfigField

+ func NewInternalField(ifield docs.FieldSpec) *ConfigField

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

+ func NewTLSToggledField(name string) *ConfigField

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

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

type ParsedConfig

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

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

v3.59.0-rc2

Nov 17, 2021

v3.59.0-rc1

Nov 10, 2021

v3.58.0

Nov 2, 2021

v3.57.0

Oct 14, 2021

Changes in this version

type ConfigField

+ func NewIntListField(name string) *ConfigField

+ func NewIntMapField(name string) *ConfigField

+ func NewStringMapField(name string) *ConfigField

type ParsedConfig

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

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

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

v3.56.0

Sep 22, 2021

Changes in this version

type Logger

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

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

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

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

v3.55.0

Sep 8, 2021

Changes in this version

type ConfigField

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

type ConfigSpec

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

v3.54.1

Sep 3, 2021

v3.54.0

Sep 1, 2021

Changes in this version

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

type StreamBuilder

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

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

v3.53.0

Aug 19, 2021

Changes in this version

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

+ func RegisterBatchBuffer(name string, spec *ConfigSpec, ctor BatchBufferConstructor) 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 ConfigView

+ func (c *ConfigView) IsDeprecated() bool

type Environment

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

type Message

+ func (m *Message) GetError() error

type ParsedConfig

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

type StreamBuilder

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

+ func (s *StreamBuilder) SetFields(pathValues ...interface{}) error

v3.52.0

Aug 2, 2021

Changes in this version

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

+ type BatchInput interface

+ Connect func(context.Context) error

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

+ func AutoRetryNacksBatched(i BatchInput) BatchInput

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

type Environment

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

type Message

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

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

v3.51.0

Jul 26, 2021

Changes in this version

+ 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 RegisterBatchOutput(name string, spec *ConfigSpec, ctor BatchOutputConstructor) error

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

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

+ func RegisterInput(name string, spec *ConfigSpec, ctor InputConstructor) 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 RunCLI(ctx context.Context)

+ type AckFunc func(ctx context.Context, err error) 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

+ type BatchProcessor interface

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

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

+ 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 Closer interface

+ Close func(ctx context.Context) error

+ type ConfigField struct

+ func NewBatchPolicyField(name string) *ConfigField

+ func NewBloblangField(name string) *ConfigField

+ func NewBoolField(name string) *ConfigField

+ func NewFloatField(name string) *ConfigField

+ func NewIntField(name string) *ConfigField

+ func NewInterpolatedStringField(name string) *ConfigField

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

+ func NewStringField(name string) *ConfigField

+ func NewStringListField(name string) *ConfigField

+ func NewTLSField(name string) *ConfigField

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

+ func (c *ConfigField) Default(v interface{}) *ConfigField

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

+ func (c *ConfigField) Example(e interface{}) *ConfigField

+ type ConfigSpec struct

+ func NewConfigSpec() *ConfigSpec

+ func NewStructConfigSpec(ctor ConfigStructConstructor) (*ConfigSpec, error)

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

+ func (c *ConfigSpec) Categories(categories ...string) *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) Stable() *ConfigSpec

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

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

+ type ConfigStructConstructor func() interface

+ type ConfigView struct

+ func (c *ConfigView) Description() string

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

+ func (c *ConfigView) Summary() string

+ type Environment struct

+ func NewEnvironment() *Environment

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

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

+ 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) RegisterCache(name string, spec *ConfigSpec, ctor CacheConstructor) error

+ func (e *Environment) RegisterInput(name string, spec *ConfigSpec, ctor InputConstructor) 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) UseBloblangEnvironment(bEnv *bloblang.Environment)

+ 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) 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))

+ type HTTPMultiplexer interface

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

+ type Input interface

+ Connect func(context.Context) error

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

+ func AutoRetryNacks(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) String(m *Message) string

+ type Lint struct

+ Line int

+ What string

+ type LintError []Lint

+ func (e LintError) Error() string

+ type Logger struct

+ func (l *Logger) Debugf(template string, args ...interface{})

+ func (l *Logger) Errorf(template string, args ...interface{})

+ func (l *Logger) Infof(template string, args ...interface{})

+ func (l *Logger) Warnf(template string, args ...interface{})

+ func (l *Logger) With(keyValuePairs ...interface{}) *Logger

+ type Message struct

+ func NewMessage(content []byte) *Message

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

+ func (m *Message) AsStructured() (interface{}, error)

+ func (m *Message) AsStructuredMut() (interface{}, error)

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

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

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

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

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

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

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

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

+ func (m *Message) SetStructured(i interface{})

+ type MessageBatch []*Message

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

+ func (b MessageBatch) Copy() MessageBatch

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

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

+ type MetricCounter struct

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

+ type MetricGauge struct

+ func (g *MetricGauge) Set(value int64, 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 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 ParsedConfig struct

+ func (p *ParsedConfig) AsStruct() interface{}

+ 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) FieldFloat(path ...string) (float64, error)

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

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

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

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

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

+ type PrintLogger interface

+ Printf func(format string, v ...interface{})

+ Println func(v ...interface{})

+ 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 Resources struct

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

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

+ func (r *Resources) Label() string

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

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

+ type Stream struct

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

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

+ type StreamBuilder struct

+ func NewStreamBuilder() *StreamBuilder

+ 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) SetHTTPMux(m HTTPMultiplexer)

+ 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) SetYAML(conf string) error

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