goka

package module

Go to main page

Versions in this module

v1

v1.1.12

May 16, 2024

Changes in this version

type DefaultUpdateContext

+ func (ctx DefaultUpdateContext) Timestamp() time.Time

type Processor

+ func (g *Processor) Done() <-chan struct{}

+ func (g *Processor) Error() error

type State

+ const ProcStateStopped

type UpdateContext

+ Timestamp func() time.Time

v1.1.11

Nov 1, 2023

v1.1.10

Sep 27, 2023

v1.1.9

Aug 10, 2023

Changes in this version

type MockClient

+ func (m *MockClient) LeaderAndEpoch(arg0 string, arg1 int32) (*sarama.Broker, int32, error)

type MockClientMockRecorder

+ func (mr *MockClientMockRecorder) LeaderAndEpoch(arg0, arg1 interface{}) *gomock.Call

v1.1.8

Apr 4, 2023

Changes in this version

+ type ContextWrapper func(ctx Context) Context

type MockClient

+ func (m *MockClient) LeastLoadedBroker() *sarama.Broker

+ func (m *MockClient) RefreshTransactionCoordinator(arg0 string) error

+ func (m *MockClient) TransactionCoordinator(arg0 string) (*sarama.Broker, error)

type MockClientMockRecorder

+ func (mr *MockClientMockRecorder) LeastLoadedBroker() *gomock.Call

+ func (mr *MockClientMockRecorder) RefreshTransactionCoordinator(arg0 interface{}) *gomock.Call

+ func (mr *MockClientMockRecorder) TransactionCoordinator(arg0 interface{}) *gomock.Call

type MockClusterAdmin

+ func (m *MockClusterAdmin) RemoveMemberFromConsumerGroup(arg0 string, arg1 []string) (*sarama.LeaveGroupResponse, error)

type MockClusterAdminMockRecorder

+ func (mr *MockClusterAdminMockRecorder) CreateACLs(arg0 interface{}) *gomock.Call

+ func (mr *MockClusterAdminMockRecorder) RemoveMemberFromConsumerGroup(arg0, arg1 interface{}) *gomock.Call

type Processor

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

type ProcessorOption

+ func WithContextWrapper(wrapper ContextWrapper) ProcessorOption

type Signal

+ func (s *Signal) WaitForStateMinWithCleanup(state State) (<-chan struct{}, func())

v1.1.7

Jul 12, 2022

v1.1.6

Jul 1, 2022

v1.1.5

Jun 29, 2022

Changes in this version

type MockClusterAdmin

+ func (m *MockClusterAdmin) CreateACLs(acls []*sarama.ResourceAcls) error

v1.1.4

Jun 28, 2022

v1.1.3

Jun 15, 2022

v1.1.2

Mar 4, 2022

Changes in this version

type MockAutoConsumer

+ func (c *MockAutoConsumer) Pause(topicPartitions map[string][]int32)

+ func (c *MockAutoConsumer) PauseAll()

+ func (c *MockAutoConsumer) Resume(topicPartitions map[string][]int32)

+ func (c *MockAutoConsumer) ResumeAll()

type MockAutoPartitionConsumer

+ func (pc *MockAutoPartitionConsumer) IsPaused() bool

+ func (pc *MockAutoPartitionConsumer) Pause()

+ func (pc *MockAutoPartitionConsumer) Resume()

type MockClient

+ func (m *MockClient) Broker(arg0 int32) (*sarama.Broker, error)

+ func (m *MockClient) RefreshBrokers(arg0 []string) error

type MockClientMockRecorder

+ func (mr *MockClientMockRecorder) Broker(arg0 interface{}) *gomock.Call

+ func (mr *MockClientMockRecorder) RefreshBrokers(arg0 interface{}) *gomock.Call

type MockClusterAdmin

+ func (m *MockClusterAdmin) AlterClientQuotas(arg0 []sarama.QuotaEntityComponent, arg1 sarama.ClientQuotasOp, arg2 bool) error

+ func (m *MockClusterAdmin) Controller() (*sarama.Broker, error)

+ func (m *MockClusterAdmin) DeleteConsumerGroupOffset(arg0, arg1 string, arg2 int32) error

+ func (m *MockClusterAdmin) DeleteUserScramCredentials(arg0 []sarama.AlterUserScramCredentialsDelete) ([]*sarama.AlterUserScramCredentialsResult, error)

+ func (m *MockClusterAdmin) DescribeClientQuotas(arg0 []sarama.QuotaFilterComponent, arg1 bool) ([]sarama.DescribeClientQuotasEntry, error)

+ func (m *MockClusterAdmin) DescribeUserScramCredentials(arg0 []string) ([]*sarama.DescribeUserScramCredentialsResult, error)

