forked from filswan/go-computing-provider
/
celery_server.go
86 lines (73 loc) · 2.11 KB
/
celery_server.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
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
package computing
import (
"sync"
"time"
"github.com/swanchain/go-computing-provider/conf"
"github.com/filswan/go-mcs-sdk/mcs/api/common/logs"
"github.com/gocelery/gocelery"
"github.com/gomodule/redigo/redis"
)
var redisPool *redis.Pool
var celeryService *CeleryService
var celeryOnce sync.Once
type CeleryService struct {
cli *gocelery.CeleryClient
}
func newRedisPool(url string, password string) *redis.Pool {
redisPool = &redis.Pool{
MaxIdle: 5, // maximum number of idle connections in the pool
MaxActive: 0, // maximum number of connections allocated by the pool at a given time
IdleTimeout: 240 * time.Second, // close connections after remaining idle for this duration
Dial: func() (redis.Conn, error) {
var conn redis.Conn
var err error
if password != "" {
conn, err = redis.DialURL(url, redis.DialPassword(password))
} else {
conn, err = redis.DialURL(url)
}
if err != nil {
return nil, err
}
return conn, err
},
TestOnBorrow: func(c redis.Conn, t time.Time) error {
_, err := c.Do("PING")
return err
},
}
return redisPool
}
func GetRedisClient() redis.Conn {
newRedisPool(conf.GetConfig().API.RedisUrl, conf.GetConfig().API.RedisPassword)
return redisPool.Get()
}
func NewCeleryService() *CeleryService {
celeryOnce.Do(
func() {
redisPool := newRedisPool(conf.GetConfig().API.RedisUrl, conf.GetConfig().API.RedisPassword)
celeryClient, err := gocelery.NewCeleryClient(
gocelery.NewRedisBroker(redisPool),
gocelery.NewRedisBackend(redisPool),
10)
if err != nil {
logs.GetLogger().Fatalf("Failed init celery service, error: %+v", err)
}
celeryService = &CeleryService{
cli: celeryClient,
}
})
return celeryService
}
func (s *CeleryService) RegisterTask(taskName string, task interface{}) {
s.cli.Register(taskName, task)
}
func (s *CeleryService) DelayTask(taskName string, params ...interface{}) (*gocelery.AsyncResult, error) {
return s.cli.Delay(taskName, params...)
}
func (s *CeleryService) Start() {
s.cli.StartWorker()
}
func (s *CeleryService) Stop() {
s.cli.StopWorker()
}