redisq

Documentation

Index

• func NewFactory() taskq.Factory
• type Queue
  • func NewQueue(opt *taskq.QueueOptions) *Queue
  • func (q *Queue) Add(msg *taskq.Message) error
  • func (q *Queue) Close() error
  • func (q *Queue) CloseTimeout(timeout time.Duration) error
  • func (q *Queue) Consumer() taskq.QueueConsumer
  • func (q *Queue) Delete(msg *taskq.Message) error
  • func (q *Queue) Len() (int, error)
  • func (q *Queue) Name() string
  • func (q *Queue) Options() *taskq.QueueOptions
  • func (q *Queue) Purge() error
  • func (q *Queue) Release(msg *taskq.Message) error
  • func (q *Queue) ReserveN(ctx context.Context, n int, waitTimeout time.Duration) ([]taskq.Message, error)
  • func (q *Queue) String() string
• type RedisStreamClient

Functions

func NewFactory

func NewFactory() taskq.Factory

Types

type Queue

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

func NewQueue

func NewQueue(opt *taskq.QueueOptions) *Queue

func (*Queue) Add

func (q *Queue) Add(msg *taskq.Message) error

Add adds message to the queue.

func (*Queue) Close

func (q *Queue) Close() error

Close is like CloseTimeout with 30 seconds timeout.

func (*Queue) CloseTimeout

func (q *Queue) CloseTimeout(timeout time.Duration) error

CloseTimeout closes the queue waiting for pending messages to be processed.

func (*Queue) Consumer

func (q *Queue) Consumer() taskq.QueueConsumer

func (*Queue) Delete

func (q *Queue) Delete(msg *taskq.Message) error

Delete deletes the message from the queue.

func (*Queue) Len

func (q *Queue) Len() (int, error)

func (*Queue) Name

func (q *Queue) Name() string

func (*Queue) Options

func (q *Queue) Options() *taskq.QueueOptions

func (*Queue) Purge

func (q *Queue) Purge() error

Purge deletes all messages from the queue.

func (*Queue) Release

func (q *Queue) Release(msg *taskq.Message) error

func (*Queue) ReserveN

func (q *Queue) ReserveN(
	ctx context.Context, n int, waitTimeout time.Duration,
) ([]taskq.Message, error)

func (*Queue) String

func (q *Queue) String() string

type RedisStreamClient

type RedisStreamClient interface {
	Del(ctx context.Context, keys ...string) *redis.IntCmd
	TxPipeline() redis.Pipeliner

	XAdd(ctx context.Context, a *redis.XAddArgs) *redis.StringCmd
	XDel(ctx context.Context, stream string, ids ...string) *redis.IntCmd
	XLen(ctx context.Context, stream string) *redis.IntCmd
	XRangeN(ctx context.Context, stream, start, stop string, count int64) *redis.XMessageSliceCmd
	XGroupCreateMkStream(ctx context.Context, stream, group, start string) *redis.StatusCmd
	XReadGroup(ctx context.Context, a *redis.XReadGroupArgs) *redis.XStreamSliceCmd
	XAck(ctx context.Context, stream, group string, ids ...string) *redis.IntCmd
	XPendingExt(ctx context.Context, a *redis.XPendingExtArgs) *redis.XPendingExtCmd
	XTrim(ctx context.Context, key string, maxLen int64) *redis.IntCmd
	XGroupDelConsumer(ctx context.Context, stream, group, consumer string) *redis.IntCmd

	ZAdd(ctx context.Context, key string, members ...*redis.Z) *redis.IntCmd
	ZRangeByScore(ctx context.Context, key string, opt *redis.ZRangeBy) *redis.StringSliceCmd
	ZRem(ctx context.Context, key string, members ...interface{}) *redis.IntCmd
	XInfoConsumers(ctx context.Context, key string, group string) *redis.XInfoConsumersCmd
}