kq

Documentation

Index

• Constants
• func Base642PEM(base64Key string) []byte
• func Base64ToPEM(base64String string, certType string) []byte
• func MustNewQueue(c KqConf, handler ConsumeHandler, opts ...QueueOption) queue.MessageQueue
• func NewQueue(c KqConf, handler ConsumeHandler, opts ...QueueOption) (queue.MessageQueue, error)
• func VerifyCertPemStr(certStr string) bool
• type ConsumeErrorHandler
• type ConsumeHandle
• type ConsumeHandler
  • func WithHandle(handle ConsumeHandle) ConsumeHandler
• type KqConf
• type KqPusherConf
• type KqSaslCaConf
• type PushOption
  • func WithAllowAutoTopicCreation() PushOption
  • func WithChunkSize(chunkSize int) PushOption
  • func WithFlushInterval(interval time.Duration) PushOption
• type Pusher
  • func NewConfPusher(c KqPusherConf, opts ...PushOption) *Pusher
  • func NewPusher(addrs []string, topic string, opts ...PushOption) *Pusher
  • func (p *Pusher) Close() error
  • func (p *Pusher) Name() string
  • func (p *Pusher) Push(v string) error
• type QueueOption
  • func WithCommitInterval(interval time.Duration) QueueOption
  • func WithErrorHandler(errorHandler ConsumeErrorHandler) QueueOption
  • func WithMaxWait(wait time.Duration) QueueOption
  • func WithMetrics(metrics *stat.Metrics) QueueOption
  • func WithQueueCapacity(queueCapacity int) QueueOption

Constants

const (
	SASL_SCRAM = 2
	SASL_PLAIN = 1

	CA_FILE = 1
	CA_TEXT = 2
)

Functions

func Base642PEM

func Base642PEM(base64Key string) []byte

func Base64ToPEM

func Base64ToPEM(base64String string, certType string) []byte

func MustNewQueue

func MustNewQueue(c KqConf, handler ConsumeHandler, opts ...QueueOption) queue.MessageQueue

func NewQueue

func NewQueue(c KqConf, handler ConsumeHandler, opts ...QueueOption) (queue.MessageQueue, error)

func VerifyCertPemStr

func VerifyCertPemStr(certStr string) bool

Types

type ConsumeErrorHandler

type ConsumeErrorHandler func(msg kafka.Message, err error)

type ConsumeHandle

type ConsumeHandle func(key, value string) error

type ConsumeHandler

type ConsumeHandler interface {
	Consume(key, value string) error
}

func WithHandle

func WithHandle(handle ConsumeHandle) ConsumeHandler

type KqConf

type KqConf struct {
	service.ServiceConf
	Brokers    []string
	Group      string
	Topic      string
	Offset     string `json:",options=first|last,default=last"`
	Conns      int    `json:",default=1"`
	Consumers  int    `json:",default=8"`
	Processors int    `json:",default=8"`
	MinBytes   int    `json:",default=10240"`    // 10K
	MaxBytes   int    `json:",default=10485760"` // 10M
	KqSaslCaConf
}

type KqPusherConf

type KqPusherConf struct {
	Brokers     []string
	Topic       string
	Compression int8 `json:",options=1|2|3|4,default=2"`
	KqSaslCaConf
}

type KqSaslCaConf

type KqSaslCaConf struct {
	Username    string `json:",optional"`
	Password    string `json:",optional"`
	ForceCommit bool   `json:",default=true"`
	SASL_WAY    int8   `json:",options=1|2,default=1"`
	CA_WAY      int8   `json:",options=1|2,default=1"`
	CaFile      string `json:",optional"`
	CertFile    string `json:",optional"`
	KeyFile     string `json:",optional"`
	CaPEM       string `json:",optional"`
	CertPEM     string `json:",optional"`
	KeyPEM      string `json:",optional"`
}

type PushOption

type PushOption func(options *pushOptions)

func WithAllowAutoTopicCreation

func WithAllowAutoTopicCreation() PushOption

WithAllowAutoTopicCreation allows the Pusher to create the given topic if it does not exist.

func WithChunkSize

func WithChunkSize(chunkSize int) PushOption

WithChunkSize customizes the Pusher with the given chunk size.

func WithFlushInterval

func WithFlushInterval(interval time.Duration) PushOption

WithFlushInterval customizes the Pusher with the given flush interval.

type Pusher

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

func NewConfPusher

func NewConfPusher(c KqPusherConf, opts ...PushOption) *Pusher

NewConfPusher returns a Pusher with the given Kafka addresses and topic.

func NewPusher

func NewPusher(addrs []string, topic string, opts ...PushOption) *Pusher

NewPusher returns a Pusher with the given Kafka addresses and topic.

func (*Pusher) Close

func (p *Pusher) Close() error

Close closes the Pusher and releases any resources used by it.

func (*Pusher) Name

func (p *Pusher) Name() string

Name returns the name of the Kafka topic that the Pusher is sending messages to.

func (*Pusher) Push

func (p *Pusher) Push(v string) error

Push sends a message to the Kafka topic.

type QueueOption

type QueueOption func(*queueOptions)

func WithCommitInterval

func WithCommitInterval(interval time.Duration) QueueOption

func WithErrorHandler

func WithErrorHandler(errorHandler ConsumeErrorHandler) QueueOption

func WithMaxWait

func WithMaxWait(wait time.Duration) QueueOption

func WithMetrics

func WithMetrics(metrics *stat.Metrics) QueueOption

func WithQueueCapacity

func WithQueueCapacity(queueCapacity int) QueueOption