mirror of
https://github.com/hibiken/asynq.git
synced 2024-11-10 11:31:58 +08:00
Rename Launcher to Background
This commit is contained in:
parent
847d724985
commit
e19c45cff3
61
background.go
Normal file
61
background.go
Normal file
@ -0,0 +1,61 @@
|
|||||||
|
package asynq
|
||||||
|
|
||||||
|
import (
|
||||||
|
"sync"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/go-redis/redis/v7"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Background is a top-level entity for the background-task processing.
|
||||||
|
type Background struct {
|
||||||
|
// running indicates whether processor and poller are both running.
|
||||||
|
running bool
|
||||||
|
mu sync.Mutex
|
||||||
|
|
||||||
|
poller *poller
|
||||||
|
processor *processor
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewBackground returns a new Background instance.
|
||||||
|
func NewBackground(numWorkers int, opt *RedisOpt) *Background {
|
||||||
|
client := redis.NewClient(&redis.Options{Addr: opt.Addr, Password: opt.Password})
|
||||||
|
rdb := newRDB(client)
|
||||||
|
poller := newPoller(rdb, 5*time.Second, []string{scheduled, retry})
|
||||||
|
processor := newProcessor(rdb, numWorkers, nil)
|
||||||
|
return &Background{
|
||||||
|
poller: poller,
|
||||||
|
processor: processor,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TaskHandler handles a given task and reports any error.
|
||||||
|
type TaskHandler func(*Task) error
|
||||||
|
|
||||||
|
// Start starts the background-task processing.
|
||||||
|
func (bg *Background) Start(handler TaskHandler) {
|
||||||
|
bg.mu.Lock()
|
||||||
|
defer bg.mu.Unlock()
|
||||||
|
if bg.running {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
bg.running = true
|
||||||
|
bg.processor.handler = handler
|
||||||
|
|
||||||
|
bg.poller.start()
|
||||||
|
bg.processor.start()
|
||||||
|
}
|
||||||
|
|
||||||
|
// Stop stops the background-task processing.
|
||||||
|
func (bg *Background) Stop() {
|
||||||
|
bg.mu.Lock()
|
||||||
|
defer bg.mu.Unlock()
|
||||||
|
if !bg.running {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
bg.running = false
|
||||||
|
bg.processor.handler = nil
|
||||||
|
|
||||||
|
bg.poller.terminate()
|
||||||
|
bg.processor.terminate()
|
||||||
|
}
|
61
launcher.go
61
launcher.go
@ -1,61 +0,0 @@
|
|||||||
package asynq
|
|
||||||
|
|
||||||
import (
|
|
||||||
"sync"
|
|
||||||
"time"
|
|
||||||
|
|
||||||
"github.com/go-redis/redis/v7"
|
|
||||||
)
|
|
||||||
|
|
||||||
// Launcher starts the processor and poller.
|
|
||||||
type Launcher struct {
|
|
||||||
// running indicates whether processor and poller are both running.
|
|
||||||
running bool
|
|
||||||
mu sync.Mutex
|
|
||||||
|
|
||||||
poller *poller
|
|
||||||
processor *processor
|
|
||||||
}
|
|
||||||
|
|
||||||
// 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})
|
|
||||||
processor := newProcessor(rdb, poolSize, nil)
|
|
||||||
return &Launcher{
|
|
||||||
poller: poller,
|
|
||||||
processor: processor,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TaskHandler handles a given task and report any error.
|
|
||||||
type TaskHandler func(*Task) error
|
|
||||||
|
|
||||||
// Start starts the processor and poller.
|
|
||||||
func (l *Launcher) Start(handler TaskHandler) {
|
|
||||||
l.mu.Lock()
|
|
||||||
defer l.mu.Unlock()
|
|
||||||
if l.running {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
l.running = true
|
|
||||||
l.processor.handler = handler
|
|
||||||
|
|
||||||
l.poller.start()
|
|
||||||
l.processor.start()
|
|
||||||
}
|
|
||||||
|
|
||||||
// Stop stops both processor and poller.
|
|
||||||
func (l *Launcher) Stop() {
|
|
||||||
l.mu.Lock()
|
|
||||||
defer l.mu.Unlock()
|
|
||||||
if !l.running {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
l.running = false
|
|
||||||
l.processor.handler = nil
|
|
||||||
|
|
||||||
l.poller.terminate()
|
|
||||||
l.processor.terminate()
|
|
||||||
}
|
|
Loading…
Reference in New Issue
Block a user