The highest tagged major version is
README
¶
Redis Streams queue adapter
redisstream is the target-oriented Redis Streams implementation of the
go-queue worker contract. It owns Redis Streams delivery,
settlement, retry, dead-letter, management, and shutdown behavior; it does not
own Redis deployment, application idempotency, or service lifecycle.
Status and requirements
This stable v1 package ships in the root module and requires Go 1.27.0 or later. It is not an independently released module.
Install
go get github.com/faustbrian/go-queue@v1
Quick start
worker, err := redisstream.NewWorkerE(
redisstream.WithAddr("127.0.0.1:6379"),
redisstream.WithStreamName("jobs"),
redisstream.WithGroup("workers"),
redisstream.WithConsumer("worker-1"),
)
if err != nil {
return err
}
defer worker.Shutdown()
Configuration is caller-owned and validated by NewWorkerE. The worker owns
its Redis client, read loop, timers, and settlement state until Shutdown.
Values are safe for their documented concurrent worker use; callbacks remain
application-owned. Delivery is at least once, and unknown settlement outcomes
must be reconciled rather than retried blindly. Addresses, credentials,
payloads, and tenant identifiers are sensitive and must not be logged.
Use this package in root module github.com/faustbrian/go-queue for Redis
Streams consumer-group work. Use the root
redisdb package for Redis Pub/Sub, or valkeystream for Valkey-native
Streams. The former github.com/faustbrian/go-queue/redisstream path remains a
deprecated compatibility facade during the v1 support interval.
See the Redis Streams guide,
migration guide, changelog,
security policy, support policy, and
MIT license. API documentation is published at
pkg.go.dev.
Navigation
Continue with the parent documentation index or the versioned Golib ecosystem index for cross-package selection and ownership.
Documentation
¶
Overview ¶
Package redisstream owns Redis Streams intake, stale pending-entry reclaim and explicit settlement. Reclaim idle time is not exclusive execution fencing. Failed/terminal records append before source acknowledgement; failures and crashes can duplicate records. Direct admission uses an atomic capacity check, while administrative mutations use separate commands with partial or unknown outcomes. Record retention is opt-in exact count trimming.
Index ¶
- Variables
- type Option
- func WithAddr(addr string) Option
- func WithBlockTime(m time.Duration) Option
- func WithCluster() Option
- func WithCommandTimeout(timeout time.Duration) Option
- func WithConnectTimeout(timeout time.Duration) Option
- func WithConnectionString(connectionString string) Option
- func WithConsumer(name string) Option
- func WithDB(db int) Option
- func WithDeadLetter(stream string, maxAttempts int64) Option
- func WithFailureStream(stream string) Option
- func WithGroup(name string) Option
- func WithLogger(l queue.Logger) Option
- func WithManagementStatus(metadata management.StatusMetadata) Option
- func WithMaxLength(m int64) Option
- func WithPassword(passwd string) Option
- func WithReclaim(minIdle, interval time.Duration, batchSize int64) Option
- func WithRecordRetention(maxRecords int64) Option
- func WithReplayDestinations(destinations ...string) Option
- func WithRequestTimeout(timeout time.Duration) Option
- func WithRunFunc(fn func(context.Context, core.TaskMessage) error) Option
- func WithSkipTLSVerify() Option
- func WithStreamName(name string) Option
- func WithTLS() Option
- func WithUsername(username string) Option
- type Stats
- type Worker
- func (*Worker) BackendName() string
- func (w *Worker) Execute(ctx context.Context, command management.Command) (management.CommandResult, error)
- func (w *Worker) Inspect(ctx context.Context, request management.InspectRequest) (management.JobRecord, error)
- func (w *Worker) ListDeadLetters(ctx context.Context, request management.PageRequest) (management.RecordPage, error)
- func (w *Worker) ListFailures(ctx context.Context, request management.PageRequest) (management.RecordPage, error)
- func (w *Worker) ObserveQueue(ctx context.Context) (management.QueueStatus, error)
- func (w *Worker) ObserveWorker(ctx context.Context) (management.WorkerStatus, error)
- func (w *Worker) Queue(task core.TaskMessage) error
- func (w *Worker) QueueName() string
- func (w *Worker) Request() (core.TaskMessage, error)
- func (w *Worker) Run(ctx context.Context, task core.TaskMessage) error
- func (w *Worker) Shutdown() error
- func (w *Worker) Stats(ctx context.Context) (Stats, error)
Constants ¶
This section is empty.
Variables ¶
var ( // ErrManagementStatusDisabled reports a worker without reporter metadata. ErrManagementStatusDisabled = errors.New("redisstream: management status disabled") // ErrInvalidManagementStatus reports malformed reporter metadata or time. ErrInvalidManagementStatus = errors.New("redisstream: invalid management status") )
var ( // ErrManagementControlDisabled reports missing management metadata or native client. ErrManagementControlDisabled = errors.Join( errors.New("redisstream: management control disabled"), management.ErrUnsupportedCapability, ) )
Functions ¶
This section is empty.
Types ¶
type Option ¶
type Option func(*options)
Option for queue system
func WithBlockTime ¶
WithBlockTime configures the preferred blocking read duration. Reads poll at least once per second so shutdown remains bounded. we use the block command to make sure if no entry is found we wait until an entry is found
func WithCommandTimeout ¶
WithCommandTimeout bounds record append and source settlement commands.
func WithConnectTimeout ¶
WithConnectTimeout bounds initial Redis connection validation.
func WithConnectionString ¶
WithConnectionString redis connection string
func WithDeadLetter ¶
WithDeadLetter configures the terminal stream and delivery-attempt limit.
func WithFailureStream ¶
WithFailureStream configures the stream retaining failed delivery attempts.
func WithManagementStatus ¶
func WithManagementStatus(metadata management.StatusMetadata) Option
WithManagementStatus enables native worker and queue status reporting.
func WithMaxLength ¶
WithMaxLength sets a hard source admission capacity for direct enqueue. Zero preserves the existing unbounded admission mode.
func WithReclaim ¶
WithReclaim configures bounded stale pending-entry recovery.
func WithRecordRetention ¶
WithRecordRetention deliberately enables exact maximum-count retention for failure and dead-letter streams. It is disabled by default.
func WithReplayDestinations ¶
WithReplayDestinations allowlists bounded logical streams for administrative replay. Replay remains disabled when this option is absent.
func WithRequestTimeout ¶
WithRequestTimeout sets how long Request waits for a stream message.
func WithRunFunc ¶
WithRunFunc setup the run func of queue
func WithSkipTLSVerify ¶
func WithSkipTLSVerify() Option
WithSkipTLSVerify returns an Option that configures the TLS settings to skip verification of the server's certificate. This is useful for connecting to servers with self-signed certificates or when certificate verification is not required. Use this option with caution as it makes the connection susceptible to man-in-the-middle attacks.
func WithTLS ¶
func WithTLS() Option
WithTLS returns an Option that configures the use of TLS for the connection. It sets the minimum TLS version to TLS 1.2.
func WithUsername ¶
WithUsername redis username This is only used for redis cluster
type Stats ¶
type Stats struct {
// Depth is Pending plus Lag, or -1 when Lag is unknown.
Depth int64
// Pending counts delivered entries still pending for this consumer group.
Pending int64
// Lag counts entries not yet delivered to the group when known.
Lag int64
// LagKnown distinguishes determinate depth from unsupported group lag.
LagKnown bool
// OldestJobAge is the oldest inspected pending/undelivered age; zero may be unavailable.
OldestJobAge time.Duration
}
Stats describes outstanding work for this worker's Redis consumer group. Depth is Pending plus Lag and is -1 when Redis cannot determine group lag.
type Worker ¶
type Worker struct {
// contains filtered or unexported fields
}
Worker owns the backend client and intake resources; use its constructor.
func NewWorkerE ¶
NewWorkerE creates a worker and returns connection and configuration errors.
func (*Worker) BackendName ¶
BackendName identifies Redis Streams in lifecycle events.
func (*Worker) Execute ¶
func (w *Worker) Execute( ctx context.Context, command management.Command, ) (management.CommandResult, error)
Execute applies a Redis Streams record mutation and caches its result by idempotency key in this process (1,024 entries, no eviction). Record reads, capacity checks, append, replay registration and source settlement use separate commands: concurrent clients and failures can leave partial or unknown effects. ctx must be non-nil; do not mutate retained Selection or Replay pointers.
func (*Worker) Inspect ¶
func (w *Worker) Inspect( ctx context.Context, request management.InspectRequest, ) (management.JobRecord, error)
Inspect returns one Redis failure or dead letter at explicit visibility.
func (*Worker) ListDeadLetters ¶
func (w *Worker) ListDeadLetters( ctx context.Context, request management.PageRequest, ) (management.RecordPage, error)
ListDeadLetters returns a bounded page of package-managed Redis dead letters.
func (*Worker) ListFailures ¶
func (w *Worker) ListFailures( ctx context.Context, request management.PageRequest, ) (management.RecordPage, error)
ListFailures returns a bounded page of package-managed Redis failure records.
func (*Worker) ObserveQueue ¶
func (w *Worker) ObserveQueue(ctx context.Context) (management.QueueStatus, error)
ObserveQueue returns honest native Redis Streams measurements.
func (*Worker) ObserveWorker ¶
func (w *Worker) ObserveWorker(ctx context.Context) (management.WorkerStatus, error)
ObserveWorker returns this worker's bounded native management observation.
func (*Worker) Queue ¶
func (w *Worker) Queue(task core.TaskMessage) error
Queue synchronously encodes and appends through a source-capacity Lua check with the configured command timeout. It does not bound task.Bytes output or propagate a caller context. Success is source admission, not settlement.
Source Files