Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security Policy
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

taskq

package module

Go to main page

Versions in this module

v3

v3.2.11

Mar 4, 2022

v3.2.10

Mar 3, 2022

Changes in this version

+ func NewConsumerConfig(numFetcher, numWorker int32) *consumerConfig

v3.2.9

Feb 21, 2022

Changes in this version

+ var ErrAsyncTask = errors.New("taskq: async task")

+ var ErrDuplicate = errors.New("taskq: message with such name already exists")

+ func SetLogger(logger *log.Logger)

+ func SetUnknownTaskOptions(opt *TaskOptions)

+ func Version() string

+ type Consumer struct

+ func NewConsumer(q Queue) *Consumer

+ func StartConsumer(ctx context.Context, q Queue) *Consumer

+ func (c *Consumer) Add(msg *Message) error

+ func (c *Consumer) AddHook(hook ConsumerHook)

+ func (c *Consumer) Len() int

+ func (c *Consumer) Options() *QueueOptions

+ func (c *Consumer) Process(msg *Message) error

+ func (c *Consumer) ProcessAll(ctx context.Context) error

+ func (c *Consumer) ProcessOne(ctx context.Context) error

+ func (c *Consumer) Purge() error

+ func (c *Consumer) Put(msg *Message)

+ func (c *Consumer) Queue() Queue

+ func (c *Consumer) Start(ctx context.Context) error

+ func (c *Consumer) Stats() *ConsumerStats

+ func (c *Consumer) Stop() error

+ func (c *Consumer) StopTimeout(timeout time.Duration) error

+ func (c *Consumer) String() string

+ type ConsumerHook interface

+ AfterProcessMessage func(*ProcessMessageEvent) error

+ BeforeProcessMessage func(*ProcessMessageEvent) error

+ type ConsumerStats struct

+ BufferSize uint32

+ Buffered uint32

+ Fails uint32

+ InFlight uint32

+ NumFetcher uint32

+ NumWorker uint32

+ Processed uint32

+ Retries uint32

+ Timing time.Duration

+ type Delayer interface

+ Delay func() time.Duration

+ type Factory interface

+ Close func() error

+ Range func(func(Queue) bool)

+ RegisterQueue func(*QueueOptions) Queue

+ StartConsumers func(context.Context) error

+ StopConsumers func() error

+ 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{}

+ ArgsBin []byte

+ ArgsCompression string

+ Ctx context.Context

+ Delay time.Duration

+ Err error

+ ID string

+ Name string

+ ReservationID string

+ ReservedCount int

+ TaskName string

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

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

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

+ func (m *Message) OnceInPeriod(period time.Duration, args ...interface{})

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

+ func (m *Message) OnceWithSchedule(tm time.Time)

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

+ func (m *Message) String() string

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

+ type ProcessMessageEvent struct

+ Message *Message

+ StartTime time.Time

+ Stash map[interface{}]interface{}

+ type Queue interface

+ Add func(msg *Message) error

+ Close func() error

+ CloseTimeout func(timeout time.Duration) error

+ Consumer func() QueueConsumer

+ Delete func(msg *Message) error

+ Len func() (int, error)

+ Name func() string

+ Options func() *QueueOptions

+ Purge func() error

+ Release func(msg *Message) error

+ ReserveN func(ctx context.Context, n int, waitTimeout time.Duration) ([]Message, error)

+ type QueueConsumer interface

+ Add func(msg *Message) error

+ AddHook func(hook ConsumerHook)

+ Len func() int

+ Options func() *QueueOptions

+ Process func(msg *Message) error

+ ProcessAll func(ctx context.Context) error

+ ProcessOne func(ctx context.Context) error

+ Purge func() error

+ Put func(msg *Message)

+ Queue func() Queue

+ Start func(ctx context.Context) error

+ Stats func() *ConsumerStats

+ Stop func() error

+ StopTimeout func(timeout time.Duration) error

+ String func() string

+ type QueueOptions struct

+ BufferSize int

+ ConsumerIdleTimeout time.Duration

+ Handler Handler

+ MaxNumFetcher int32

+ MaxNumWorker int32

+ MinNumWorker int32

+ Name string

+ PauseErrorsThreshold int

+ RateLimit redis_rate.Limit

+ RateLimiter *redis_rate.Limiter

+ Redis Redis

+ ReservationSize int

+ ReservationTimeout time.Duration

+ Storage Storage

+ WaitTimeout time.Duration

+ WorkerLimit int32

+ func (opt *QueueOptions) Init()

+ 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

+ 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

+ type Storage interface

+ Exists func(ctx context.Context, key string) bool

+ func NewLocalStorage() Storage

+ type Task struct

+ func RegisterTask(opt *TaskOptions) *Task

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

+ func (t *Task) Name() string

+ func (t *Task) Options() *TaskOptions

+ func (t *Task) String() string

+ func (t *Task) WithArgs(ctx context.Context, args ...interface{}) *Message

+ type TaskMap struct

+ var Tasks TaskMap

+ func (r *TaskMap) Get(name string) *Task

+ func (r *TaskMap) HandleMessage(msg *Message) error

+ func (r *TaskMap) Range(fn func(name string, task *Task) bool)

+ func (r *TaskMap) Register(opt *TaskOptions) (*Task, error)

+ func (r *TaskMap) Reset()

+ func (r *TaskMap) Unregister(task *Task)

+ type TaskOptions struct

+ DeferFunc func()

+ FallbackHandler interface{}

+ Handler interface{}

+ MaxBackoff time.Duration

+ MinBackoff time.Duration

+ Name string

+ RetryLimit int

Other modules containing this package

github.com/frain-dev/taskq