forked from gocelery/gocelery
-
Notifications
You must be signed in to change notification settings - Fork 0
/
redis_backend.go
53 lines (47 loc) · 1.25 KB
/
redis_backend.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
package gocelery
import (
"encoding/json"
"fmt"
"github.com/gomodule/redigo/redis"
)
// RedisCeleryBackend is CeleryBackend for Redis
type RedisCeleryBackend struct {
*redis.Pool
}
// NewRedisCeleryBackend creates new RedisCeleryBackend
func NewRedisCeleryBackend(uri string) *RedisCeleryBackend {
return &RedisCeleryBackend{
Pool: NewRedisPool(uri),
}
}
// GetResult calls API to get asynchronous result
// Should be called by AsyncResult
func (cb *RedisCeleryBackend) GetResult(taskID string) (*ResultMessage, error) {
//"celery-task-meta-" + taskID
conn := cb.Get()
defer conn.Close()
val, err := conn.Do("GET", fmt.Sprintf("celery-task-meta-%s", taskID))
if err != nil {
return nil, err
}
if val == nil {
return nil, fmt.Errorf("result not available")
}
var resultMessage ResultMessage
err = json.Unmarshal(val.([]byte), &resultMessage)
if err != nil {
return nil, err
}
return &resultMessage, nil
}
// SetResult pushes result back into backend
func (cb *RedisCeleryBackend) SetResult(taskID string, result *ResultMessage) error {
resBytes, err := json.Marshal(result)
if err != nil {
return err
}
conn := cb.Get()
defer conn.Close()
_, err = conn.Do("SETEX", fmt.Sprintf("celery-task-meta-%s", taskID), 86400, resBytes)
return err
}