-
Notifications
You must be signed in to change notification settings - Fork 2.7k
/
main.go
145 lines (117 loc) · 3.72 KB
/
main.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
package main
import (
"fmt"
"log"
"time"
"github.com/streadway/amqp"
"github.com/instana/golang-sensor"
ot "github.com/opentracing/opentracing-go"
ext "github.com/opentracing/opentracing-go/ext"
)
const (
Service = "Dispatch"
)
var amqpUri string = "amqp://guest:guest@rabbitmq:5672/"
var (
rabbitChan *amqp.Channel
rabbitCloseError chan *amqp.Error
rabbitReady chan bool
)
func connectToRabbitMQ(uri string) *amqp.Connection {
for {
conn, err := amqp.Dial(uri)
if err == nil {
return conn
}
log.Println(err)
log.Printf("Reconnecting to %s\n", uri)
time.Sleep(1 * time.Second)
}
}
func rabbitConnector(uri string) {
var rabbitErr *amqp.Error
for {
rabbitErr = <-rabbitCloseError
if rabbitErr != nil {
log.Printf("Connecting to %s\n", amqpUri)
rabbitConn := connectToRabbitMQ(uri)
rabbitConn.NotifyClose(rabbitCloseError)
var err error
// create mappings here
rabbitChan, err = rabbitConn.Channel()
failOnError(err, "Failed to create channel")
// create exchange
err = rabbitChan.ExchangeDeclare("robot-shop", "direct", true, false, false, false, nil)
failOnError(err, "Failed to create exchange")
// create queue
queue, err := rabbitChan.QueueDeclare("orders", true, false, false, false, nil)
failOnError(err, "Failed to create queue")
// bind queue to exchange
err = rabbitChan.QueueBind(queue.Name, "orders", "robot-shop", false, nil)
failOnError(err, "Failed to bind queue")
// signal ready
rabbitReady <- true
}
}
}
func failOnError(err error, msg string) {
if err != nil {
log.Fatalf("$s : %s", msg, err)
panic(fmt.Sprintf("%s : %s", msg, err))
}
}
func createSpan(headers map[string]interface{}) {
// headers is map[string]interface{}
// carrier is map[string]string
carrier := make(ot.TextMapCarrier)
// convert by copying k, v
for k, v := range headers {
carrier[k] = v.(string)
}
// opentracing
var span ot.Span
tracer := ot.GlobalTracer()
spanContext, err := tracer.Extract(ot.HTTPHeaders, carrier)
if err == nil {
log.Println("Creating span")
// create span
span = tracer.StartSpan("dispatch", ot.ChildOf(spanContext), ext.SpanKindConsumer)
ext.MessageBusDestination.Set(span, "orders")
ext.Component.Set(span, "dispatch")
defer span.Finish()
time.Sleep(42 * time.Millisecond)
} else {
log.Println("Failed to get span context")
log.Println(err)
}
}
func main() {
// Instana tracing
ot.InitGlobalTracer(instana.NewTracerWithOptions(&instana.Options{
Service: Service,
LogLevel: instana.Info}))
// MQ error channel
rabbitCloseError = make(chan *amqp.Error)
// MQ ready channel
rabbitReady = make(chan bool)
go rabbitConnector(amqpUri)
rabbitCloseError <- amqp.ErrClosed
go func() {
for {
// wait for rabbit to be ready
ready := <-rabbitReady
log.Printf("Rabbit MQ ready %v\n", ready)
// subscribe to bound queue
msgs, err := rabbitChan.Consume("orders", "", true, false, false, false, nil)
failOnError(err, "Failed to consume")
for d := range msgs {
log.Printf("Order %s\n", d.Body)
log.Printf("Headers %v\n", d.Headers)
go createSpan(d.Headers)
}
}
}()
log.Println("Waiting for messages")
forever := make(chan bool)
<-forever
}