+ func (m *MockClusterAdmin) IncrementalAlterConfig(arg0 sarama.ConfigResourceType, arg1 string, ...) error

+ func (m *MockClusterAdmin) UpsertUserScramCredentials(arg0 []sarama.AlterUserScramCredentialsUpsert) ([]*sarama.AlterUserScramCredentialsResult, error)

type MockClusterAdminMockRecorder

+ func (mr *MockClusterAdminMockRecorder) AlterClientQuotas(arg0, arg1, arg2 interface{}) *gomock.Call

+ func (mr *MockClusterAdminMockRecorder) Controller() *gomock.Call

+ func (mr *MockClusterAdminMockRecorder) DeleteConsumerGroupOffset(arg0, arg1, arg2 interface{}) *gomock.Call

+ func (mr *MockClusterAdminMockRecorder) DeleteUserScramCredentials(arg0 interface{}) *gomock.Call

+ func (mr *MockClusterAdminMockRecorder) DescribeClientQuotas(arg0, arg1 interface{}) *gomock.Call

+ func (mr *MockClusterAdminMockRecorder) DescribeUserScramCredentials(arg0 interface{}) *gomock.Call

+ func (mr *MockClusterAdminMockRecorder) IncrementalAlterConfig(arg0, arg1, arg2, arg3 interface{}) *gomock.Call

+ func (mr *MockClusterAdminMockRecorder) UpsertUserScramCredentials(arg0 interface{}) *gomock.Call

type MockConsumerGroup

+ func (cg *MockConsumerGroup) Pause(partitions map[string][]int32)

+ func (cg *MockConsumerGroup) PauseAll()

+ func (cg *MockConsumerGroup) Resume(partitions map[string][]int32)

+ func (cg *MockConsumerGroup) ResumeAll()

v1.1.1

Dec 7, 2021

v1.1.0

Nov 18, 2021

Changes in this version

+ var ErrVisitAborted = errors.New("VisitAll aborted due to context cancel or rebalance")

+ func Debug(gokaDebug, saramaDebug bool)

+ func ResetSuffixes()

+ func SetLoopSuffix(suffix string)

+ func SetSaramaLogger(logger Logger)

+ func SetTableSuffix(suffix string)

+ type DefaultUpdateContext struct

+ func (ctx DefaultUpdateContext) Headers() Headers

+ func (ctx DefaultUpdateContext) Offset() int64

+ func (ctx DefaultUpdateContext) Partition() int32

+ func (ctx DefaultUpdateContext) Topic() Stream

type Edge

+ func Visitor(name string, cb ProcessCallback) Edge

type EmitterOption

+ func WithEmitterDefaultHeaders(headers Headers) EmitterOption

type GroupGraph

+ func (gg *GroupGraph) AllEdges() Edges

+ type Headers map[string][]byte

+ func HeadersFromSarama(saramaHeaders []*sarama.RecordHeader) Headers

+ func (h Headers) Merged(headersList ...Headers) Headers

+ func (h Headers) ToSarama() []sarama.RecordHeader

+ func (h Headers) ToSaramaPtr() []*sarama.RecordHeader

+ type Logger interface

+ Print func(...interface{})

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

+ Println func(...interface{})

+ func DefaultLogger() Logger

+ type PPRunMode int

type PartitionProcessor

+ func (pp *PartitionProcessor) Start(setupCtx, ctx context.Context) error

+ func (pp *PartitionProcessor) VisitValues(ctx context.Context, name string, meta interface{}, visited *int64) error

type Processor

+ func (g *Processor) StateReader() StateReader

+ func (g *Processor) VisitAll(ctx context.Context, name string, meta interface{}) error

+ func (g *Processor) VisitAllWithStats(ctx context.Context, name string, meta interface{}) (int64, error)

type ProcessorOption

+ func WithHotStandby() ProcessorOption

+ func WithProducerDefaultHeaders(hdr Headers) ProcessorOption

+ func WithRecoverAhead() ProcessorOption

type State

+ const PPStateStopped

+ type StateReader interface

+ IsState func(State) bool

+ ObserveStateChange func() *StateChangeObserver

+ State func() State

+ WaitForState func(state State) <-chan struct{}

+ WaitForStateMin func(state State) <-chan struct{}

type TableStats

+ RunMode PPRunMode

type TopicManagerConfig

+ CreateTopicTimeout time.Duration

+ type UpdateContext interface

+ Headers func() Headers

+ Offset func() int64

+ Partition func() int32

+ Topic func() Stream

v1.1.0-beta.4

Oct 27, 2021

v1.1.0-beta.3

Oct 25, 2021

v1.1.0-beta.2

Oct 19, 2021

