kafka

May 17, 2024

Changes in this version

+ const ErrNilConfig

+ type Message = kafka.Message

Mar 31, 2024

Changes in this version

+ const AuthTypeNone

+ const AuthTypePlain

+ const AuthTypeScram256

+ const AuthTypeScram512

+ const DefaultTimeout

+ const ErrConsumerAlreadyConnected

+ const ErrInvalidAuthType

+ const ErrMissingAdminBroker

+ const ErrMissingConsumerBroker

+ const ErrMissingConsumerGroup

+ const ErrMissingConsumerTopic

+ const ErrMissingProducerBroker

+ const ErrMissingProducerTopic

+ const ErrProducerClosed

+ type AdminConfig struct

+ AuthType string

+ Brokers string

+ Password string

+ Username string

+ func (c AdminConfig) Validate() error

+ type ConsumerConfig struct

+ AuthType string

+ Brokers string

+ Group string

+ Password string

+ Topic string

+ Username string

+ func (c ConsumerConfig) Validate() error

+ type ConsumerFunc func(ctx context.Context, message kafka.Message) error

+ type KafkaAdmin struct

+ Conn *kafka.Conn

+ func NewAdmin(ctx context.Context, cfg AdminConfig) (*KafkaAdmin, error)

+ func (c *KafkaAdmin) Connect() error

+ func (c *KafkaAdmin) CreateTopic(topic string, numPartitions int, replicationFactor int) error

+ func (c *KafkaAdmin) DeleteTopic(topic string) error

+ func (c *KafkaAdmin) Disconnect()

+ func (c *KafkaAdmin) GetTopics(topics ...string) ([]kafka.Partition, error)

+ func (c *KafkaAdmin) IsConnected() bool

+ func (c *KafkaAdmin) ListTopics() ([]string, error)

+ func (c *KafkaAdmin) TopicExists(topic string) (bool, error)

+ type KafkaConsumer struct

+ Brokers string

+ Group string

+ Reader *kafka.Reader

+ Topic string

+ func NewConsumer(ctx context.Context, cfg ConsumerConfig) (*KafkaConsumer, error)

+ func (c *KafkaConsumer) ChannelSubscribe(ch chan kafka.Message) error

+ func (c *KafkaConsumer) Connect()

+ func (c *KafkaConsumer) Disconnect()

+ func (c *KafkaConsumer) GetConfig() *kafka.ReaderConfig

+ func (c *KafkaConsumer) IsConnected() bool

+ func (c *KafkaConsumer) ReadMessage() (kafka.Message, error)

+ func (c *KafkaConsumer) Rewind() error

+ func (c *KafkaConsumer) Subscribe(handler ConsumerFunc) error

+ func (c *KafkaConsumer) SubscribeWithOffsets(handler ConsumerFunc) error

+ type KafkaProducer struct

+ Brokers string

+ Topic string

+ Writer *kafka.Writer

+ func NewProducer(ctx context.Context, cfg ProducerConfig) (*KafkaProducer, error)

+ func (c *KafkaProducer) Disconnect()

+ func (p *KafkaProducer) IsConnected() bool

+ func (p *KafkaProducer) Write(value []byte, key ...[]byte) error

+ func (p *KafkaProducer) WriteJson(data interface{}, key ...[]byte) error

+ func (p *KafkaProducer) WriteMulti(values ...[]byte) error

+ func (p *KafkaProducer) WriteMultiJson(values ...interface{}) error

+ type ProducerConfig struct

+ AuthType string

+ Brokers string

+ Password string

+ Topic string

+ Username string

+ func (c ProducerConfig) Validate() error