disq

package module

Go to main page

Versions in this module

v0

v0.1.8

May 30, 2022

Changes in this version

type Broker

+ Config func() Config

+ Name func() string

type Redis

+ XInfoStream func(ctx context.Context, key string) *redis.XInfoStreamCmd

+ XPending func(ctx context.Context, stream, group string) *redis.XPendingCmd

+ XRange func(ctx context.Context, stream, start, stop string) *redis.XMessageSliceCmd

type TaskMap

+ func (s *TaskMap) UpdateTask(name string, task *Task) error

v0.1.7

May 19, 2022

v0.1.6

May 19, 2022

Changes in this version

+ func ErrorHandler(msg *Message, msgErr error, retries *uint32) error

+ func FormatHandlerError(msg *Message, retrylimit int) error

+ type DisqError interface

+ GetDelay func() time.Duration

+ GetRateLimit func() bool

+ type Error struct

+ Delay time.Duration

+ Err error

+ RateLimit bool

+ func (e *Error) Error() string

+ func (e *Error) GetDelay() time.Duration

+ func (e *Error) GetRateLimit() bool

v0.1.5

May 13, 2022

Changes in this version

type Message

+ func (m *Message) MarshalArgs() ([]byte, error)

v0.1.4

May 12, 2022

Changes in this version

+ func BytesToString(b []byte) string

+ func StringToBytes(s string) []byte

v0.1.3

May 12, 2022

Changes in this version

type Message

+ Name string

v0.1.2

May 12, 2022

v0.1.1

May 12, 2022

Changes in this version

type Message

+ ArgsBin []byte

+ func (m *Message) MarshalBinary() ([]byte, error)

+ func (m *Message) UnmarshalBinary(b []byte) error

v0.1.0

May 11, 2022

Changes in this version

+ var Logger *log.Logger

+ func ConsumerName() string

+ func Delay(msg *Message, msgErr error) time.Duration

+ func DurEqual(d1, d2 time.Duration, threshold int) bool

+ func Scheduler(name string, c *redis.Client, fn func(ctx context.Context) (int, error))

+ func SetLogger(logger *log.Logger)

+ func UnixMs(tm time.Time) int64

+ func Version() string

+ func WithRedisLock(ctx context.Context, name string, redis Redis, ...) error

+ type Broker interface

+ Consume func(context.Context)

+ Delete func(*Message) error

+ FetchN func(context.Context, int, time.Duration) ([]Message, error)

+ Len func() (int, error)

+ Process func(*Message) error

+ Publish func(*Message) error

+ Stats func() *Stats

+ Status func() bool

+ Stop func() error

+ type Config interface

+ Init func() error

+ type Delayer interface

+ Delay func() time.Duration

+ type Handler interface

+ HandleMessage func(msg *Message) error

+ func NewHandler(fn interface{}) Handler

+ type HandlerFunc func(*Message) error

+ func (fn HandlerFunc) HandleMessage(msg *Message) error

+ type Message struct

+ Args []interface{}

+ Ctx context.Context

+ Delay time.Duration

+ Err error

+ ExecutionTime time.Time

+ ID string

+ RetryCount int

+ TaskName string

+ func NewMessage(ctx context.Context, args ...interface{}) *Message

+ func (m *Message) SetDelay(delay time.Duration)

+ type MessageRaw Message

+ type Redis interface

+ Del func(ctx context.Context, keys ...string) *redis.IntCmd

+ Eval func(ctx context.Context, script string, keys []string, args ...interface{}) *redis.Cmd

+ EvalSha func(ctx context.Context, sha1 string, keys []string, args ...interface{}) *redis.Cmd

+ LIndex func(ctx context.Context, key string, index int64) *redis.StringCmd

+ LLen func(ctx context.Context, key string) *redis.IntCmd

+ LPop func(ctx context.Context, key string) *redis.StringCmd

+ LPopCount func(ctx context.Context, key string, count int) *redis.StringSliceCmd

+ LPush func(ctx context.Context, key string, values ...interface{}) *redis.IntCmd

+ LRange func(ctx context.Context, key string, start, stop int64) *redis.StringSliceCmd

+ LRem func(ctx context.Context, key string, count int64, value interface{}) *redis.IntCmd

+ LTrim func(ctx context.Context, key string, start, stop int64) *redis.StatusCmd

+ Pipelined func(ctx context.Context, fn func(pipe redis.Pipeliner) error) ([]redis.Cmder, error)

+ ScriptExists func(ctx context.Context, scripts ...string) *redis.BoolSliceCmd

+ ScriptLoad func(ctx context.Context, script string) *redis.StringCmd

+ SetNX func(ctx context.Context, key string, value interface{}, expiration time.Duration) *redis.BoolCmd

+ TxPipeline func() redis.Pipeliner

+ XAck func(ctx context.Context, stream, group string, ids ...string) *redis.IntCmd

+ XAdd func(ctx context.Context, a *redis.XAddArgs) *redis.StringCmd

+ XDel func(ctx context.Context, stream string, ids ...string) *redis.IntCmd

+ XGroupCreateMkStream func(ctx context.Context, stream, group, start string) *redis.StatusCmd

+ XGroupDelConsumer func(ctx context.Context, stream, group, consumer string) *redis.IntCmd

+ XInfoConsumers func(ctx context.Context, key string, group string) *redis.XInfoConsumersCmd

+ XLen func(ctx context.Context, stream string) *redis.IntCmd

+ XPendingExt func(ctx context.Context, a *redis.XPendingExtArgs) *redis.XPendingExtCmd

+ XRangeN func(ctx context.Context, stream, start, stop string, count int64) *redis.XMessageSliceCmd

+ XReadGroup func(ctx context.Context, a *redis.XReadGroupArgs) *redis.XStreamSliceCmd

+ XTrim func(ctx context.Context, key string, maxLen int64) *redis.IntCmd

+ ZAdd func(ctx context.Context, key string, members ...*redis.Z) *redis.IntCmd

+ ZRangeByScore func(ctx context.Context, key string, opt *redis.ZRangeBy) *redis.StringSliceCmd

+ ZRem func(ctx context.Context, key string, members ...interface{}) *redis.IntCmd

+ type Stats struct

+ Fails uint32

+ Name string

+ Processed uint32

+ Retries uint32

+ type Task struct

+ func NewTask(opt *TaskOptions) *Task

+ func RegisterTask(opt *TaskOptions) (*Task, error)

+ func (t *Task) HandleMessage(msg *Message) error

+ func (t *Task) Name() string

+ func (t *Task) RetryLimit() int

+ type TaskMap struct

+ var Tasks TaskMap

+ func (s *TaskMap) LoadTask(name string) (*Task, error)

+ func (s *TaskMap) RegisterTasks(opts *TaskOptions) (*Task, error)

+ type TaskOptions struct

+ Handler interface{}

+ Name string

+ RetryLimit int

+ type Worker struct

+ func NewWorker(brokers []Broker) *Worker

+ func (w *Worker) AddBroker(ctx context.Context, broker Broker)

+ func (w *Worker) Brokers() []Broker

+ func (w *Worker) Start(ctx context.Context) error

+ func (w *Worker) Stats() []*Stats

+ func (w *Worker) Stop() error

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL