Documentation
¶
Index ¶
- type ConsumerConfig
- type Memory
- type Message
- 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]interface{}
- 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]interface{})
- type NSQ
- type ProducerConfig
- type Redis
- type StreamConsumer
- type StreamProducer
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type ConsumerConfig ¶ added in v1.1.51
type ConsumerConfig struct {
VisibilityTimeout int `mapstructure:"visibilityTimeout"` // seconds
BlockingTimeout int `mapstructure:"blockingTimeout"` // seconds
ReclaimInterval int `mapstructure:"reclaimInterval"` // seconds
BufferSize int `mapstructure:"bufferSize"`
Concurrency int `mapstructure:"concurrency"`
MaxDeliveries int64 `mapstructure:"maxDeliveries"` // max reclaim attempts; 0 = unlimited
}
ConsumerConfig configures the stream consumer group.
type Message ¶
type Message struct {
// contains filtered or unexported fields
}
Message is a standalone queue message with no external dependencies. Fields are unexported to enforce mutex-protected access.
func (*Message) GetErrorCount ¶
func (*Message) GetValues ¶
GetValues returns a shallow copy of the values map so callers cannot mutate the Message's internal state without going through setters.
func (*Message) SetErrorCount ¶
type NSQ ¶
type NSQ struct {
// contains filtered or unexported fields
}
type ProducerConfig ¶ added in v1.1.51
type ProducerConfig struct {
StreamMaxLength int64 `mapstructure:"streamMaxLength"`
ApproximateMaxLength bool `mapstructure:"approximateMaxLength"`
}
ProducerConfig configures the stream producer.
type Redis ¶
type Redis struct {
// contains filtered or unexported fields
}
Redis implements storage.AdapterQueue using native Redis Streams.
func NewRedis ¶
func NewRedis(client *redis.Client, producerCfg *ProducerConfig, consumerCfg *ConsumerConfig) (*Redis, error)
NewRedis creates a Redis-backed queue adapter using a single client for both producer and consumer. For the three-client architecture, use NewRedisWithConsumer instead.
func NewRedisWithConsumer ¶ added in v1.1.51
func NewRedisWithConsumer(producerClient, consumerClient *redis.Client, producerCfg *ProducerConfig, consumerCfg *ConsumerConfig) (*Redis, error)
NewRedisWithConsumer creates a Redis-backed queue adapter with separate clients for producer and consumer. The producerClient is used for XADD (non-blocking), the consumerClient is used for XREADGROUP (blocking).
type StreamConsumer ¶ added in v1.1.51
type StreamConsumer struct {
// contains filtered or unexported fields
}
StreamConsumer reads messages from a Redis Stream consumer group with at-least-once delivery semantics.
Three goroutine types:
- poll(): XREADGROUP BLOCK — pushes messages into a buffered channel
- reclaim(): periodic XPENDING + batch XCLAIM for timed-out messages
- N*work(): calls the user callback, then XACK on success
func NewStreamConsumer ¶ added in v1.1.51
func NewStreamConsumer(client *redis.Client, stream, group string, cfg *ConsumerConfig, handler storage.ConsumerFunc) (*StreamConsumer, error)
NewStreamConsumer creates a consumer bound to the given stream and group.
func (*StreamConsumer) Errors ¶ added in v1.1.51
func (sc *StreamConsumer) Errors() <-chan error
Errors returns a read-only channel that surfaces consumer errors. The channel is buffered and never blocks the consumer — when full, errors are silently dropped. Callers MUST drain this channel (e.g. with a dedicated goroutine that logs/alerts) or errors will be lost.
func (*StreamConsumer) Run ¶ added in v1.1.51
func (sc *StreamConsumer) Run()
Run starts the poll, reclaim, and worker goroutines.
func (*StreamConsumer) Shutdown ¶ added in v1.1.51
func (sc *StreamConsumer) Shutdown()
Shutdown signals all goroutines to stop and waits for them to drain.
type StreamProducer ¶ added in v1.1.51
type StreamProducer struct {
// contains filtered or unexported fields
}
StreamProducer publishes messages to Redis Streams.
func NewStreamProducer ¶ added in v1.1.51
func NewStreamProducer(client *redis.Client, cfg *ProducerConfig) *StreamProducer
NewStreamProducer creates a producer that writes to Redis Streams via XADD.
Source Files