Documentation
¶
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 = 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 approximate 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 ¶
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 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 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
Source Files