redisdb

package
v1.0.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 28, 2026 License: MIT Imports: 19 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
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")
)
View Source 
var (
	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 WithAddr 

func WithAddr(addr string) Option

WithAddr setup the addr of redis

func WithBlockTime 

func WithBlockTime(m time.Duration) Option

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 WithCluster 

func WithCluster() Option

WithCluster redis cluster

func WithCommandTimeout 

func WithCommandTimeout(timeout time.Duration) Option

WithCommandTimeout bounds record append and source settlement commands.

func WithConnectTimeout 

func WithConnectTimeout(timeout time.Duration) Option

WithConnectTimeout bounds initial Redis connection validation.

func WithConnectionString 

func WithConnectionString(connectionString string) Option

WithConnectionString redis connection string

func WithConsumer 

func WithConsumer(name string) Option

WithConsumer consumer name

func WithDB 

func WithDB(db int) Option

WithPassword redis password

func WithDeadLetter 

func WithDeadLetter(stream string, maxAttempts int64) Option

WithDeadLetter configures the terminal stream and delivery-attempt limit.

func WithFailureStream 

func WithFailureStream(stream string) Option

WithFailureStream configures the stream retaining failed delivery attempts.

func WithGroup 

func WithGroup(name string) Option

WithGroup group name

func WithLogger 

func WithLogger(l queue.Logger) Option

WithLogger set custom logger

func WithManagementStatus 

func WithManagementStatus(metadata management.StatusMetadata) Option

WithManagementStatus enables native worker and queue status reporting.

func WithMaxLength 

func WithMaxLength(m int64) Option

WithMaxLength sets a hard source admission capacity for direct enqueue. Zero preserves the existing unbounded admission mode.

func WithPassword 

func WithPassword(passwd string) Option

WithPassword redis password

func WithReclaim 

func WithReclaim(minIdle, interval time.Duration, batchSize int64) Option

WithReclaim configures bounded stale pending-entry recovery.

func WithRecordRetention 

func WithRecordRetention(maxRecords int64) Option

WithRecordRetention deliberately enables approximate maximum-count retention for failure and dead-letter streams. It is disabled by default.

func WithReplayDestinations 

func WithReplayDestinations(destinations ...string) Option

WithReplayDestinations allowlists bounded logical streams for administrative replay. Replay remains disabled when this option is absent.

func WithRequestTimeout 

func WithRequestTimeout(timeout time.Duration) Option

WithRequestTimeout sets how long Request waits for a stream message.

func WithRunFunc 

func WithRunFunc(fn func(context.Context, core.TaskMessage) error) Option

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 WithStreamName 

func WithStreamName(name string) Option

WithStreamName Stream name

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 

func WithUsername(username string) Option

WithUsername redis username This is only used for redis cluster

type Stats 

type Stats struct {
	Depth        int64
	Pending      int64
	Lag          int64
	LagKnown     bool
	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 for Redis

func NewWorker 

func NewWorker(opts ...Option) *Worker

NewWorker for struc

func NewWorkerE 

func NewWorkerE(opts ...Option) (*Worker, error)

NewWorkerE creates a worker and returns connection and configuration errors.

func (*Worker) BackendName 

func (*Worker) BackendName() string

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 one bounded Redis Streams management command.

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 send notification to queue

func (*Worker) QueueName 

func (w *Worker) QueueName() string

QueueName returns the configured Redis stream.

func (*Worker) Request 

func (w *Worker) Request() (core.TaskMessage, error)

Request a new task

func (*Worker) Run 

func (w *Worker) Run(ctx context.Context, task core.TaskMessage) error

Run start the worker

func (*Worker) Shutdown 

func (w *Worker) Shutdown() error

Shutdown worker

func (*Worker) Stats 

func (w *Worker) Stats(ctx context.Context) (Stats, error)

Stats returns consumer-group depth and the age of its oldest outstanding job.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.