v1.1.0-beta.1

Oct 3, 2021

v1.1.0-beta.0

Sep 13, 2021

v1.0.6

Feb 10, 2021

Changes in this version

+ type ContextOption func(*ctxOptions)

+ func WithCtxEmitHeaders(headers map[string][]byte) ContextOption

+ type MockClusterAdmin struct

+ func NewMockClusterAdmin(ctrl *gomock.Controller) *MockClusterAdmin

+ func (m *MockClusterAdmin) AlterConfig(arg0 sarama.ConfigResourceType, arg1 string, arg2 map[string]*string, ...) error

+ func (m *MockClusterAdmin) AlterPartitionReassignments(arg0 string, arg1 [][]int32) error

+ func (m *MockClusterAdmin) Close() error

+ func (m *MockClusterAdmin) CreateACL(arg0 sarama.Resource, arg1 sarama.Acl) error

+ func (m *MockClusterAdmin) CreatePartitions(arg0 string, arg1 int32, arg2 [][]int32, arg3 bool) error

+ func (m *MockClusterAdmin) CreateTopic(arg0 string, arg1 *sarama.TopicDetail, arg2 bool) error

+ func (m *MockClusterAdmin) DeleteACL(arg0 sarama.AclFilter, arg1 bool) ([]sarama.MatchingAcl, error)

+ func (m *MockClusterAdmin) DeleteConsumerGroup(arg0 string) error

+ func (m *MockClusterAdmin) DeleteRecords(arg0 string, arg1 map[int32]int64) error

+ func (m *MockClusterAdmin) DeleteTopic(arg0 string) error

+ func (m *MockClusterAdmin) DescribeCluster() ([]*sarama.Broker, int32, error)

+ func (m *MockClusterAdmin) DescribeConfig(arg0 sarama.ConfigResource) ([]sarama.ConfigEntry, error)

+ func (m *MockClusterAdmin) DescribeConsumerGroups(arg0 []string) ([]*sarama.GroupDescription, error)

+ func (m *MockClusterAdmin) DescribeLogDirs(arg0 []int32) (map[int32][]sarama.DescribeLogDirsResponseDirMetadata, error)

+ func (m *MockClusterAdmin) DescribeTopics(arg0 []string) ([]*sarama.TopicMetadata, error)

+ func (m *MockClusterAdmin) EXPECT() *MockClusterAdminMockRecorder

+ func (m *MockClusterAdmin) ListAcls(arg0 sarama.AclFilter) ([]sarama.ResourceAcls, error)

+ func (m *MockClusterAdmin) ListConsumerGroupOffsets(arg0 string, arg1 map[string][]int32) (*sarama.OffsetFetchResponse, error)

+ func (m *MockClusterAdmin) ListConsumerGroups() (map[string]string, error)

+ func (m *MockClusterAdmin) ListPartitionReassignments(arg0 string, arg1 []int32) (map[string]map[int32]*sarama.PartitionReplicaReassignmentsStatus, error)

+ func (m *MockClusterAdmin) ListTopics() (map[string]sarama.TopicDetail, error)

+ type MockClusterAdminMockRecorder struct

+ func (mr *MockClusterAdminMockRecorder) AlterConfig(arg0, arg1, arg2, arg3 interface{}) *gomock.Call

+ func (mr *MockClusterAdminMockRecorder) AlterPartitionReassignments(arg0, arg1 interface{}) *gomock.Call

+ func (mr *MockClusterAdminMockRecorder) Close() *gomock.Call

+ func (mr *MockClusterAdminMockRecorder) CreateACL(arg0, arg1 interface{}) *gomock.Call

+ func (mr *MockClusterAdminMockRecorder) CreatePartitions(arg0, arg1, arg2, arg3 interface{}) *gomock.Call

+ func (mr *MockClusterAdminMockRecorder) CreateTopic(arg0, arg1, arg2 interface{}) *gomock.Call

+ func (mr *MockClusterAdminMockRecorder) DeleteACL(arg0, arg1 interface{}) *gomock.Call

+ func (mr *MockClusterAdminMockRecorder) DeleteConsumerGroup(arg0 interface{}) *gomock.Call

+ func (mr *MockClusterAdminMockRecorder) DeleteRecords(arg0, arg1 interface{}) *gomock.Call

+ func (mr *MockClusterAdminMockRecorder) DeleteTopic(arg0 interface{}) *gomock.Call

+ func (mr *MockClusterAdminMockRecorder) DescribeCluster() *gomock.Call

+ func (mr *MockClusterAdminMockRecorder) DescribeConfig(arg0 interface{}) *gomock.Call

+ func (mr *MockClusterAdminMockRecorder) DescribeConsumerGroups(arg0 interface{}) *gomock.Call

