-
Notifications
You must be signed in to change notification settings - Fork 0
/
sender.go
128 lines (107 loc) · 3.07 KB
/
sender.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
package messageq
import (
"fmt"
"sync"
"github.com/prometheus/common/log"
desc "github.com/iterum-provenance/iterum-go/descriptors"
"github.com/iterum-provenance/iterum-go/transmit"
"github.com/iterum-provenance/iterum-go/util"
"github.com/streadway/amqp"
)
// Sender is the structure that listens to a channel and redirects messages to rabbitMQ
type Sender struct {
toSend <-chan transmit.Serializable // data.RemoteFragmentDesc
ToLineate chan<- transmit.Serializable // desc.RemoteFragmentDesc
TargetQueue string
BrokerURL string
fragments int
}
// NewSender creates a new sender which receives messages from a channel and sends them on the message queue.
func NewSender(toSend, toLineate chan transmit.Serializable, brokerURL, targetQueue string) (sender Sender, err error) {
sender = Sender{
toSend,
toLineate,
targetQueue,
brokerURL,
0,
}
return
}
// DeclareQueue defines an amqp Queue
func (sender Sender) DeclareQueue(queueName string, ch *amqp.Channel) amqp.Queue {
q, err := ch.QueueDeclare(
queueName, // name
false, // durable
false, // delete when unused
false, // exclusive
false, // no-wait
nil, // arguments
)
util.Ensure(err, fmt.Sprintf("Created queue %v", queueName))
return q
}
// StartBlocking listens to the channel, and send remoteFragments to the message queue on the OUTPUT_QUEUE queue.
func (sender Sender) StartBlocking() {
log.Infof("Connecting to %s.\n", sender.BrokerURL)
conn, err := amqp.Dial(sender.BrokerURL)
util.Ensure(err, "Connected to RabbitMQ")
defer conn.Close()
ch, err := conn.Channel()
util.Ensure(err, "Opened channel")
defer ch.Close()
queues := make(map[string]amqp.Queue)
targetq := sender.DeclareQueue(sender.TargetQueue, ch)
queues[sender.TargetQueue] = targetq
for {
remoteFragMsg, ok := <-sender.toSend
if !ok {
break
}
remoteFragment := *remoteFragMsg.(*desc.RemoteFragmentDesc)
var q amqp.Queue
if remoteFragment.Metadata.TargetQueue != nil {
queueName := *remoteFragment.Metadata.TargetQueue
if _, ok := queues[queueName]; !ok {
queues[queueName] = sender.DeclareQueue(queueName, ch)
}
q = queues[queueName]
} else {
q = targetq
}
log.Debugf("Sending %v to queue '%v'\n", remoteFragment, q.Name)
mqFragment := newFragmentDesc(remoteFragment)
body, err := mqFragment.Serialize()
if err != nil {
log.Errorln(err)
}
sender.ToLineate <- &remoteFragment
err = ch.Publish(
"", // exchange
q.Name, // routing key
false, // mandatory
false,
amqp.Publishing{
DeliveryMode: amqp.Persistent,
ContentType: "application/json",
Body: body,
})
if err != nil {
log.Errorln(err)
}
sender.fragments++
}
sender.Stop()
}
// Start asychronously calls StartBlocking via Gorouting
func (sender Sender) Start(wg *sync.WaitGroup) {
wg.Add(1)
go func() {
defer wg.Done()
sender.StartBlocking()
}()
}
// Stop finishes up and notifies the user of its progress
func (sender Sender) Stop() {
log.Infof("MQSender finishing up, published %v messages\n", sender.fragments)
close(sender.ToLineate)
}