kafka

Sep 14, 2022

Changes in this version

type IProducer

+ AddHeader func(key, value string)

Sep 13, 2022

Changes in this version

+ type OptFunc func()

Sep 2, 2022

Changes in this version

+ const TraceIDHeaderKey

type Message

+ Context func() context.Context

+ GetHeader func(key string) string

type Producer

+ func (p *Producer) AddHeader(key, value string)

type SaramaMessage

+ func (M SaramaMessage) Context() context.Context

+ func (M SaramaMessage) GetHeader(key string) string

Dec 8, 2021

Changes in this version

+ const ConsumerMinBrokersHealthy

+ const ProducerMinBrokersHealthy

+ type HealthInfo struct

+ HasTopic bool

+ Reachable bool

+ type HealthInfoMap map[*sarama.Broker]HealthInfo

+ func (h *HealthInfoMap) ErrorMsg() string

+ func (h *HealthInfoMap) UpdateStatus(state *health.CheckState, minHealthyThreshold int, msgHealthy string) error

Aug 10, 2021

Changes in this version

+ var MaxRetryInterval = 31 * time.Second

+ func SetMaxRetryInterval(maxPause time.Duration)

Jul 1, 2021

Changes in this version

+ func GetPrincipal(app, subnet, domain string) string

+ type Acls []*sarama.AclCreation

+ func (t Acls) Apply(adm sarama.ClusterAdmin) error

+ type AdminConfig struct

+ KafkaVersion *string

+ KeepAlive *time.Duration

+ RetryBackoff *time.Duration

+ RetryMax *int

+ SecurityConfig *SecurityConfig

type SecurityConfig

+ func GetSecurityConfig(caCerts, clientCert, clientKey string, skipVerify bool) *SecurityConfig

+ type TopicAuth struct

+ App string

+ Hosts []string

+ Operations []sarama.AclOperation

+ Subnets []string

+ Topic string

+ func (t TopicAuth) GetAcls(domain string) Acls

+ type TopicAuthList struct

+ Acls []TopicAuth

+ Brokers []string

+ Domain string

+ func (t TopicAuthList) Apply(adm sarama.ClusterAdmin) error

May 28, 2021

Changes in this version

+ func NewAdmin(brokerAddrs []string, pConfig *ProducerConfig) (sarama.ClusterAdmin, error)

May 13, 2021

Changes in this version

+ var ErrTLSCannotLoadCACerts = errors.New("Cannot load CA Certs")

type ConsumerGroupConfig

+ SecurityConfig SecurityConfig

type ProducerConfig

+ SecurityConfig SecurityConfig

+ type SecurityConfig struct

+ ClientCert *string

+ ClientKey *string

+ InsecureSkipVerify bool

+ Protocol *string

+ RootCACerts *string

Nov 13, 2020

Changes in this version

type Message

+ CommitAndRelease func()

+ Mark func()

+ Release func()

type SaramaMessage

+ func (M SaramaMessage) CommitAndRelease()

+ func (M SaramaMessage) Mark()

+ func (M SaramaMessage) Release()

Oct 22, 2020

Changes in this version

type ConsumerGroupConfig

+ Offset *int64

Sep 29, 2020

Changes in this version

+ const Closed

+ const Closer

+ const Errors

+ const MsgHealthyConsumerGroup

+ const MsgHealthyProducer

+ const OffsetNewest

+ const OffsetOldest

+ const Output

+ const Ready

+ const ServiceName

+ const Upstream

+ const UpstreamDone

+ var ConsumeErrRetryPeriod = 250 * time.Millisecond

+ var ErrInitSarama = errors.New("Failed to initialise client")

+ var ErrShutdownTimedOut = errors.New("Shutdown context timed out")

+ var ErrUninitialisedProducer = errors.New("Producer is not initialised")

+ var InitRetryPeriod = 250 * time.Millisecond

+ func SetMaxMessageSize(maxSize int32)

+ type ConsumerGroup struct

+ func NewConsumerGroup(ctx context.Context, brokerAddrs []string, topic, group string, ...) (*ConsumerGroup, error)

