taskq

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

type ConsumerHook interface {
	BeforeProcessJob(context.Context, *ProcessJobEvent) context.Context
	AfterProcessJob(context.Context, *ProcessJobEvent)
}

type ConsumerStats struct {
	NumWorker  uint32
	NumFetcher uint32

	BufferSize uint32
	Buffered   uint32

	InFlight  uint32
	Processed uint32
	Retries   uint32
	Fails     uint32
}

type Delayer interface {
	Delay() time.Duration
}

type Factory interface {
	RegisterQueue(*QueueConfig) Queue
	Range(func(Queue) bool)
	StartConsumers(context.Context) error
	StopConsumers() error
	Close() error
}

type Handler interface {
	HandleJob(ctx context.Context, msg *Job) error
}

type HandlerFunc func(*Job) error

type Job struct {
	// SQS/IronMQ message id.
	ID string `msgpack:"1,omitempty,alias:ID"`

	// Optional name for the message. Jobs with the same name
	// are processed only once.
	Name string `msgpack:"-"`

	// Delay specifies the duration the queue must wait
	// before executing the message.
	Delay time.Duration `msgpack:"-"`

	// Args passed to the handler.
	Args []interface{} `msgpack:"-"`

	// Binary representation of the args.
	ArgsCompression string `msgpack:"2,omitempty,alias:ArgsCompression"`
	ArgsBin         []byte `msgpack:"3,alias:ArgsBin"`

	// SQS/IronMQ reservation id that is used to release/delete the message.
	ReservationID string `msgpack:"-"`

	// The number of times the message has been reserved or released.
	ReservedCount int `msgpack:"4,omitempty,alias:ReservedCount"`

	TaskName string `msgpack:"5,alias:TaskName"`
	Err      error  `msgpack:"-"`
	// contains filtered or unexported fields
}

type ProcessJobEvent struct {
	Job       *Job
	StartTime time.Time

	Stash map[interface{}]interface{}
}

type Queue interface {
	fmt.Stringer
	Name() string
	Options() *QueueConfig
	Consumer() QueueConsumer

	Len(ctx context.Context) (int, error)
	AddJob(ctx context.Context, msg *Job) error
	ReserveN(ctx context.Context, n int, waitTimeout time.Duration) ([]Job, error)
	Release(ctx context.Context, msg *Job) error
	Delete(ctx context.Context, msg *Job) error
	Purge(ctx context.Context) error
	Close() error
	CloseTimeout(timeout time.Duration) error
}

type QueueConfig struct {
	// Queue name.
	Name string

	NumWorker int
	// Global limit of concurrently running workers across all servers.
	// Overrides NumWorker.
	WorkerLimit int
	// Maximum number of goroutines fetching messages.
	// Default is 8 * number of CPUs.
	NumFetcher int

	// Number of messages reserved by a fetcher in the queue in one request.
	// Default is 10 messages.
	ReservationSize int
	// Time after which the reserved message is returned to the queue.
	// Default is 5 minutes.
	ReservationTimeout time.Duration
	// Time that a long polling receive call waits for a message to become
	// available before returning an empty response.
	// Default is 10 seconds.
	WaitTimeout time.Duration
	// Size of the buffer where reserved messages are stored.
	// Default is the same as ReservationSize.
	BufferSize int

	// Number of consecutive failures after which queue processing is paused.
	// Default is 100 failures.
	PauseErrorsThreshold int

	// Processing rate limit.
	RateLimit redis_rate.Limit

	// Optional rate limiter. The default is to use Redis.
	RateLimiter *redis_rate.Limiter

	// Redis client that is used for storing metadata.
	Redis Redis

	// Optional storage interface. The default is to use Redis.
	Storage Storage

	// Optional message handler. The default is the global Tasks registry.
	Handler Handler

	// ConsumerIdleTimeout Time after which the consumer need to be deleted.
	// Default is 6 hour
	ConsumerIdleTimeout time.Duration

	// SchedulerBackoffTime is the time of backoff for the scheduler(
	// Scheduler was designed to clean zombie Consumer and requeue pending msgs, and so on.
	// Default is randomly between 1~1.5s
	// We can change it to a bigger value so that it won't slowdown the redis when using redis queue.
	// It will be between SchedulerBackoffTime and SchedulerBackoffTime+250ms.
	SchedulerBackoffTime time.Duration
	// contains filtered or unexported fields
}

type QueueConsumer interface {
	// AddHook adds a hook into message processing.
	AddHook(hook ConsumerHook)
	Queue() Queue
	Options() *QueueConfig
	Len() int
	// Stats returns processor stats.
	Stats() *ConsumerStats
	AddJob(ctx context.Context, job *Job) error
	// Start starts consuming messages in the queue.
	Start(ctx context.Context) error
	// Stop is StopTimeout with 30 seconds timeout.
	Stop() error
	// StopTimeout waits workers for timeout duration to finish processing current
	// messages and stops workers.
	StopTimeout(timeout time.Duration) error
	// ProcessAll starts workers to process messages in the queue and then stops
	// them when all messages are processed.
	ProcessAll(ctx context.Context) error
	// ProcessOne processes at most one message in the queue.
	ProcessOne(ctx context.Context) error
	// Process is low-level API to process message bypassing the internal queue.
	Process(ctx context.Context, msg *Job) error
	Put(ctx context.Context, msg *Job)
	// Purge discards messages from the internal queue.
	Purge(ctx context.Context) error
	String() string
}

type Redis interface {
	redis.Scripter

	Del(ctx context.Context, keys ...string) *redis.IntCmd
	SetNX(ctx context.Context, key string, value interface{}, expiration time.Duration) *redis.BoolCmd
	Pipelined(ctx context.Context, fn func(pipe redis.Pipeliner) error) ([]redis.Cmder, error)
}

type Storage interface {
	Exists(ctx context.Context, key string) bool
}

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

type TaskConfig struct {
	// Function called to process a message.
	// There are three permitted types of signature:
	// 1. A zero-argument function
	// 2. A function whose arguments are assignable in type from those which are passed in the message
	// 3. A function which takes a single `*Job` argument
	// The handler function may also optionally take a Context as a first argument and may optionally return an error.
	// If the handler takes a Context, when it is invoked it will be passed the same Context as that which was passed to
	// `StartConsumer`. If the handler returns a non-nil error the message processing will fail and will be retried/.
	Handler interface{}
	// Function called to process failed message after the specified number of retries have all failed.
	// The FallbackHandler accepts the same types of function as the Handler.
	FallbackHandler interface{}

	// Optional function used by Consumer with defer statement
	// to recover from panics.
	DeferFunc func()

	// Number of tries/releases after which the message fails permanently
	// and is deleted.
	// Default is 64 retries.
	RetryLimit int
	// Minimum backoff time between retries.
	// Default is 30 seconds.
	MinBackoff time.Duration
	// Maximum backoff time between retries.
	// Default is 30 minutes.
	MaxBackoff time.Duration
	// contains filtered or unexported fields
}

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