+ func (mr *MockClusterAdminMockRecorder) DescribeLogDirs(arg0 interface{}) *gomock.Call

+ func (mr *MockClusterAdminMockRecorder) DescribeTopics(arg0 interface{}) *gomock.Call

+ func (mr *MockClusterAdminMockRecorder) ListAcls(arg0 interface{}) *gomock.Call

+ func (mr *MockClusterAdminMockRecorder) ListConsumerGroupOffsets(arg0, arg1 interface{}) *gomock.Call

+ func (mr *MockClusterAdminMockRecorder) ListConsumerGroups() *gomock.Call

+ func (mr *MockClusterAdminMockRecorder) ListPartitionReassignments(arg0, arg1 interface{}) *gomock.Call

+ func (mr *MockClusterAdminMockRecorder) ListTopics() *gomock.Call

+ type TMConfigMismatchBehavior int

+ const TMConfigMismatchBehaviorFail

+ const TMConfigMismatchBehaviorIgnore

+ const TMConfigMismatchBehaviorWarn

type TopicManagerConfig

+ Logger logger.Logger

+ MismatchBehavior TMConfigMismatchBehavior

v1.0.5

Dec 7, 2020

Changes in this version

+ var StrictCopartitioningStrategy = &copartitioningStrategy

+ func NewPromiseWithFinisher() (*Promise, PromiseFinisher)

type Context

+ DeferCommit func() func(error)

+ Group func() Group

type PartitionTable

+ func (p *PartitionTable) Iterator() (storage.Iterator, error)

+ func (p *PartitionTable) IteratorWithRange(start []byte, limit []byte) (storage.Iterator, error)

+ type PromiseFinisher func(msg *sarama.ProducerMessage, err error) *Promise

v1.0.4

Oct 16, 2020

v1.0.3

Sep 29, 2020

Changes in this version

type Emitter

+ func (e *Emitter) EmitSyncWithHeaders(key string, msg interface{}, headers map[string][]byte) error

v1.0.2

Sep 4, 2020

v1.0.1

Aug 18, 2020

Changes in this version

type Emitter

+ func (e *Emitter) EmitWithHeaders(key string, msg interface{}, headers map[string][]byte) (*Promise, error)

type MockConsumerGroupSession

+ func (cgs *MockConsumerGroupSession) Commit()

type MockProducer

+ func (m *MockProducer) EmitWithHeaders(arg0, arg1 string, arg2 []byte, arg3 map[string][]byte) *Promise

type MockProducerMockRecorder

+ func (mr *MockProducerMockRecorder) EmitWithHeaders(arg0, arg1, arg2, arg3 interface{}) *gomock.Call

type Producer

+ EmitWithHeaders func(topic string, key string, value []byte, headers map[string][]byte) *Promise

v1.0.0

Jul 9, 2020

Changes in this version

+ var CopartitioningStrategy = new(copartitioningStrategy)

+ var ErrEmitterAlreadyClosed error = errors.New("emitter already closed")

+ func DefaultConfig() *sarama.Config

+ func DefaultConsumerGroupBuilder(brokers []string, group, clientID string) (sarama.ConsumerGroup, error)

+ func DefaultSaramaConsumerBuilder(brokers []string, clientID string) (sarama.Consumer, error)

+ func NewMockController(t gomock.TestReporter) *gomock.Controller

+ func ReplaceGlobalConfig(config *sarama.Config)

+ type Assignment map[int32]int64

+ type Backoff interface

+ Duration func() time.Duration

+ Reset func()

+ func DefaultBackoffBuilder() (Backoff, error)

+ func NewSimpleBackoff(step time.Duration) Backoff

+ type BackoffBuilder func() (Backoff, error)

+ type Broker interface

+ Addr func() string

+ Connected func() (bool, error)

+ CreateTopics func(request *sarama.CreateTopicsRequest) (*sarama.CreateTopicsResponse, error)

+ Open func(conf *sarama.Config) error

+ type ConsumerGroupBuilder func(brokers []string, group, clientID string) (sarama.ConsumerGroup, error)

+ func ConsumerGroupBuilderWithConfig(config *sarama.Config) ConsumerGroupBuilder

type InputStats

+ LastOffset int64

+ OffsetLag int64

+ type MockAutoConsumer struct

+ func NewMockAutoConsumer(t *testing.T, config *sarama.Config) *MockAutoConsumer

+ func (c *MockAutoConsumer) Close() error

+ func (c *MockAutoConsumer) ConsumePartition(topic string, partition int32, offset int64) (sarama.PartitionConsumer, error)

+ func (c *MockAutoConsumer) ExpectConsumePartition(topic string, partition int32, offset int64) *MockAutoPartitionConsumer