+ func (cg *ConsumerGroup) Channels() *ConsumerGroupChannels

+ func (cg *ConsumerGroup) Checker(ctx context.Context, state *health.CheckState) error

+ func (cg *ConsumerGroup) Close(ctx context.Context) (err error)

+ func (cg *ConsumerGroup) Initialise(ctx context.Context) error

+ func (cg *ConsumerGroup) IsInitialised() bool

+ func (cg *ConsumerGroup) StopListeningToConsumer(ctx context.Context) (err error)

+ type ConsumerGroupChannels struct

+ Closed chan struct{}

+ Closer chan struct{}

+ Errors chan error

+ Ready chan struct{}

+ Upstream chan Message

+ func CreateConsumerGroupChannels(bufferSize int) *ConsumerGroupChannels

+ func (consumerChannels *ConsumerGroupChannels) LogErrors(ctx context.Context, errMsg string)

+ func (consumerChannels *ConsumerGroupChannels) Validate() error

+ type ConsumerGroupConfig struct

+ KafkaVersion *string

+ KeepAlive *time.Duration

+ RetryBackoff *time.Duration

+ RetryBackoffFunc *func(retries int) time.Duration

+ type ErrBrokersNotReachable struct

+ Addrs []string

+ func (e *ErrBrokersNotReachable) Error() string

+ type ErrInvalidBrokers struct

+ Addrs []string

+ func (e *ErrInvalidBrokers) Error() string

+ type ErrNoChannel struct

+ ChannelNames []string

+ func (e *ErrNoChannel) Error() string

+ type IConsumerGroup interface

+ Channels func() *ConsumerGroupChannels

+ Checker func(ctx context.Context, state *health.CheckState) error

+ Close func(ctx context.Context) (err error)

+ Initialise func(ctx context.Context) error

+ IsInitialised func() bool

+ StopListeningToConsumer func(ctx context.Context) (err error)

+ type IProducer interface

+ Channels func() *ProducerChannels

+ Checker func(ctx context.Context, state *health.CheckState) error

+ Close func(ctx context.Context) (err error)

+ Initialise func(ctx context.Context) error

+ IsInitialised func() bool

+ type Message interface

+ Commit func()

+ GetData func() []byte

+ Offset func() int64

+ UpstreamDone func() chan struct{}

+ type Producer struct

+ func NewProducer(ctx context.Context, brokerAddrs []string, topic string, ...) (producer *Producer, err error)

+ func (p *Producer) Channels() *ProducerChannels

+ func (p *Producer) Checker(ctx context.Context, state *health.CheckState) error

+ func (p *Producer) Close(ctx context.Context) (err error)

+ func (p *Producer) Initialise(ctx context.Context) error

+ func (p *Producer) IsInitialised() bool

+ type ProducerChannels struct

+ Closed chan struct{}

+ Closer chan struct{}

+ Errors chan error

+ Output chan []byte

+ Ready chan struct{}

+ func CreateProducerChannels() *ProducerChannels

+ func (producerChannels *ProducerChannels) LogErrors(ctx context.Context, errMsg string)

+ func (producerChannels *ProducerChannels) Validate() error

+ type ProducerConfig struct

+ KafkaVersion *string

+ KeepAlive *time.Duration

+ MaxMessageBytes *int

+ RetryBackoff *time.Duration

+ RetryBackoffFunc *func(retries, maxRetries int) time.Duration

+ RetryMax *int

+ type SaramaAsyncProducer = sarama.AsyncProducer

+ type SaramaConsumerGroup = sarama.ConsumerGroup

+ type SaramaConsumerGroupClaim = sarama.ConsumerGroupClaim

+ type SaramaConsumerGroupSession = sarama.ConsumerGroupSession

+ type SaramaMessage struct

+ func (M SaramaMessage) Commit()

+ func (M SaramaMessage) GetData() []byte

+ func (M SaramaMessage) Offset() int64

+ func (M SaramaMessage) UpstreamDone() chan struct{}