kafka

package module

Go to main page

Versions in this module

v1

v1.1.7

Aug 13, 2020

v1.1.6

May 29, 2020

Changes in this version

type IConsumerGroup

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

type IProducer

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

v1.1.5

Mar 27, 2020

v1.1.4

Mar 12, 2020

v1.1.3

Mar 9, 2020

Changes in this version

+ type IConsumerGroup interface

+ Channels func() *ConsumerGroupChannels

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

+ CommitAndRelease func(msg Message)

+ Initialise func(ctx context.Context) error

+ IsInitialised func() bool

+ Release func()

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

+ type IProducer interface

+ Channels func() *ProducerChannels

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

+ Initialise func(ctx context.Context) error

+ IsInitialised func() bool

v1.1.2

Mar 6, 2020

v1.1.1

Feb 28, 2020

Changes in this version

type ConsumerGroupChannels

+ Init chan struct{}

v1.1.0

Feb 17, 2020

Changes in this version

type ConsumerGroup

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

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

type ConsumerGroupChannels

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

type Producer

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

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

type ProducerChannels

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

v1.0.0

Feb 4, 2020

Changes in this version

+ const Closed

+ const Closer

+ const Errors

+ const Init

+ const MsgHealthyConsumerGroup

+ const MsgHealthyProducer

+ const OffsetNewest

+ const OffsetOldest

+ const Output

+ const ServiceName

+ const Upstream

+ const UpstreamDone

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

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

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

+ func SetMaxMessageSize(maxSize int32)

+ type AsyncProducer = sarama.AsyncProducer

+ type ConsumerGroup struct

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

+ func NewConsumerWithClusterClient(ctx context.Context, brokers []string, topic string, group string, ...) (cg ConsumerGroup, err error)

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

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

+ func (cg *ConsumerGroup) CommitAndRelease(msg Message)

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

+ func (cg *ConsumerGroup) IsInitialised() bool

+ func (cg *ConsumerGroup) Release()

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

+ type ConsumerGroupChannels struct

+ Closed chan struct{}

+ Closer chan struct{}

+ Errors chan error

+ Upstream chan Message

+ UpstreamDone chan bool

+ func CreateConsumerGroupChannels(sync bool) ConsumerGroupChannels

+ func (cCh *ConsumerGroupChannels) Validate() error

+ 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 Message interface

+ Commit func()

+ GetData func() []byte

+ Offset func() int64

+ type MessageConsumer interface

+ Closer func() chan bool

+ Errors func() chan error

+ Incoming func() chan Message

+ type MessageProducer interface

+ Closer func() chan bool

+ Errors func() chan error

+ Output func() chan []byte

+ type Producer struct

+ func NewProducer(ctx context.Context, brokers []string, topic string, envMax int, ...) (Producer, error)

+ func NewProducerWithSaramaClient(ctx context.Context, brokers []string, topic string, envMax int, ...) (producer Producer, err error)

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

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

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

+ func (p *Producer) IsInitialised() bool

+ type ProducerChannels struct

+ Closed chan struct{}

+ Closer chan struct{}

+ Errors chan error

+ Init chan struct{}

+ Output chan []byte

+ func CreateProducerChannels() ProducerChannels

+ func (pCh *ProducerChannels) Validate() error

+ type Sarama interface

+ NewAsyncProducer func(addrs []string, conf *sarama.Config) (sarama.AsyncProducer, error)

+ type SaramaClient struct

+ func (s *SaramaClient) NewAsyncProducer(addrs []string, conf *sarama.Config) (AsyncProducer, error)

+ type SaramaCluster interface

+ NewConsumer func(addrs []string, groupID string, topics []string, config *cluster.Config) (SaramaClusterConsumer, error)

+ type SaramaClusterClient struct

+ func (c *SaramaClusterClient) NewConsumer(addrs []string, groupID string, topics []string, config *cluster.Config) (SaramaClusterConsumer, error)

+ type SaramaClusterConsumer interface

+ Close func() (err error)

+ CommitOffsets func() error

+ Errors func() <-chan error

+ MarkOffset func(msg *sarama.ConsumerMessage, metadata string)

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

+ Notifications func() <-chan *cluster.Notification

+ type SaramaMessage struct

+ func (M SaramaMessage) Commit()

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

+ func (M SaramaMessage) Offset() int64

Other modules containing this package

github.com/ONSdigital/dp-kafka/v2

github.com/ONSdigital/dp-kafka/v3

github.com/ONSdigital/dp-kafka/v4

github.com/ONSdigital/dp-kafka/v5

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