+ func (c *MockAutoConsumer) HighWaterMarks() map[string]map[int32]int64

+ func (c *MockAutoConsumer) Partitions(topic string) ([]int32, error)

+ func (c *MockAutoConsumer) SetTopicMetadata(metadata map[string][]int32)

+ func (c *MockAutoConsumer) Topics() ([]string, error)

+ type MockAutoPartitionConsumer struct

+ func (pc *MockAutoPartitionConsumer) AsyncClose()

+ func (pc *MockAutoPartitionConsumer) Close() error

+ func (pc *MockAutoPartitionConsumer) Errors() <-chan *sarama.ConsumerError

+ func (pc *MockAutoPartitionConsumer) ExpectErrorsDrainedOnClose()

+ func (pc *MockAutoPartitionConsumer) ExpectMessagesDrainedOnClose()

+ func (pc *MockAutoPartitionConsumer) HighWaterMarkOffset() int64

+ func (pc *MockAutoPartitionConsumer) Messages() <-chan *sarama.ConsumerMessage

+ func (pc *MockAutoPartitionConsumer) YieldError(err error)

+ func (pc *MockAutoPartitionConsumer) YieldMessage(msg *sarama.ConsumerMessage)

+ type MockBroker struct

+ func NewMockBroker(ctrl *gomock.Controller) *MockBroker

+ func (m *MockBroker) Addr() string

+ func (m *MockBroker) Connected() (bool, error)

+ func (m *MockBroker) CreateTopics(arg0 *sarama.CreateTopicsRequest) (*sarama.CreateTopicsResponse, error)

+ func (m *MockBroker) EXPECT() *MockBrokerMockRecorder

+ func (m *MockBroker) Open(arg0 *sarama.Config) error

+ type MockBrokerMockRecorder struct

+ func (mr *MockBrokerMockRecorder) Addr() *gomock.Call

+ func (mr *MockBrokerMockRecorder) Connected() *gomock.Call

+ func (mr *MockBrokerMockRecorder) CreateTopics(arg0 interface{}) *gomock.Call

+ func (mr *MockBrokerMockRecorder) Open(arg0 interface{}) *gomock.Call

+ type MockClient struct

+ func NewMockClient(ctrl *gomock.Controller) *MockClient

+ func (m *MockClient) Brokers() []*sarama.Broker

+ func (m *MockClient) Close() error

+ func (m *MockClient) Closed() bool

+ func (m *MockClient) Config() *sarama.Config

+ func (m *MockClient) Controller() (*sarama.Broker, error)

+ func (m *MockClient) Coordinator(arg0 string) (*sarama.Broker, error)

+ func (m *MockClient) EXPECT() *MockClientMockRecorder

+ func (m *MockClient) GetOffset(arg0 string, arg1 int32, arg2 int64) (int64, error)

+ func (m *MockClient) InSyncReplicas(arg0 string, arg1 int32) ([]int32, error)

+ func (m *MockClient) InitProducerID() (*sarama.InitProducerIDResponse, error)

+ func (m *MockClient) Leader(arg0 string, arg1 int32) (*sarama.Broker, error)

+ func (m *MockClient) OfflineReplicas(arg0 string, arg1 int32) ([]int32, error)

+ func (m *MockClient) Partitions(arg0 string) ([]int32, error)

+ func (m *MockClient) RefreshController() (*sarama.Broker, error)

+ func (m *MockClient) RefreshCoordinator(arg0 string) error

+ func (m *MockClient) RefreshMetadata(arg0 ...string) error

+ func (m *MockClient) Replicas(arg0 string, arg1 int32) ([]int32, error)

+ func (m *MockClient) Topics() ([]string, error)

+ func (m *MockClient) WritablePartitions(arg0 string) ([]int32, error)

+ type MockClientMockRecorder struct

+ func (mr *MockClientMockRecorder) Brokers() *gomock.Call

+ func (mr *MockClientMockRecorder) Close() *gomock.Call

+ func (mr *MockClientMockRecorder) Closed() *gomock.Call

+ func (mr *MockClientMockRecorder) Config() *gomock.Call

+ func (mr *MockClientMockRecorder) Controller() *gomock.Call

+ func (mr *MockClientMockRecorder) Coordinator(arg0 interface{}) *gomock.Call

+ func (mr *MockClientMockRecorder) GetOffset(arg0, arg1, arg2 interface{}) *gomock.Call

+ func (mr *MockClientMockRecorder) InSyncReplicas(arg0, arg1 interface{}) *gomock.Call

+ func (mr *MockClientMockRecorder) InitProducerID() *gomock.Call

+ func (mr *MockClientMockRecorder) Leader(arg0, arg1 interface{}) *gomock.Call

