kafka

README

Kafka

Broker: Kafka

You can run examples by replace configs use your kafka instance configs.

Documentation

Index

• type Consumer
  • func NewConsumer(config *ConsumerConfig) *Consumer
  • func (c *Consumer) Cleanup(sarama.ConsumerGroupSession) error
  • func (c *Consumer) Close() error
  • func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (c *Consumer) Setup(sarama.ConsumerGroupSession) error
  • func (c *Consumer) Start()
  • func (c *Consumer) Subscribe(handler ConsumerHandler)
• type ConsumerConfig
• type ConsumerHandler
• type ProducerConfig
• type SyncProducer
  • func NewSyncProducer(config *ProducerConfig) *SyncProducer
  • func (sp *SyncProducer) Close() error
  • func (sp *SyncProducer) SendSyncMsg(ctx context.Context, content string) error
• type TraceInterceptor
  • func (ti *TraceInterceptor) OnSend(message *sarama.ProducerMessage)

Types

type Consumer

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

func NewConsumer

func NewConsumer(config *ConsumerConfig) *Consumer

func (*Consumer) Cleanup

func (c *Consumer) Cleanup(sarama.ConsumerGroupSession) error

Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited

func (*Consumer) Close

func (c *Consumer) Close() error

func (*Consumer) ConsumeClaim

func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().

func (*Consumer) Setup

func (c *Consumer) Setup(sarama.ConsumerGroupSession) error

Setup is run at the beginning of a new session, before ConsumeClaim

func (*Consumer) Start

func (c *Consumer) Start()

func (*Consumer) Subscribe

func (c *Consumer) Subscribe(handler ConsumerHandler)

type ConsumerConfig

type ConsumerConfig struct {
	Addr  []string
	Topic []string
	Gid   string

	EnableSASLAuth bool
	SASLMechanism  string
	SASLUser       string
	SASLPassword   string
	SASLHandshake  bool

	DialTimeout time.Duration

	ConsumeOldest     bool
	EnableReturnError bool

	ClientID string
}

type ConsumerHandler

type ConsumerHandler func(context.Context, *sarama.ConsumerMessage) error

type ProducerConfig

type ProducerConfig struct {
	Addr  []string
	Topic []string

	EnableSASLAuth bool
	SASLMechanism  string
	SASLUser       string
	SASLPassword   string
	SASLHandshake  bool

	DialTimeout      time.Duration
	SlowSendDuration time.Duration

	EnableReturnSuccess bool

	ClientID string
}

type SyncProducer

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

func NewSyncProducer

func NewSyncProducer(config *ProducerConfig) *SyncProducer

func (*SyncProducer) Close

func (sp *SyncProducer) Close() error

func (*SyncProducer) SendSyncMsg

func (sp *SyncProducer) SendSyncMsg(ctx context.Context, content string) error

type TraceInterceptor

type TraceInterceptor struct {
	TraceID string
}

func (*TraceInterceptor) OnSend

func (ti *TraceInterceptor) OnSend(message *sarama.ProducerMessage)