kafka

Mar 2, 2023

Changes in this version

+ var SaramaNewBroker interfaces.BrokerGenerator = func(addr string) interfaces.SaramaBroker

+ func DrainTopic(ctx context.Context, cfg *DrainTopicConfig, in *DrainTopicInput, ...) error

+ func DrainTopics(ctx context.Context, cfg *DrainTopicConfig, in ...*DrainTopicInput)

+ func SafeSendConsumerMessage(ch chan<- *sarama.ConsumerMessage, val *sarama.ConsumerMessage) (err error)

type ConsumerGroup

+ func NewConsumerGroupWithGenerators(ctx context.Context, cgConfig *ConsumerGroupConfig, ...) (*ConsumerGroup, error)

+ func (cg *ConsumerGroup) GroupName() string

+ func (cg *ConsumerGroup) SaramaCgHandler() *SaramaHandler

+ func (cg *ConsumerGroup) Topic() string

+ type DrainTopicConfig struct

+ BatchSize int

+ BrokerAddrs []string

+ KafkaVersion *string

+ Timeout time.Duration

+ type DrainTopicInput struct

+ GroupName string

+ Topic string

type Producer

+ func NewProducerWithGenerators(ctx context.Context, pConfig *ProducerConfig, ...) (*Producer, error)

+ type SaramaHandler struct

+ func (sh *SaramaHandler) Cleanup(session sarama.ConsumerGroupSession) error