+ func (mr *MockClientMockRecorder) OfflineReplicas(arg0, arg1 interface{}) *gomock.Call

+ func (mr *MockClientMockRecorder) Partitions(arg0 interface{}) *gomock.Call

+ func (mr *MockClientMockRecorder) RefreshController() *gomock.Call

+ func (mr *MockClientMockRecorder) RefreshCoordinator(arg0 interface{}) *gomock.Call

+ func (mr *MockClientMockRecorder) RefreshMetadata(arg0 ...interface{}) *gomock.Call

+ func (mr *MockClientMockRecorder) Replicas(arg0, arg1 interface{}) *gomock.Call

+ func (mr *MockClientMockRecorder) Topics() *gomock.Call

+ func (mr *MockClientMockRecorder) WritablePartitions(arg0 interface{}) *gomock.Call

+ type MockConsumerGroup struct

+ func NewMockConsumerGroup(t *testing.T) *MockConsumerGroup

+ func (cg *MockConsumerGroup) Close() error

+ func (cg *MockConsumerGroup) Consume(ctx context.Context, topics []string, handler sarama.ConsumerGroupHandler) error

+ func (cg *MockConsumerGroup) Errors() <-chan error

+ func (cg *MockConsumerGroup) FailOnConsume(err error)

+ func (cg *MockConsumerGroup) SendError(err error)

+ func (cg *MockConsumerGroup) SendMessage(message *sarama.ConsumerMessage) <-chan struct{}

+ func (cg *MockConsumerGroup) SendMessageWait(message *sarama.ConsumerMessage)

+ type MockConsumerGroupClaim struct

+ func NewMockConsumerGroupClaim(topic string, partition int32) *MockConsumerGroupClaim

+ func (cgc *MockConsumerGroupClaim) HighWaterMarkOffset() int64

+ func (cgc *MockConsumerGroupClaim) InitialOffset() int64

+ func (cgc *MockConsumerGroupClaim) Messages() <-chan *sarama.ConsumerMessage

+ func (cgc *MockConsumerGroupClaim) Partition() int32

+ func (cgc *MockConsumerGroupClaim) Topic() string

+ type MockConsumerGroupSession struct

+ func (cgs *MockConsumerGroupSession) Claims() map[string][]int32

+ func (cgs *MockConsumerGroupSession) Context() context.Context

+ func (cgs *MockConsumerGroupSession) GenerationID() int32

+ func (cgs *MockConsumerGroupSession) MarkMessage(msg *sarama.ConsumerMessage, metadata string)

+ func (cgs *MockConsumerGroupSession) MarkOffset(topic string, partition int32, offset int64, metadata string)

+ func (cgs *MockConsumerGroupSession) MemberID() string

+ func (cgs *MockConsumerGroupSession) ResetOffset(topic string, partition int32, offset int64, metadata string)

+ func (cgs *MockConsumerGroupSession) SendMessage(msg *sarama.ConsumerMessage)

+ type MockProducer struct

+ func NewMockProducer(ctrl *gomock.Controller) *MockProducer

+ func (m *MockProducer) Close() error

+ func (m *MockProducer) EXPECT() *MockProducerMockRecorder

+ func (m *MockProducer) Emit(arg0, arg1 string, arg2 []byte) *Promise

+ type MockProducerMockRecorder struct

+ func (mr *MockProducerMockRecorder) Close() *gomock.Call

+ func (mr *MockProducerMockRecorder) Emit(arg0, arg1, arg2 interface{}) *gomock.Call

+ type MockStorage struct

+ func NewMockStorage(ctrl *gomock.Controller) *MockStorage

+ func (m *MockStorage) Close() error

+ func (m *MockStorage) Delete(arg0 string) error

+ func (m *MockStorage) EXPECT() *MockStorageMockRecorder

+ func (m *MockStorage) Get(arg0 string) ([]byte, error)

+ func (m *MockStorage) GetOffset(arg0 int64) (int64, error)

+ func (m *MockStorage) Has(arg0 string) (bool, error)

+ func (m *MockStorage) Iterator() (storage.Iterator, error)

+ func (m *MockStorage) IteratorWithRange(arg0, arg1 []byte) (storage.Iterator, error)

+ func (m *MockStorage) MarkRecovered() error

+ func (m *MockStorage) Open() error

+ func (m *MockStorage) Recovered() bool

+ func (m *MockStorage) Set(arg0 string, arg1 []byte) error

+ func (m *MockStorage) SetOffset(arg0 int64) error

+ type MockStorageMockRecorder struct

+ func (mr *MockStorageMockRecorder) Close() *gomock.Call

+ func (mr *MockStorageMockRecorder) Delete(arg0 interface{}) *gomock.Call

