2
0
mirror of https://github.com/hibiken/asynq.git synced 2024-12-25 23:32:17 +08:00
asynq/background.go

104 lines
2.2 KiB
Go
Raw Normal View History

2019-11-24 07:22:43 +08:00
package asynq
import (
2019-11-27 22:33:04 +08:00
"fmt"
"log"
2019-11-24 07:44:42 +08:00
"os"
"os/signal"
2019-11-24 07:22:43 +08:00
"sync"
"time"
2019-12-04 13:01:26 +08:00
"github.com/hibiken/asynq/internal/rdb"
2019-11-24 07:22:43 +08:00
)
// Background is a top-level entity for the background-task processing.
type Background struct {
mu sync.Mutex
running bool
2019-11-24 07:22:43 +08:00
2019-12-04 13:01:26 +08:00
rdb *rdb.RDB
2019-11-24 07:22:43 +08:00
poller *poller
processor *processor
}
// NewBackground returns a new Background instance.
2019-12-04 11:43:01 +08:00
func NewBackground(numWorkers int, config *RedisConfig) *Background {
2019-12-04 13:01:26 +08:00
r := rdb.NewRDB(newRedisClient(config))
2019-12-04 23:14:37 +08:00
poller := newPoller(r, 5*time.Second)
2019-12-04 13:01:26 +08:00
processor := newProcessor(r, numWorkers, nil)
2019-11-24 07:22:43 +08:00
return &Background{
2019-12-04 13:01:26 +08:00
rdb: r,
2019-11-24 07:22:43 +08:00
poller: poller,
processor: processor,
}
}
// A Handler processes a task.
//
// ProcessTask should return nil if the processing of a task
// is successful.
//
// If ProcessTask return a non-nil error or panics, the task
// will be retried after delay.
type Handler interface {
ProcessTask(*Task) error
}
// The HandlerFunc type is an adapter to allow the use of
// ordinary functions as a Handler. If f is a function
// with the appropriate signature, HandlerFunc(f) is a
// Handler that calls f.
type HandlerFunc func(*Task) error
// ProcessTask calls fn(task)
func (fn HandlerFunc) ProcessTask(task *Task) error {
return fn(task)
}
2019-11-24 07:22:43 +08:00
2019-11-24 07:44:42 +08:00
// Run starts the background-task processing and blocks until
// an os signal to exit the program is received. Once it receives
// a signal, it gracefully shuts down all pending workers and other
// goroutines to process the tasks.
func (bg *Background) Run(handler Handler) {
2019-11-24 07:44:42 +08:00
bg.start(handler)
defer bg.stop()
// Wait for a signal to exit.
sigs := make(chan os.Signal, 1)
signal.Notify(sigs, os.Interrupt, os.Kill)
<-sigs
fmt.Println()
log.Println("[INFO] Starting graceful shutdown...")
2019-11-24 07:44:42 +08:00
}
// starts the background-task processing.
func (bg *Background) start(handler Handler) {
2019-11-24 07:22:43 +08:00
bg.mu.Lock()
defer bg.mu.Unlock()
if bg.running {
return
}
2019-11-24 07:22:43 +08:00
bg.running = true
bg.processor.handler = handler
bg.poller.start()
bg.processor.start()
}
2019-11-24 07:44:42 +08:00
// stops the background-task processing.
func (bg *Background) stop() {
2019-11-24 07:22:43 +08:00
bg.mu.Lock()
defer bg.mu.Unlock()
if !bg.running {
return
}
bg.poller.terminate()
bg.processor.terminate()
2019-12-04 13:01:26 +08:00
bg.rdb.Close()
bg.processor.handler = nil
bg.running = false
2019-11-24 07:22:43 +08:00
}