+ func (sh *SaramaHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

+ func (sh *SaramaHandler) Setup(session sarama.ConsumerGroupSession) error

Oct 24, 2022

Changes in this version

+ type OptFunc func()

Sep 13, 2022

Changes in this version

+ const TraceIDHeaderKey

type IProducer

+ AddHeader func(key, value 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

Feb 3, 2022

Changes in this version

+ const ErrorChanBufferSize

Jan 25, 2022

Changes in this version

type ConsumerGroup

+ func (cg *ConsumerGroup) StateWait(state State)

type IConsumerGroup

+ StateWait func(state State)

+ type StateChan struct

+ func NewStateChan() *StateChan

+ func (sc *StateChan) Channel() chan struct{}

+ func (sc *StateChan) RWMutex() *sync.RWMutex

+ func (sc *StateChan) Wait()

type StateMachine

+ func (sm *StateMachine) GetChan(s State) *StateChan

Dec 17, 2021

Changes in this version

+ const Closed

+ const Closer

+ const Consume

+ const Errors

+ const Initialised

+ const MsgHealthyConsumerGroup

+ const MsgHealthyProducer

+ const OffsetNewest

+ const OffsetOldest

+ const Output

+ const ServiceName

+ const Upstream

+ const UpstreamDone

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

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

+ func GetRetryTime(attempt int, retryTime, maxRetryTime time.Duration) time.Duration

+ func NewAdmin(brokerAddrs []string, adminCfg *AdminConfig) (sarama.ClusterAdmin, error)

+ func SafeClose(ch chan struct{}) (justClosed bool)

+ func SafeCloseBool(ch chan bool) (justClosed bool)

+ func SafeCloseBytes(ch chan []byte) (justClosed bool)

+ func SafeCloseErr(ch chan error) (justClosed bool)

+ func SafeCloseMessage(ch chan Message) (justClosed bool)

+ func SafeSendBool(ch chan bool, val bool) (err error)

+ func SafeSendBytes(ch chan []byte, val []byte) (err error)

+ func SafeSendErr(ch chan error, val error) (err error)

+ func SafeSendProducerMessage(ch chan<- *sarama.ProducerMessage, val *sarama.ProducerMessage) (err error)

+ func SetMaxMessageSize(maxSize int32)

+ func UnwrapLogData(err error) log.Data

+ func WaitWithTimeout(wg *sync.WaitGroup) bool

+ 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

+ func (a *AdminConfig) Get() (*sarama.Config, error)

+ type Batch struct

+ func NewBatch(batchSize int) *Batch

+ func (batch *Batch) Add(ctx context.Context, message Message)

+ func (batch *Batch) Clear()

+ func (batch *Batch) Commit()

+ func (batch *Batch) IsEmpty() bool

+ func (batch *Batch) IsFull() bool

+ func (batch *Batch) Size() int

+ type BatchHandler func(ctx context.Context, batch []Message) error

+ type Commiter interface

+ Commit func() bool

+ type ConsumerGroup struct

+ func NewConsumerGroup(ctx context.Context, cgConfig *ConsumerGroupConfig) (*ConsumerGroup, error)

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

+ func (cg *ConsumerGroup) Checker(ctx context.Context, state *healthcheck.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) LogErrors(ctx context.Context)

+ func (cg *ConsumerGroup) OnHealthUpdate(status string)

+ func (cg *ConsumerGroup) RegisterBatchHandler(ctx context.Context, batchHandler BatchHandler) error

+ func (cg *ConsumerGroup) RegisterHandler(ctx context.Context, h Handler) error

+ func (cg *ConsumerGroup) Start() error

+ func (cg *ConsumerGroup) State() string

+ func (cg *ConsumerGroup) Stop()

+ func (cg *ConsumerGroup) StopAndWait()

+ type ConsumerGroupChannels struct

+ Closed chan struct{}

+ Closer chan struct{}

+ Consume chan bool

+ Errors chan error

+ Initialised chan struct{}

+ State *ConsumerStateChannels

+ Upstream chan Message

+ func CreateConsumerGroupChannels(bufferSize int) *ConsumerGroupChannels

+ func (consumerChannels *ConsumerGroupChannels) Validate() error

+ type ConsumerGroupConfig struct

+ BatchSize *int

+ BatchWaitTime *time.Duration

+ BrokerAddrs []string

+ GroupName string

+ KafkaVersion *string

+ KeepAlive *time.Duration

+ MaxRetryPeriod *time.Duration

+ MessageConsumeTimeout *time.Duration

+ MinBrokersHealthy *int

+ MinRetryPeriod *time.Duration

+ NumWorkers *int

+ Offset *int64

+ RetryBackoff *time.Duration

+ RetryBackoffFunc *func(retries int) time.Duration

+ SecurityConfig *SecurityConfig

+ Topic string

+ func (c *ConsumerGroupConfig) Get() (*sarama.Config, error)

+ func (c *ConsumerGroupConfig) Validate() error

+ type ConsumerStateChannels struct

+ Closing chan struct{}

+ Consuming chan struct{}

+ Initialising chan struct{}

+ Starting chan struct{}

+ Stopped chan struct{}

+ Stopping chan struct{}

+ type Error struct

+ func NewError(err error, logData map[string]interface{}) *Error

+ func (e *Error) Error() string

+ func (e *Error) LogData() map[string]interface{}

+ func (e *Error) Unwrap() error

+ type Handler func(ctx context.Context, workerID int, msg Message) error

+ type HealthInfo struct

+ HasTopic bool

+ Reachable bool

+ type HealthInfoMap struct

+ func Healthcheck(ctx context.Context, brokers []SaramaBroker, topic string, cfg *sarama.Config) HealthInfoMap

+ func (h *HealthInfoMap) Set(broker SaramaBroker, healthInfo HealthInfo)

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

+ type IConsumerGroup interface

+ Channels func() *ConsumerGroupChannels

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

+ Close func(ctx context.Context) error

+ Initialise func(ctx context.Context) error

+ IsInitialised func() bool

+ LogErrors func(ctx context.Context)

+ OnHealthUpdate func(status string)

+ RegisterBatchHandler func(ctx context.Context, batchHandler BatchHandler) error

+ RegisterHandler func(ctx context.Context, h Handler) error

+ Start func() error

+ State func() string

+ Stop func()

+ StopAndWait func()

+ type IProducer interface

+ Channels func() *ProducerChannels

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

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

+ Initialise func(ctx context.Context) error

+ IsInitialised func() bool

+ LogErrors func(ctx context.Context)

+ Send func(schema *avro.Schema, event interface{}) error

+ type Message interface

+ Commit func()

+ CommitAndRelease func()

+ GetData func() []byte

+ Mark func()

+ Offset func() int64

+ Release func()

+ UpstreamDone func() chan struct{}

+ type Producer struct

+ func NewProducer(ctx context.Context, pConfig *ProducerConfig) (producer *Producer, err error)

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

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

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

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

+ func (p *Producer) IsInitialised() bool

+ func (p *Producer) LogErrors(ctx context.Context)

+ func (p *Producer) Send(schema *avro.Schema, event interface{}) error

+ type ProducerChannels struct

+ Closed chan struct{}

+ Closer chan struct{}

+ Errors chan error

+ Initialised chan struct{}

+ Output chan []byte

+ func CreateProducerChannels() *ProducerChannels

+ func (producerChannels *ProducerChannels) Validate() error

+ type ProducerConfig struct

+ BrokerAddrs []string

+ KafkaVersion *string

+ KeepAlive *time.Duration

+ MaxMessageBytes *int

+ MaxRetryPeriod *time.Duration

+ MinBrokersHealthy *int

+ MinRetryPeriod *time.Duration

+ RetryBackoff *time.Duration

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

+ RetryMax *int

+ SecurityConfig *SecurityConfig

+ Topic string

+ func (p *ProducerConfig) Get() (*sarama.Config, error)

+ func (p *ProducerConfig) Validate() error

+ type SaramaAsyncProducer = sarama.AsyncProducer

+ type SaramaBroker interface

+ Addr func() string

+ Close func() error

+ Connected func() (bool, error)

+ GetMetadata func(request *sarama.MetadataRequest) (*sarama.MetadataResponse, error)

+ Open func(conf *sarama.Config) error

+ type SaramaConsumerGroup = sarama.ConsumerGroup

+ type SaramaConsumerGroupClaim = sarama.ConsumerGroupClaim

+ type SaramaConsumerGroupSession = sarama.ConsumerGroupSession

+ type SaramaMessage struct

+ func NewSaramaMessage(m *sarama.ConsumerMessage, s sarama.ConsumerGroupSession, ud chan struct{}) *SaramaMessage

+ func (m SaramaMessage) Commit()

+ func (m SaramaMessage) CommitAndRelease()

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

+ func (m SaramaMessage) Mark()

+ func (m SaramaMessage) Offset() int64

+ func (m SaramaMessage) Release()

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

+ type SecurityConfig struct

+ ClientCert string

+ ClientKey string

+ InsecureSkipVerify bool

+ RootCACerts string

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

+ type State int

+ const Closing

+ const Consuming

+ const Initialising

+ const Starting

+ const Stopped

+ const Stopping

+ func (s State) String() string

+ type StateMachine struct

+ func NewConsumerStateMachine() *StateMachine

+ func (sm *StateMachine) Get() State

+ func (sm *StateMachine) Set(newState State)

+ func (sm *StateMachine) SetIf(allowed []State, newState State) error

+ func (sm *StateMachine) String() string

+ 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