goka

package module

Go to main page

Versions in this module

v0

v0.1.4

Jan 22, 2020

Changes in this version

type Context

+ Headers func() map[string][]byte

v0.1.3

Jun 28, 2019

Changes in this version

+ func DefaultRebalance(a kafka.Assignment)

type ProcessorOption

+ func WithRebalanceCallback(cb RebalanceCallback) ProcessorOption

+ type RebalanceCallback func(a kafka.Assignment)

v0.1.2

Jun 18, 2019

Changes in this version

type Iterator

+ Err func() error

type ProcessorOption

+ func WithGroupGraphHook(hook func(gg *GroupGraph)) ProcessorOption

type Tester

+ EmitterProducerBuilder func() kafka.ProducerBuilder

+ RegisterView func(Table, Codec)

type ViewOption

+ func WithViewTester(t Tester) ViewOption

v0.1.1

Jan 29, 2019

Changes in this version

type Context

+ Context func() context.Context

type EmitterOption

+ func WithEmitterTester(t Tester) EmitterOption

type Tester

+ RegisterEmitter func(Stream, Codec)

+ RegisterGroupGraph func(*GroupGraph)

v0.1.0

Aug 3, 2018

Changes in this version

+ func DefaultHasher() func() hash.Hash32

+ func DefaultProcessorStoragePath(group Group) string

+ func DefaultUpdate(s storage.Storage, partition int32, key string, value []byte) error

+ func DefaultViewStoragePath() string

+ func NewConstHasher(part uint32) hash.Hash32

+ type Codec interface

+ Decode func(data []byte) (value interface{}, err error)

+ Encode func(value interface{}) (data []byte, err error)

+ type Context interface

+ Delete func()

+ Emit func(topic Stream, key string, value interface{})

+ Fail func(err error)

+ Join func(topic Table) interface{}

+ Key func() string

+ Lookup func(topic Table, key string) interface{}

+ Loopback func(key string, value interface{})

+ Offset func() int64

+ Partition func() int32

+ SetValue func(value interface{})

+ Timestamp func() time.Time

+ Topic func() Stream

+ Value func() interface{}

+ type Edge interface

+ Codec func() Codec

+ String func() string

+ Topic func() string

+ func Input(topic Stream, c Codec, cb ProcessCallback) Edge

+ func Inputs(topics Streams, c Codec, cb ProcessCallback) Edge

+ func Join(topic Table, c Codec) Edge

+ func Lookup(topic Table, c Codec) Edge

+ func Loop(c Codec, cb ProcessCallback) Edge

+ func Output(topic Stream, c Codec) Edge

+ func Persist(c Codec) Edge

+ type Edges []Edge

+ func (e Edges) Topics() []string

+ type Emitter struct

+ func NewEmitter(brokers []string, topic Stream, codec Codec, options ...EmitterOption) (*Emitter, error)

+ func (e *Emitter) Emit(key string, msg interface{}) (*kafka.Promise, error)

+ func (e *Emitter) EmitSync(key string, msg interface{}) error

+ func (e *Emitter) Finish() error

+ type EmitterOption func(*eoptions)

+ func WithEmitterClientID(clientID string) EmitterOption

+ func WithEmitterHasher(hasher func() hash.Hash32) EmitterOption

+ func WithEmitterLogger(log logger.Logger) EmitterOption

+ func WithEmitterProducerBuilder(pb kafka.ProducerBuilder) EmitterOption

+ func WithEmitterTopicManagerBuilder(tmb kafka.TopicManagerBuilder) EmitterOption

+ type Getter func(string) (interface{}, error)

+ type Group string

+ type GroupGraph struct

+ func DefineGroup(group Group, edges ...Edge) *GroupGraph

+ func (gg *GroupGraph) Group() Group

+ func (gg *GroupGraph) GroupTable() Edge

+ func (gg *GroupGraph) InputStreams() Edges

+ func (gg *GroupGraph) JointTables() Edges

+ func (gg *GroupGraph) LookupTables() Edges

+ func (gg *GroupGraph) LoopStream() Edge

+ func (gg *GroupGraph) OutputStreams() Edges

