kafka

package

v0.0.0-...-19dc864 Latest Latest Go to latest Published: Nov 17, 2025 License: GPL-3.0 Imports: 7 Imported by: 0

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/ducnpdev/godev-kit

Links

Open Source Insights

Documentation ¶

Overview ¶

Package kafka implements Kafka producer and consumer functionality.

Index ¶

type Consumer
- func NewConsumer(brokers []string, topic, groupID string, handler MessageHandler, ...) *Consumer
type Manager
- func NewManager(brokers []string, logger zerolog.Logger) *Manager
- func NewManagerWithConfig(brokers []string, logger zerolog.Logger, producerEnabled, consumerEnabled bool) *Manager
type MessageHandler
type Producer
- func NewProducer(brokers []string, logger zerolog.Logger) *Producer
- func (p *Producer) Close() error
- func (p *Producer) SendMessage(ctx context.Context, topic string, key []byte, value interface{}) error

Constants ¶

This section is empty.

Variables ¶

This section is empty.

Functions ¶

This section is empty.

Types ¶

type Consumer ¶

type Consumer struct {
	// contains filtered or unexported fields
}

Consumer -.

func NewConsumer ¶

func NewConsumer(brokers []string, topic, groupID string, handler MessageHandler, logger zerolog.Logger) *Consumer

NewConsumer -.

func (*Consumer) Close ¶

func (c *Consumer) Close() error

Close -.

func (*Consumer) ConsumeMessages ¶

func (c *Consumer) ConsumeMessages(ctx context.Context, handler func(key, value []byte) error) error

ConsumeMessages -.

func (*Consumer) Start ¶

func (c *Consumer) Start(ctx context.Context) error

Start -.

type Manager ¶

type Manager struct {
	// contains filtered or unexported fields
}

Manager -.

func NewManager ¶

func NewManager(brokers []string, logger zerolog.Logger) *Manager

NewManager -.

func NewManagerWithConfig ¶

func NewManagerWithConfig(brokers []string, logger zerolog.Logger, producerEnabled, consumerEnabled bool) *Manager

NewManagerWithConfig creates a new manager with configuration

func (*Manager) AddConsumer ¶

func (m *Manager) AddConsumer(topic, groupID string, handler MessageHandler) error

AddConsumer -.

func (*Manager) Close ¶

func (m *Manager) Close() error

Close -.

func (*Manager) DisableConsumer ¶

func (m *Manager) DisableConsumer()

DisableConsumer disables the Kafka consumer

func (*Manager) DisableProducer ¶

func (m *Manager) DisableProducer()

DisableProducer disables the Kafka producer

func (*Manager) EnableConsumer ¶

func (m *Manager) EnableConsumer()

EnableConsumer enables the Kafka consumer

func (*Manager) EnableProducer ¶

func (m *Manager) EnableProducer()

EnableProducer enables the Kafka producer

func (*Manager) GetStatus ¶

func (m *Manager) GetStatus() map[string]interface{}

GetStatus returns the current status of both producer and consumer

func (*Manager) IsConsumerEnabled ¶

func (m *Manager) IsConsumerEnabled() bool

IsConsumerEnabled returns the current consumer status

func (*Manager) IsProducerEnabled ¶

func (m *Manager) IsProducerEnabled() bool

IsProducerEnabled returns the current producer status

func (*Manager) SendMessage ¶

func (m *Manager) SendMessage(ctx context.Context, topic string, key []byte, value interface{}) error

SendMessage -.

func (*Manager) StartAllConsumers ¶

func (m *Manager) StartAllConsumers(ctx context.Context)

StartAllConsumers -.

func (*Manager) StartConsumer ¶

func (m *Manager) StartConsumer(ctx context.Context, topic string) error

StartConsumer -.

type MessageHandler ¶

type MessageHandler func(ctx context.Context, key, value []byte) error

MessageHandler -.

type Producer ¶

type Producer struct {
	// contains filtered or unexported fields
}

Producer -.

func NewProducer ¶

func NewProducer(brokers []string, logger zerolog.Logger) *Producer

NewProducer -.

func (*Producer) Close ¶

func (p *Producer) Close() error

Close -.

func (*Producer) SendMessage ¶

func (p *Producer) SendMessage(ctx context.Context, topic string,
	key []byte, value interface{}) error

SendMessage -.

Source Files ¶

View all Source files

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