async_task

Documentation

Index

• Constants
• type AsyncTask
  • func NewAsyncTask(opt AsyncTaskOption) (*AsyncTask, error)
• type AsyncTaskOption
• type Client
  • func NewClient(redisAddr string, db int) (*Client, error)
  • func (s *Client) Execute(ctx context.Context, task *Task, opts ...Option) (*TaskInfo, error)
• type HandlerFunc
• type Option
  • func ExecuteAtOpt(t time.Time) Option
  • func MaxRetriesOpt(n int) Option
  • func QueueOpt(queue string) Option
  • func TaskIDOpt(id string) Option
• type OptionType
• type Scheduler
  • func NewScheduler(redisAddr string, poolSize int, db int) (*Scheduler, error)
  • func (s *Scheduler) RegisterHandler(workerName string, f func(ctx context.Context, task *Task) error)
  • func (s *Scheduler) Start() error
  • func (s *Scheduler) Stop()
• type Task
  • func NewTask(typeName string, payload map[string]interface{}, opt ...Option) *Task
  • func (t Task) Payload() map[string]interface{}
  • func (t Task) TypeName() string
• type TaskInfo

Constants

const (
	QueueImmediate = "queue:immediate"
	QueueDelayed   = "queue:delayed"
	QueueRetry     = "queue:retry"
)

Types

type AsyncTask

type AsyncTask struct {
	Scheduler *Scheduler
	Client    *Client
}

func NewAsyncTask

func NewAsyncTask(opt AsyncTaskOption) (*AsyncTask, error)

type AsyncTaskOption

type AsyncTaskOption struct {
	RedisAddr         string
	RedisPort         int64
	RedisPassword     string
	RedisDB           int
	RedisMaxRetries   int
	RedisPoolSize     int
	AsyncTaskPoolSize int
}

type Client

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

func NewClient

func NewClient(redisAddr string, db int) (*Client, error)

func (*Client) Execute

func (s *Client) Execute(ctx context.Context, task *Task, opts ...Option) (*TaskInfo, error)

type HandlerFunc

type HandlerFunc func(ctx context.Context, task *Task) error

type Option

type Option interface {
	Type() OptionType
	Value() interface{}
}

func ExecuteAtOpt

func ExecuteAtOpt(t time.Time) Option

func MaxRetriesOpt

func MaxRetriesOpt(n int) Option

func QueueOpt

func QueueOpt(queue string) Option

func TaskIDOpt

func TaskIDOpt(id string) Option

type OptionType

type OptionType int

const (
	MaxRetries OptionType = iota + 1
	ExecuteAt
	TaskID
	Queue
)

type Scheduler

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

func NewScheduler

func NewScheduler(redisAddr string, poolSize int, db int) (*Scheduler, error)

func (*Scheduler) RegisterHandler

func (s *Scheduler) RegisterHandler(workerName string, f func(ctx context.Context, task *Task) error)

func (*Scheduler) Start

func (s *Scheduler) Start() error

启动服务

func (*Scheduler) Stop

func (s *Scheduler) Stop()

关闭服务

type Task

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

func NewTask

func NewTask(typeName string, payload map[string]interface{}, opt ...Option) *Task

func (Task) Payload

func (t Task) Payload() map[string]interface{}

func (Task) TypeName

func (t Task) TypeName() string

type TaskInfo

type TaskInfo struct {
	ID           string                 `json:"id"`
	TypeName     string                 `json:"type_name"`
	QueueName    string                 `json:"queue_name"`
	Payload      map[string]interface{} `json:"payload"`
	MaxRetries   int                    `json:"max_retries"`
	RetryCount   int                    `json:"retry_count"`
	ExecuteAt    time.Time              `json:"execute_at"`
	FailedReason string                 `json:"failed_reason"`
}