queue

package

v0.6.0 Latest Latest Go to latest Published: Aug 31, 2025 License: MIT Imports: 23 Imported by: 0

Details

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

Repository

github.com/mss-boot-io/mss-boot

Links

Open Source Insights

Documentation ¶

Index ¶

func SCRAMClientGeneratorFuncSHA256() sarama.SCRAMClient
func SCRAMClientGeneratorFuncSHA512() sarama.SCRAMClient
type ConsumerGroupHandler
type ConsumerRegister
type Kafka
- func NewKafka(brokers []string, c *sarama.Config, h ConsumerGroupHandler, provider string) (k *Kafka, err error)
- func (e *Kafka) Append(opts ...storage.Option) error
- func (e *Kafka) Register(opts ...storage.Option)
- func (e *Kafka) Run(ctx context.Context)
- func (e *Kafka) Shutdown()
- func (*Kafka) String() string
type KafkaRunReader
type Memory
- func NewMemory(poolNum uint) *Memory
- func (m *Memory) Append(opts ...storage.Option) error
- func (m *Memory) Register(opts ...storage.Option)
- func (m *Memory) Run(context.Context)
- func (m *Memory) Shutdown()
- func (*Memory) String() string
type Message
- func (m *Message) GetContext() context.Context
- func (m *Message) GetErrorCount() int
- func (m *Message) GetID() string
- func (m *Message) GetPrefix() (prefix string)
- func (m *Message) GetStream() string
- func (m *Message) GetValues() map[string]any
- func (m *Message) SetContext(ctx context.Context)
- func (m *Message) SetErrorCount(count int)
- func (m *Message) SetID(id string)
- func (m *Message) SetPrefix(prefix string)
- func (m *Message) SetStream(stream string)
- func (m *Message) SetValues(values map[string]any)
type MessageHandler
- func (h *MessageHandler) Cleanup(sarama.ConsumerGroupSession) error
- func (h *MessageHandler) ConsumeClaim(s sarama.ConsumerGroupSession, c sarama.ConsumerGroupClaim) error
- func (h *MessageHandler) SetConsumerFunc(f storage.ConsumerFunc)
- func (h *MessageHandler) Setup(s sarama.ConsumerGroupSession) error
type NSQ
- func NewNSQ(cfg *nsq.Config, lookup, adminAddr string, addresses ...string) (*NSQ, error)
- func (e *NSQ) Append(opts ...storage.Option) error
- func (e *NSQ) Register(opts ...storage.Option)
- func (e *NSQ) Run(context.Context)
- func (e *NSQ) Shutdown()
- func (*NSQ) String() string
type NodesResp
type Redis
- func NewRedis(producerOptions *redisqueue.ProducerOptions, ...) (*Redis, error)
- func (r *Redis) Append(opts ...storage.Option) error
- func (r *Redis) Register(opts ...storage.Option)
- func (r *Redis) Run(context.Context)
- func (r *Redis) Shutdown()
- func (*Redis) String() string
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 ¶

This section is empty.

Variables ¶

This section is empty.

Functions ¶

func SCRAMClientGeneratorFuncSHA256 ¶

func SCRAMClientGeneratorFuncSHA256() sarama.SCRAMClient

func SCRAMClientGeneratorFuncSHA512 ¶

func SCRAMClientGeneratorFuncSHA512() sarama.SCRAMClient

Types ¶

type ConsumerGroupHandler ¶

type ConsumerGroupHandler interface {
	sarama.ConsumerGroupHandler
	SetConsumerFunc(f storage.ConsumerFunc)
}

type ConsumerRegister ¶

type ConsumerRegister struct {
	Topic     string
	GroupID   string
	Partition int
	Func      ConsumerGroupHandler
}

type Kafka ¶

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

func NewKafka ¶

func NewKafka(brokers []string, c *sarama.Config, h ConsumerGroupHandler, provider string) (k *Kafka, err error)

func (*Kafka) Append ¶

func (e *Kafka) Append(opts ...storage.Option) error

func (*Kafka) Register ¶

func (e *Kafka) Register(opts ...storage.Option)

func (*Kafka) Run ¶

func (e *Kafka) Run(ctx context.Context)

func (*Kafka) Shutdown ¶

func (e *Kafka) Shutdown()

func (*Kafka) String ¶

