/
queue.go
44 lines (37 loc) · 924 Bytes
/
queue.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
package main
import (
"github.com/hibiken/asynq"
)
type (
// taskHandler is an interface for task handlers.
taskHandler interface {
Register(*asynq.ServeMux)
}
)
// setupQueue creates a new queue client and registers task handlers.
func runQueueServer(redisConnOpt asynq.RedisConnOpt, log asynq.Logger, handlers ...taskHandler) func() error {
return func() error {
// Setup asynq server
srv := asynq.NewServer(
redisConnOpt,
asynq.Config{
Concurrency: workerConcurrency,
Logger: log,
Queues: map[string]int{
queueName: workerConcurrency,
},
},
)
// Run server
return srv.Run(registerQueueHandlers(handlers...))
}
}
// registerQueueHandlers registers handlers for each task type.
func registerQueueHandlers(handlers ...taskHandler) *asynq.ServeMux {
mux := asynq.NewServeMux()
// Register handlers
for _, h := range handlers {
h.Register(mux)
}
return mux
}