+ func (mr *MockStorageMockRecorder) Get(arg0 interface{}) *gomock.Call

+ func (mr *MockStorageMockRecorder) GetOffset(arg0 interface{}) *gomock.Call

+ func (mr *MockStorageMockRecorder) Has(arg0 interface{}) *gomock.Call

+ func (mr *MockStorageMockRecorder) Iterator() *gomock.Call

+ func (mr *MockStorageMockRecorder) IteratorWithRange(arg0, arg1 interface{}) *gomock.Call

+ func (mr *MockStorageMockRecorder) MarkRecovered() *gomock.Call

+ func (mr *MockStorageMockRecorder) Open() *gomock.Call

+ func (mr *MockStorageMockRecorder) Recovered() *gomock.Call

+ func (mr *MockStorageMockRecorder) Set(arg0, arg1 interface{}) *gomock.Call

+ func (mr *MockStorageMockRecorder) SetOffset(arg0 interface{}) *gomock.Call

+ type MockTopicManager struct

+ func NewMockTopicManager(ctrl *gomock.Controller) *MockTopicManager

+ func (m *MockTopicManager) Close() error

+ func (m *MockTopicManager) EXPECT() *MockTopicManagerMockRecorder

+ func (m *MockTopicManager) EnsureStreamExists(arg0 string, arg1 int) error

+ func (m *MockTopicManager) EnsureTableExists(arg0 string, arg1 int) error

+ func (m *MockTopicManager) EnsureTopicExists(arg0 string, arg1, arg2 int, arg3 map[string]string) error

+ func (m *MockTopicManager) GetOffset(arg0 string, arg1 int32, arg2 int64) (int64, error)

+ func (m *MockTopicManager) Partitions(arg0 string) ([]int32, error)

+ type MockTopicManagerMockRecorder struct

+ func (mr *MockTopicManagerMockRecorder) Close() *gomock.Call

+ func (mr *MockTopicManagerMockRecorder) EnsureStreamExists(arg0, arg1 interface{}) *gomock.Call

+ func (mr *MockTopicManagerMockRecorder) EnsureTableExists(arg0, arg1 interface{}) *gomock.Call

+ func (mr *MockTopicManagerMockRecorder) EnsureTopicExists(arg0, arg1, arg2, arg3 interface{}) *gomock.Call

+ func (mr *MockTopicManagerMockRecorder) GetOffset(arg0, arg1, arg2 interface{}) *gomock.Call

+ func (mr *MockTopicManagerMockRecorder) Partitions(arg0 interface{}) *gomock.Call

+ type PartitionProcStats struct

+ Input map[string]*InputStats

+ Joined map[string]*TableStats

+ Now time.Time

+ Output map[string]*OutputStats

+ TableStats *TableStats

+ type PartitionProcessor struct

+ func (pp *PartitionProcessor) EnqueueMessage(msg *sarama.ConsumerMessage)

+ func (pp *PartitionProcessor) Errors() <-chan error

+ func (pp *PartitionProcessor) Recovered() bool

+ func (pp *PartitionProcessor) Setup(ctx context.Context) error

+ func (pp *PartitionProcessor) Stop() error

type PartitionStatus

+ const PartitionConnecting

+ const PartitionInitializing

+ const PartitionStopped

+ type PartitionTable struct

+ func (p *PartitionTable) CatchupForever(ctx context.Context, restartOnError bool) error

+ func (p *PartitionTable) Close() error

+ func (p *PartitionTable) CurrentState() PartitionStatus

+ func (p *PartitionTable) Delete(key string) error

+ func (p *PartitionTable) Get(key string) ([]byte, error)

+ func (p *PartitionTable) GetOffset(defValue int64) (int64, error)

+ func (p *PartitionTable) Has(key string) (bool, error)

+ func (p *PartitionTable) IsRecovered() bool

+ func (p *PartitionTable) RunStatsLoop(ctx context.Context)

+ func (p *PartitionTable) Set(key string, value []byte) error

+ func (p *PartitionTable) SetOffset(value int64) error

+ func (p *PartitionTable) SetupAndRecover(ctx context.Context, restartOnError bool) error

+ func (p *PartitionTable) TrackMessageWrite(ctx context.Context, length int)

+ func (p *PartitionTable) WaitRecovered() chan struct{}

type Processor

+ func (g *Processor) Cleanup(session sarama.ConsumerGroupSession) error

+ func (g *Processor) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

+ func (g *Processor) Setup(session sarama.ConsumerGroupSession) error

+ func (g *Processor) StatsWithContext(ctx context.Context) *ProcessorStats

+ func (g *Processor) Stop()

+ func (g *Processor) WaitForReady()

type ProcessorOption

