consumergroup

package module

Go to main page

Versions in this module

v0

v0.2.7

Apr 1, 2020

v0.2.6

Feb 18, 2020

v0.2.5

May 31, 2019

v0.2.4

May 31, 2019

v0.2.3

Nov 6, 2018

Changes in this version

type ConsumerGroup

+ func (cg *ConsumerGroup) GetOffsets() map[string]interface{}

+ func (cg *ConsumerGroup) OnClose(cb func())

+ func (cg *ConsumerGroup) OnLoad(cb func())

+ func (cg *ConsumerGroup) Owners() map[string]map[int32]string

v0.2.2

Jul 10, 2018

Changes in this version

type Config

+ Chroot string

+ ConsumerID string

type ConsumerGroup

+ func (cg *ConsumerGroup) Start() error

+ func (cg *ConsumerGroup) Stop()

v0.2.0

Jul 26, 2017

Changes in this version

type Config

+ ClaimPartitionRetryTimes int

+ GroupID string

+ TopicList []string

+ ZkList []string

+ ZkSessionTimeout time.Duration

type ConsumerGroup

+ func (cg *ConsumerGroup) GetErrors(topic string) (<-chan *sarama.ConsumerError, bool)

+ func (cg *ConsumerGroup) GetMessages(topic string) (<-chan *sarama.ConsumerMessage, bool)

v0.1.0

Jun 27, 2017

Changes in this version

+ const BROKERS_DIR

+ const BROKERS_PATH

+ const CG_INIT

+ const CG_START

+ const CG_STOPPED

+ const CONSUMERS_DIR

+ const CONSUMERS_PATH

+ const DEBUG

+ const ERROR

+ const INFO

+ const OFFSETS_PATH

+ const OWNER_PATH

+ const WARN

+ var ErrInvalidConsumerID = errors.New("Invalid consumer ID")

+ var ErrInvalidGroup = errors.New("Invalid group")

+ var ErrInvalidPartition = "Invalid partition %s"

+ var ErrInvalidTopic = errors.New("Invalid topic")

+ func GenConsumerID() string

+ func SliceRemoveDuplicates(slice []string) []string

+ func ZKCreateEphemeralPath(c *zk.Conn, zkPath string, data []byte) error

+ func ZKCreatePersistentPath(c *zk.Conn, zkPath string, data []byte) error

+ func ZKCreateRecursive(c *zk.Conn, zkPath string, flags int32, data []byte) error

+ func ZKSetPersistentPath(c *zk.Conn, zkPath string, data []byte) error

+ type Config struct

+ ClaimPartitionRetry int

+ ClaimPartitionRetryInterval time.Duration

+ ErrorChannelBufferSize int

+ OffsetAutoCommitEnable bool

+ OffsetAutoCommitInterval time.Duration

+ OffsetAutoReset int64

+ SaramaConfig *sarama.Config

+ func NewConfig(groupID string, topicList []string) (*Config, error)

+ type ConsumerGroup struct

+ func NewConsumerGroup(storage GroupStorage, config *Config) (*ConsumerGroup, error)

+ func (cg *ConsumerGroup) CommitOffset(topic string, partition int32, offset int64) error

+ func (cg *ConsumerGroup) ExitGroup()

+ func (cg *ConsumerGroup) GetTopicErrorsChannel(topic string) (<-chan *sarama.ConsumerError, error)

+ func (cg *ConsumerGroup) GetTopicNextMessageChannel(topic string) (<-chan *sarama.ConsumerMessage, error)

+ func (cg *ConsumerGroup) IsStopped() bool

+ func (cg *ConsumerGroup) JoinGroup() error

+ func (cg *ConsumerGroup) SetLogger(logger Logger)

+ type GroupStorage interface

+ ClaimPartition func(group, topic string, partition int32, consumerID string) error

+ CommitOffset func(group, topic string, partition int32, offset int64) error

+ DeleteConsumer func(group, consumerID string) error

+ ExistsConsumer func(group, consumerID string) (bool, error)

+ GetBrokerList func() ([]string, error)

+ GetConsumerList func(group string) ([]string, error)

+ GetOffset func(group, topic string, partition int32) (int64, error)

+ GetPartitionOwner func(group, topic string, partition int32) (string, error)

+ RegisterConsumer func(group, consumerID string, data []byte) error

+ ReleasePartition func(group, topic string, partition int32) error

+ WatchConsumerList func(group string) (<-chan zk.Event, error)

+ type InnerLogger struct

+ func NewInnerLog(level int) *InnerLogger

+ func (logger *InnerLogger) Debug(args ...interface{})

+ func (logger *InnerLogger) Debugf(format string, args ...interface{})

+ func (logger *InnerLogger) Error(args ...interface{})

+ func (logger *InnerLogger) Errorf(format string, args ...interface{})

+ func (logger *InnerLogger) Info(args ...interface{})

+ func (logger *InnerLogger) Infof(format string, args ...interface{})

+ func (logger *InnerLogger) Warn(args ...interface{})

+ func (logger *InnerLogger) Warnf(format string, args ...interface{})

+ type Logger interface

+ Debug func(args ...interface{})

+ Debugf func(format string, args ...interface{})

+ Error func(args ...interface{})

+ Errorf func(format string, args ...interface{})

+ Info func(args ...interface{})

+ Infof func(format string, args ...interface{})

+ Warn func(args ...interface{})

+ Warnf func(format string, args ...interface{})

+ type ZKGroupStorage struct

+ func NewZKGroupStorage(serverList []string, sessionTimeout time.Duration) (*ZKGroupStorage, error)

+ func (s *ZKGroupStorage) ClaimPartition(group, topic string, partition int32, consumerID string) error

+ func (s *ZKGroupStorage) CommitOffset(group, topic string, partition int32, offset int64) error

+ func (s *ZKGroupStorage) DeleteConsumer(group, consumerID string) error

+ func (s *ZKGroupStorage) ExistsConsumer(group, consumerID string) (bool, error)

+ func (s *ZKGroupStorage) GetBrokerList() ([]string, error)

+ func (s *ZKGroupStorage) GetClient() (*zk.Conn, error)

+ func (s *ZKGroupStorage) GetConsumerList(group string) ([]string, error)

+ func (s *ZKGroupStorage) GetOffset(group, topic string, partition int32) (int64, error)

+ func (s *ZKGroupStorage) GetPartitionOwner(group, topic string, partition int32) (string, error)

+ func (s *ZKGroupStorage) RegisterConsumer(group, consumerID string, data []byte) error

+ func (s *ZKGroupStorage) ReleasePartition(group, topic string, partition int32) error

+ func (s *ZKGroupStorage) WatchConsumerList(group string) (<-chan zk.Event, error)

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