mirror of
https://github.com/hibiken/asynq.git
synced 2024-12-25 23:32:17 +08:00
Move client and launcher to its own files
This commit is contained in:
parent
85a04cbabb
commit
e9069bfb47
92
asynq.go
92
asynq.go
@ -8,21 +8,9 @@ TODOs:
|
|||||||
- [P2] Web UI
|
- [P2] Web UI
|
||||||
*/
|
*/
|
||||||
|
|
||||||
import (
|
|
||||||
"sync"
|
|
||||||
"time"
|
|
||||||
|
|
||||||
"github.com/go-redis/redis/v7"
|
|
||||||
)
|
|
||||||
|
|
||||||
// Max retry count by default
|
// Max retry count by default
|
||||||
const defaultMaxRetry = 25
|
const defaultMaxRetry = 25
|
||||||
|
|
||||||
// Client is an interface for scheduling tasks.
|
|
||||||
type Client struct {
|
|
||||||
rdb *rdb
|
|
||||||
}
|
|
||||||
|
|
||||||
// Task represents a task to be performed.
|
// Task represents a task to be performed.
|
||||||
type Task struct {
|
type Task struct {
|
||||||
// Type indicates the kind of the task to be performed.
|
// Type indicates the kind of the task to be performed.
|
||||||
@ -59,83 +47,3 @@ type RedisOpt struct {
|
|||||||
Addr string
|
Addr string
|
||||||
Password string
|
Password string
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewClient creates and returns a new client.
|
|
||||||
func NewClient(opt *RedisOpt) *Client {
|
|
||||||
client := redis.NewClient(&redis.Options{Addr: opt.Addr, Password: opt.Password})
|
|
||||||
return &Client{rdb: newRDB(client)}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Process enqueues the task to be performed at a given time.
|
|
||||||
func (c *Client) Process(task *Task, executeAt time.Time) error {
|
|
||||||
msg := &taskMessage{
|
|
||||||
Type: task.Type,
|
|
||||||
Payload: task.Payload,
|
|
||||||
Queue: "default",
|
|
||||||
Retry: defaultMaxRetry,
|
|
||||||
}
|
|
||||||
return c.enqueue(msg, executeAt)
|
|
||||||
}
|
|
||||||
|
|
||||||
// enqueue pushes a given task to the specified queue.
|
|
||||||
func (c *Client) enqueue(msg *taskMessage, executeAt time.Time) error {
|
|
||||||
if time.Now().After(executeAt) {
|
|
||||||
return c.rdb.push(msg)
|
|
||||||
}
|
|
||||||
return c.rdb.zadd(scheduled, float64(executeAt.Unix()), msg)
|
|
||||||
}
|
|
||||||
|
|
||||||
//-------------------- Launcher --------------------
|
|
||||||
|
|
||||||
// Launcher starts the manager and poller.
|
|
||||||
type Launcher struct {
|
|
||||||
// running indicates whether manager and poller are both running.
|
|
||||||
running bool
|
|
||||||
mu sync.Mutex
|
|
||||||
|
|
||||||
poller *poller
|
|
||||||
|
|
||||||
manager *manager
|
|
||||||
}
|
|
||||||
|
|
||||||
// NewLauncher creates and returns a new Launcher.
|
|
||||||
func NewLauncher(poolSize int, opt *RedisOpt) *Launcher {
|
|
||||||
client := redis.NewClient(&redis.Options{Addr: opt.Addr, Password: opt.Password})
|
|
||||||
rdb := newRDB(client)
|
|
||||||
poller := newPoller(rdb, 5*time.Second, []string{scheduled, retry})
|
|
||||||
manager := newManager(rdb, poolSize, nil)
|
|
||||||
return &Launcher{
|
|
||||||
poller: poller,
|
|
||||||
manager: manager,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TaskHandler handles a given task and report any error.
|
|
||||||
type TaskHandler func(*Task) error
|
|
||||||
|
|
||||||
// Start starts the manager and poller.
|
|
||||||
func (l *Launcher) Start(handler TaskHandler) {
|
|
||||||
l.mu.Lock()
|
|
||||||
defer l.mu.Unlock()
|
|
||||||
if l.running {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
l.running = true
|
|
||||||
l.manager.handler = handler
|
|
||||||
|
|
||||||
l.poller.start()
|
|
||||||
l.manager.start()
|
|
||||||
}
|
|
||||||
|
|
||||||
// Stop stops both manager and poller.
|
|
||||||
func (l *Launcher) Stop() {
|
|
||||||
l.mu.Lock()
|
|
||||||
defer l.mu.Unlock()
|
|
||||||
if !l.running {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
l.running = false
|
|
||||||
|
|
||||||
l.poller.terminate()
|
|
||||||
l.manager.terminate()
|
|
||||||
}
|
|
||||||
|
37
client.go
Normal file
37
client.go
Normal file
@ -0,0 +1,37 @@
|
|||||||
|
package asynq
|
||||||
|
|
||||||
|
import (
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/go-redis/redis/v7"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Client is an interface for scheduling tasks.
|
||||||
|
type Client struct {
|
||||||
|
rdb *rdb
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewClient creates and returns a new client.
|
||||||
|
func NewClient(opt *RedisOpt) *Client {
|
||||||
|
client := redis.NewClient(&redis.Options{Addr: opt.Addr, Password: opt.Password})
|
||||||
|
return &Client{rdb: newRDB(client)}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Process enqueues the task to be performed at a given time.
|
||||||
|
func (c *Client) Process(task *Task, executeAt time.Time) error {
|
||||||
|
msg := &taskMessage{
|
||||||
|
Type: task.Type,
|
||||||
|
Payload: task.Payload,
|
||||||
|
Queue: "default",
|
||||||
|
Retry: defaultMaxRetry,
|
||||||
|
}
|
||||||
|
return c.enqueue(msg, executeAt)
|
||||||
|
}
|
||||||
|
|
||||||
|
// enqueue pushes a given task to the specified queue.
|
||||||
|
func (c *Client) enqueue(msg *taskMessage, executeAt time.Time) error {
|
||||||
|
if time.Now().After(executeAt) {
|
||||||
|
return c.rdb.push(msg)
|
||||||
|
}
|
||||||
|
return c.rdb.zadd(scheduled, float64(executeAt.Unix()), msg)
|
||||||
|
}
|
61
launcher.go
Normal file
61
launcher.go
Normal file
@ -0,0 +1,61 @@
|
|||||||
|
package asynq
|
||||||
|
|
||||||
|
import (
|
||||||
|
"sync"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/go-redis/redis/v7"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Launcher starts the manager and poller.
|
||||||
|
type Launcher struct {
|
||||||
|
// running indicates whether manager and poller are both running.
|
||||||
|
running bool
|
||||||
|
mu sync.Mutex
|
||||||
|
|
||||||
|
poller *poller
|
||||||
|
|
||||||
|
manager *manager
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewLauncher creates and returns a new Launcher.
|
||||||
|
func NewLauncher(poolSize int, opt *RedisOpt) *Launcher {
|
||||||
|
client := redis.NewClient(&redis.Options{Addr: opt.Addr, Password: opt.Password})
|
||||||
|
rdb := newRDB(client)
|
||||||
|
poller := newPoller(rdb, 5*time.Second, []string{scheduled, retry})
|
||||||
|
manager := newManager(rdb, poolSize, nil)
|
||||||
|
return &Launcher{
|
||||||
|
poller: poller,
|
||||||
|
manager: manager,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TaskHandler handles a given task and report any error.
|
||||||
|
type TaskHandler func(*Task) error
|
||||||
|
|
||||||
|
// Start starts the manager and poller.
|
||||||
|
func (l *Launcher) Start(handler TaskHandler) {
|
||||||
|
l.mu.Lock()
|
||||||
|
defer l.mu.Unlock()
|
||||||
|
if l.running {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
l.running = true
|
||||||
|
l.manager.handler = handler
|
||||||
|
|
||||||
|
l.poller.start()
|
||||||
|
l.manager.start()
|
||||||
|
}
|
||||||
|
|
||||||
|
// Stop stops both manager and poller.
|
||||||
|
func (l *Launcher) Stop() {
|
||||||
|
l.mu.Lock()
|
||||||
|
defer l.mu.Unlock()
|
||||||
|
if !l.running {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
l.running = false
|
||||||
|
|
||||||
|
l.poller.terminate()
|
||||||
|
l.manager.terminate()
|
||||||
|
}
|
Loading…
Reference in New Issue
Block a user