+ func (gg *GroupGraph) Validate() error

+ type InputStats struct

+ Bytes int

+ Count uint

+ Delay time.Duration

+ type Iterator interface

+ Key func() string

+ Next func() bool

+ Release func()

+ Seek func(key string) bool

+ Value func() (interface{}, error)

+ type NilHandling int

+ const NilDecode

+ const NilIgnore

+ const NilProcess

+ type OutputStats struct

+ Bytes int

+ Count uint

+ type PartitionStats struct

+ Input map[string]InputStats

+ Now time.Time

+ Output map[string]OutputStats

+ Table struct{ ... }

+ type PartitionStatus int

+ const PartitionPreparing

+ const PartitionRecovering

+ const PartitionRunning

+ type ProcessCallback func(ctx Context, msg interface{})

+ type Processor struct

+ func NewProcessor(brokers []string, gg *GroupGraph, options ...ProcessorOption) (*Processor, error)

+ func (g *Processor) Get(key string) (interface{}, error)

+ func (g *Processor) Graph() *GroupGraph

+ func (g *Processor) Recovered() bool

+ func (g *Processor) Run(ctx context.Context) (rerr error)

+ func (g *Processor) Stats() *ProcessorStats

+ type ProcessorOption func(*poptions)

+ func WithClientID(clientID string) ProcessorOption

+ func WithConsumerBuilder(cb kafka.ConsumerBuilder) ProcessorOption

+ func WithHasher(hasher func() hash.Hash32) ProcessorOption

+ func WithLogger(log logger.Logger) ProcessorOption

+ func WithNilHandling(nh NilHandling) ProcessorOption

+ func WithPartitionChannelSize(size int) ProcessorOption

+ func WithProducerBuilder(pb kafka.ProducerBuilder) ProcessorOption

+ func WithStorageBuilder(sb storage.Builder) ProcessorOption

+ func WithTester(t Tester) ProcessorOption

+ func WithTopicManagerBuilder(tmb kafka.TopicManagerBuilder) ProcessorOption

+ func WithUpdateCallback(cb UpdateCallback) ProcessorOption

+ type ProcessorStats struct

+ Group map[int32]*PartitionStats

+ Joined map[int32]map[string]*PartitionStats

+ Lookup map[string]*ViewStats

+ type Stream string

+ type Streams []Stream

+ type Table string

+ func GroupTable(group Group) Table

+ type Tester interface

+ ConsumerBuilder func() kafka.ConsumerBuilder

+ ProducerBuilder func() kafka.ProducerBuilder

+ StorageBuilder func() storage.Builder

+ TopicManagerBuilder func() kafka.TopicManagerBuilder

+ type UpdateCallback func(s storage.Storage, partition int32, key string, value []byte) error

+ type View struct

+ func NewView(brokers []string, topic Table, codec Codec, options ...ViewOption) (*View, error)

+ func (v *View) Evict(key string) error

+ func (v *View) Get(key string) (interface{}, error)

+ func (v *View) Has(key string) (bool, error)

+ func (v *View) Iterator() (Iterator, error)

+ func (v *View) IteratorWithRange(start, limit string) (Iterator, error)

+ func (v *View) Recovered() bool

+ func (v *View) Run(ctx context.Context) error

+ func (v *View) Stats() *ViewStats

+ func (v *View) Terminate() error

+ func (v *View) Topic() string

+ type ViewOption func(*voptions)

+ func WithViewCallback(cb UpdateCallback) ViewOption

+ func WithViewClientID(clientID string) ViewOption

+ func WithViewConsumerBuilder(cb kafka.ConsumerBuilder) ViewOption

+ func WithViewHasher(hasher func() hash.Hash32) ViewOption

+ func WithViewLogger(log logger.Logger) ViewOption

+ func WithViewPartitionChannelSize(size int) ViewOption

+ func WithViewRestartable() ViewOption

+ func WithViewStorageBuilder(sb storage.Builder) ViewOption

+ func WithViewTopicManagerBuilder(tmb kafka.TopicManagerBuilder) ViewOption

+ type ViewStats struct

+ Partitions map[int32]*PartitionStats

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