kafka

Documentation

Overview

Package kafka https://github.com/Shopify/sarama/blob/master/examples/sasl_scram_client/scram_client.go

Index

• Constants
• Variables
• type Config
  • func NewDefultConfig() *Config
  • func (conf *Config) ValidatePublisherConfig() error
  • func (conf *Config) ValidateSubscriberConfig() error
• type Publisher
  • func NewPublisher(conf *Config) (*Publisher, error)
  • func (b *Publisher) Connect() error
  • func (b *Publisher) Debug(l logger.Logger)
  • func (b *Publisher) Disconnect() error
  • func (b *Publisher) Pub(topic string, e *event.Event) error
• type Subscriber
  • func NewSubscriber(conf *Config) (*Subscriber, error)
  • func (s *Subscriber) Cleanup(sarama.ConsumerGroupSession) error
  • func (s *Subscriber) Connect() error
  • func (s *Subscriber) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (b *Subscriber) Debug(l logger.Logger)
  • func (s *Subscriber) Disconnect() error
  • func (s *Subscriber) Setup(sarama.ConsumerGroupSession) error
  • func (s *Subscriber) Sub(topic string, h bus.EventHandler) error
• type XDGSCRAMClient
  • func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error)
  • func (x *XDGSCRAMClient) Done() bool
  • func (x *XDGSCRAMClient) Step(challenge string) (response string, err error)

Constants

const (
	MetaKafkaPartition = "kafka.partition"
	MetaKafkaKey       = "kafka.key"
	MetaKafkaHeaders   = "kafka.headers"
)

Variables

var (
	// SHA256 todo
	SHA256 scram.HashGeneratorFcn = func() hash.Hash { return sha256.New() }
	// SHA512 todo
	SHA512 scram.HashGeneratorFcn = func() hash.Hash { return sha512.New() }
)

var (
	// ErrNoBroker todo
	ErrNoBroker = errors.New("connect kafka error, no broker here")
)

Types

type Config

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

Config 配置

func NewDefultConfig

func NewDefultConfig() *Config

NewDefultConfig todo

func (*Config) ValidatePublisherConfig

func (conf *Config) ValidatePublisherConfig() error

ValidatePublisherConfig todo

func (*Config) ValidateSubscriberConfig

func (conf *Config) ValidateSubscriberConfig() error

ValidateSubscriberConfig todo

type Publisher

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

Publisher kafka broker

func NewPublisher

func NewPublisher(conf *Config) (*Publisher, error)

NewPublisher kafka broker

func (*Publisher) Connect

func (b *Publisher) Connect() error

Connect 连接

func (*Publisher) Debug

func (b *Publisher) Debug(l logger.Logger)

Debug 日志

func (*Publisher) Disconnect

func (b *Publisher) Disconnect() error

Disconnect 端口连接

func (*Publisher) Pub

func (b *Publisher) Pub(topic string, e *event.Event) error

Pub 发布事件

type Subscriber

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

Subscriber kafka broker

func NewSubscriber

func NewSubscriber(conf *Config) (*Subscriber, error)

NewSubscriber kafka broker

func (*Subscriber) Cleanup

func (s *Subscriber) Cleanup(sarama.ConsumerGroupSession) error

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

func (*Subscriber) Connect

func (s *Subscriber) Connect() error

Connect 连接

func (*Subscriber) ConsumeClaim

func (s *Subscriber) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

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

func (*Subscriber) Debug

func (b *Subscriber) Debug(l logger.Logger)

Debug 日志

func (*Subscriber) Disconnect

func (s *Subscriber) Disconnect() error

Disconnect 端口连接

func (*Subscriber) Setup

func (s *Subscriber) Setup(sarama.ConsumerGroupSession) error

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

func (*Subscriber) Sub

func (s *Subscriber) Sub(topic string, h bus.EventHandler) error

Sub 订阅事件

type XDGSCRAMClient

type XDGSCRAMClient struct {
	*scram.Client
	*scram.ClientConversation
	scram.HashGeneratorFcn
}

XDGSCRAMClient todo

func (*XDGSCRAMClient) Begin

func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error)

Begin todo

func (*XDGSCRAMClient) Done

func (x *XDGSCRAMClient) Done() bool

Done todo

func (*XDGSCRAMClient) Step

func (x *XDGSCRAMClient) Step(challenge string) (response string, err error)

Step todo