forked from gocelery/gocelery
-
Notifications
You must be signed in to change notification settings - Fork 0
/
amqp_backend.go
159 lines (135 loc) · 3.39 KB
/
amqp_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
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
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
// Copyright (c) 2019 Sick Yoon
// This file is part of gocelery which is released under MIT license.
// See file LICENSE for full license details.
package gocelery
import (
"encoding/json"
"strings"
"time"
"github.com/streadway/amqp"
)
/* AMQPCeleryBackend CeleryBackend for AMQP
The difference between amqpbackend and rpcbackend:
amqpbackend => reply to task_id queue
rpcbackend => reply to celery_backend exchange and route the message to the client
side who have waiting for it through the {oid}_result queue which is binding to
celery_backend exchange.
*/
type AMQPCeleryBackend struct {
*AMQPSession
ExpireDuration time.Duration
}
// NewAMQPCeleryBackend creates new AMQPCeleryBackend
func NewAMQPCeleryBackend(host string) *AMQPCeleryBackend {
session, err := NewAMQPSession(host)
if err != nil {
panic(err)
}
backend := NewAMQPCeleryBackendByAMQPSession(session)
return backend
}
// NewAMQPCeleryBackendByConnAndChannel creates new AMQPCeleryBackend by AMQP connection and channel
func NewAMQPCeleryBackendByAMQPSession(session *AMQPSession) *AMQPCeleryBackend {
backend := &AMQPCeleryBackend{
AMQPSession: session,
ExpireDuration: 24 * time.Hour,
}
return backend
}
// GetResult retrieves result from AMQP queue
func (b *AMQPCeleryBackend) GetResult(taskID string) (*ResultMessage, error) {
queueName := strings.Replace(taskID, "-", "", -1)
args := amqp.Table{"x-expires": int32(b.ExpireDuration.Microseconds())}
_, err := b.QueueDeclare(
queueName, // name
true, // durable
true, // autoDelete
false, // exclusive
false, // noWait
args, // args
)
if err != nil {
return nil, err
}
err = b.ExchangeDeclare(
"default",
"direct",
true,
true,
false,
false,
nil,
)
if err != nil {
return nil, err
}
// open channel temporarily
channel, err := b.Consume(queueName, "", false, false, false, false, nil)
if err != nil {
return nil, err
}
var resultMessage ResultMessage
select {
case delivery := <-channel:
deliveryAck(delivery)
if err = json.Unmarshal(delivery.Body, &resultMessage); err != nil {
return nil, err
}
default:
return nil, ResultNotAvailableYet
}
return &resultMessage, nil
}
// SetResult sets result back to AMQP queue
func (b *AMQPCeleryBackend) SetResult(taskID string, result *ResultMessage) error {
result.ID = taskID
//queueName := taskID
queueName := strings.Replace(taskID, "-", "", -1)
// autodelete is automatically set to true by python
// (406) PRECONDITION_FAILED - inequivalent arg 'durable' for queue 'bc58c0d895c7421eb7cb2b9bbbd8b36f' in vhost '/': received 'true' but current is 'false'
args := amqp.Table{"x-expires": int32(b.ExpireDuration.Microseconds())}
_, err := b.QueueDeclare(
queueName, // name
true, // durable
true, // autoDelete
false, // exclusive
false, // noWait
args, // args
)
if err != nil {
return err
}
err = b.ExchangeDeclare(
"default",
"direct",
true,
true,
false,
false,
nil,
)
if err != nil {
return err
}
resBytes, err := json.Marshal(result)
if err != nil {
return err
}
message := amqp.Publishing{
DeliveryMode: amqp.Persistent,
Timestamp: time.Now(),
ContentType: "application/json",
Body: resBytes,
}
return b.Publish(
"",
queueName,
false,
false,
message,
)
}
func (b *AMQPCeleryBackend) Init(string) error {
// amqp backend do nothing through init phase
return nil
}