+ func WithBackoffBuilder(bb BackoffBuilder) ProcessorOption

+ func WithBackoffResetTimeout(duration time.Duration) ProcessorOption

+ func WithConsumerGroupBuilder(cgb ConsumerGroupBuilder) ProcessorOption

+ func WithConsumerSaramaBuilder(cgb SaramaConsumerBuilder) ProcessorOption

+ type Producer interface

+ Close func() error

+ Emit func(topic string, key string, value []byte) *Promise

+ func DefaultProducerBuilder(brokers []string, clientID string, hasher func() hash.Hash32) (Producer, error)

+ func NewProducer(brokers []string, config *sarama.Config) (Producer, error)

+ type ProducerBuilder func(brokers []string, clientID string, hasher func() hash.Hash32) (Producer, error)

+ func ProducerBuilderWithConfig(config *sarama.Config) ProducerBuilder

+ type Promise struct

+ func NewPromise() *Promise

+ func (p *Promise) Finish(msg *sarama.ProducerMessage, err error) *Promise

+ func (p *Promise) Then(callback func(err error)) *Promise

+ func (p *Promise) ThenWithMessage(callback func(msg *sarama.ProducerMessage, err error)) *Promise

+ type RecoveryStats struct

+ Hwm int64

+ Offset int64

+ RecoveryTime time.Time

+ StartTime time.Time

+ type SaramaConsumerBuilder func(brokers []string, clientID string) (sarama.Consumer, error)

+ func SaramaConsumerBuilderWithConfig(config *sarama.Config) SaramaConsumerBuilder

+ type Signal struct

+ func NewSignal(states ...State) *Signal

+ func (s *Signal) IsState(state State) bool

+ func (s *Signal) ObserveStateChange() *StateChangeObserver

+ func (s *Signal) SetState(state State) *Signal

+ func (s *Signal) State() State

+ func (s *Signal) WaitForState(state State) chan struct{}

+ func (s *Signal) WaitForStateMin(state State) chan struct{}

+ type State int

+ const PPStateIdle

+ const PPStateRecovering

+ const PPStateRunning

+ const PPStateStopping

+ const ProcStateIdle

+ const ProcStateRunning

+ const ProcStateSetup

+ const ProcStateStarting

+ const ProcStateStopping

+ type StateChangeObserver struct

+ func (s *StateChangeObserver) C() <-chan State

+ func (s *StateChangeObserver) Stop()

type Streams

+ func StringsToStreams(strings ...string) Streams

+ type TableStats struct

+ Input *InputStats

+ Recovery *RecoveryStats

+ Stalled bool

+ Status PartitionStatus

+ Writes *OutputStats

type Tester

+ ConsumerGroupBuilder func() ConsumerGroupBuilder

+ type TopicManager interface

+ Close func() error

+ EnsureStreamExists func(topic string, npar int) error

+ EnsureTableExists func(topic string, npar int) error

+ EnsureTopicExists func(topic string, npar, rfactor int, config map[string]string) error

+ GetOffset func(topic string, partitionID int32, time int64) (int64, error)

+ Partitions func(topic string) ([]int32, error)

+ func DefaultTopicManagerBuilder(brokers []string) (TopicManager, error)

+ func NewTopicManager(brokers []string, saramaConfig *sarama.Config, ...) (TopicManager, error)

+ type TopicManagerBuilder func(brokers []string) (TopicManager, error)

+ func TopicManagerBuilderWithConfig(config *sarama.Config, tmConfig *TopicManagerConfig) TopicManagerBuilder

+ func TopicManagerBuilderWithTopicManagerConfig(tmConfig *TopicManagerConfig) TopicManagerBuilder

+ type TopicManagerConfig struct

+ Stream struct{ ... }

+ Table struct{ ... }

+ func NewTopicManagerConfig() *TopicManagerConfig

type View

+ func (v *View) CurrentState() ViewState

+ func (v *View) ObserveStateChanges() *StateChangeObserver

+ func (v *View) WaitRunning() <-chan struct{}

type ViewOption

+ func WithViewAutoReconnect() ViewOption

+ func WithViewBackoffBuilder(bb BackoffBuilder) ViewOption

+ func WithViewBackoffResetTimeout(duration time.Duration) ViewOption

+ func WithViewConsumerSaramaBuilder(cgb SaramaConsumerBuilder) ViewOption

+ type ViewState int

+ const ViewStateCatchUp

+ const ViewStateConnecting

+ const ViewStateIdle

+ const ViewStateInitializing

+ const ViewStateRunning

v0

v0.9.0-beta3

Apr 9, 2020

v0.9.0-beta2

Mar 20, 2020

v0.9.0-beta1

Mar 19, 2020

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