kq

package

v1.1.82 Latest Latest Go to latest Published: Jul 19, 2024 License: MIT Imports: 27 Imported by: 0

Details

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

Repository

github.com/sturn/go-queue

Links

Open Source Insights

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 ¶

View Source

const (
	SASL_SCRAM = 2
	SASL_PLAIN = 1

	CA_FILE = 1
	CA_TEXT = 2
)

Variables ¶

This section is empty.

Functions ¶

func Base642PEM ¶ added in v1.1.82

func Base642PEM(base64Key string) []byte

func Base64ToPEM ¶ added in v1.1.82

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 ¶ added in v1.1.82

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 ¶ added in v1.1.82

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

type KqSaslCaConf ¶ added in v1.1.82

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 ¶ added in v1.1.82

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

Source Files ¶

View all Source files

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