func (*Kafka) String() string

type KafkaRunReader ¶

type KafkaRunReader struct {
	Topic   string
	GroupID string
	Func    storage.ConsumerFunc
}

type Memory ¶

type Memory struct {
	PoolNum uint
	// contains filtered or unexported fields
}

func NewMemory ¶

func NewMemory(poolNum uint) *Memory

NewMemory 内存模式

func (*Memory) Append ¶

func (m *Memory) Append(opts ...storage.Option) error

func (*Memory) Register ¶

func (m *Memory) Register(opts ...storage.Option)

func (*Memory) Run ¶

func (m *Memory) Run(context.Context)

func (*Memory) Shutdown ¶

func (m *Memory) Shutdown()

func (*Memory) String ¶

func (*Memory) String() string

type Message ¶

type Message struct {
	redisqueue.Message
	ErrorCount int
	// contains filtered or unexported fields
}

func (*Message) GetContext ¶

func (m *Message) GetContext() context.Context

func (*Message) GetErrorCount ¶

func (m *Message) GetErrorCount() int

func (*Message) GetID ¶

func (m *Message) GetID() string

func (*Message) GetPrefix ¶

func (m *Message) GetPrefix() (prefix string)

func (*Message) GetStream ¶

func (m *Message) GetStream() string

func (*Message) GetValues ¶

func (m *Message) GetValues() map[string]any

func (*Message) SetContext ¶

func (m *Message) SetContext(ctx context.Context)

func (*Message) SetErrorCount ¶

func (m *Message) SetErrorCount(count int)

func (*Message) SetID ¶

func (m *Message) SetID(id string)

func (*Message) SetPrefix ¶

func (m *Message) SetPrefix(prefix string)

func (*Message) SetStream ¶

func (m *Message) SetStream(stream string)

func (*Message) SetValues ¶

func (m *Message) SetValues(values map[string]any)

type MessageHandler ¶

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

func (*MessageHandler) Cleanup ¶

func (h *MessageHandler) Cleanup(sarama.ConsumerGroupSession) error

func (*MessageHandler) ConsumeClaim ¶

func (h *MessageHandler) ConsumeClaim(s sarama.ConsumerGroupSession, c sarama.ConsumerGroupClaim) error

func (*MessageHandler) SetConsumerFunc ¶

func (h *MessageHandler) SetConsumerFunc(f storage.ConsumerFunc)

func (*MessageHandler) Setup ¶

func (h *MessageHandler) Setup(s sarama.ConsumerGroupSession) error

type NSQ ¶

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

func NewNSQ ¶

func NewNSQ(cfg *nsq.Config, lookup, adminAddr string, addresses ...string) (*NSQ, error)

NewNSQ nsq模式只能监听一个channel

func (*NSQ) Append ¶

func (e *NSQ) Append(opts ...storage.Option) error

Append 消息入生产者

func (*NSQ) Register ¶

func (e *NSQ) Register(opts ...storage.Option)

Register 监听消费者

func (*NSQ) Run ¶

func (e *NSQ) Run(context.Context)

func (*NSQ) Shutdown ¶

func (e *NSQ) Shutdown()

func (*NSQ) String ¶

func (*NSQ) String() string

String 字符串类型

type NodesResp ¶

type NodesResp struct {
	Nodes   []*peerInfo `json:"nodes"`
	Message string      `json:"message"`
}

type Redis ¶

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

Redis cache implement

func NewRedis ¶

func NewRedis(
	producerOptions *redisqueue.ProducerOptions,
	consumerOptions *redisqueue.ConsumerOptions,
) (*Redis, error)

NewRedis redis模式

func (*Redis) Append ¶

func (r *Redis) Append(opts ...storage.Option) error

func (*Redis) Register ¶

func (r *Redis) Register(opts ...storage.Option)

func (*Redis) Run ¶

func (r *Redis) Run(context.Context)

func (*Redis) Shutdown ¶

func (r *Redis) Shutdown()

func (*Redis) String ¶

func (*Redis) String() string

type XDGSCRAMClient ¶

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

func (*XDGSCRAMClient) Begin ¶

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

func (*XDGSCRAMClient) Done ¶

func (x *XDGSCRAMClient) Done() bool

func (*XDGSCRAMClient) Step ¶

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

Source Files ¶

View all Source files

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