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
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
|
// This example declares a durable Exchange, an ephemeral (auto-delete) Queue,
// binds the Queue to the Exchange with a binding key, and consumes every
// message published to that Exchange with that routing key.
//
package main
import (
"flag"
"fmt"
"log"
"time"
"github.com/NeowayLabs/wabbit"
"github.com/NeowayLabs/wabbit/amqp"
)
var (
uri = flag.String("uri", "amqp://guest:guest@localhost:5672/", "AMQP URI")
exchange = flag.String("exchange", "test-exchange", "Durable, non-auto-deleted AMQP exchange name")
exchangeType = flag.String("exchange-type", "direct", "Exchange type - direct|fanout|topic|x-custom")
queue = flag.String("queue", "test-queue", "Ephemeral AMQP queue name")
bindingKey = flag.String("key", "test-key", "AMQP binding key")
consumerTag = flag.String("consumer-tag", "simple-consumer", "AMQP consumer tag (should not be blank)")
lifetime = flag.Duration("lifetime", 5*time.Second, "lifetime of process before shutdown (0s=infinite)")
)
func init() {
flag.Parse()
}
func main() {
c, err := NewConsumer(*uri, *exchange, *exchangeType, *queue, *bindingKey, *consumerTag)
if err != nil {
log.Fatalf("%s", err)
}
if *lifetime > 0 {
log.Printf("running for %s", *lifetime)
time.Sleep(*lifetime)
} else {
log.Printf("running forever")
select {}
}
log.Printf("shutting down")
if err := c.Shutdown(); err != nil {
log.Fatalf("error during shutdown: %s", err)
}
}
type Consumer struct {
conn wabbit.Conn
channel wabbit.Channel
tag string
done chan error
}
func NewConsumer(amqpURI, exchange, exchangeType, queueName, key, ctag string) (*Consumer, error) {
c := &Consumer{
conn: nil,
channel: nil,
tag: ctag,
done: make(chan error),
}
var err error
log.Printf("dialing %q", amqpURI)
c.conn, err = amqp.Dial(amqpURI)
if err != nil {
return nil, fmt.Errorf("Dial: %s", err)
}
go func() {
fmt.Printf("closing: %s", <-c.conn.NotifyClose(make(chan wabbit.Error)))
}()
log.Printf("got Connection, getting Channel")
c.channel, err = c.conn.Channel()
if err != nil {
return nil, fmt.Errorf("Channel: %s", err)
}
log.Printf("got Channel, declaring Exchange (%q)", exchange)
if err = c.channel.ExchangeDeclare(
exchange, // name of the exchange
exchangeType, // type
wabbit.Option{
"durable": true,
"delete": false,
"internal": false,
"noWait": false,
},
); err != nil {
return nil, fmt.Errorf("Exchange Declare: %s", err)
}
log.Printf("declared Exchange, declaring Queue %q", queueName)
queue, err := c.channel.QueueDeclare(
queueName, // name of the queue
wabbit.Option{
"durable": true,
"delete": false,
"exclusive": false,
"noWait": false,
},
)
if err != nil {
return nil, fmt.Errorf("Queue Declare: %s", err)
}
log.Printf("declared Queue (%q %d messages, %d consumers), binding to Exchange (key %q)",
queue.Name(), queue.Messages(), queue.Consumers(), key)
if err = c.channel.QueueBind(
queue.Name(), // name of the queue
key, // bindingKey
exchange, // sourceExchange
wabbit.Option{
"noWait": false,
},
); err != nil {
return nil, fmt.Errorf("Queue Bind: %s", err)
}
log.Printf("Queue bound to Exchange, starting Consume (consumer tag %q)", c.tag)
deliveries, err := c.channel.Consume(
queue.Name(), // name
c.tag, // consumerTag,
wabbit.Option{
"noAck": false,
"exclusive": false,
"noLocal": false,
"noWait": false,
},
)
if err != nil {
return nil, fmt.Errorf("Queue Consume: %s", err)
}
go handle(deliveries, c.done)
return c, nil
}
func (c *Consumer) Shutdown() error {
// will close() the deliveries channel
if err := c.channel.Cancel(c.tag, true); err != nil {
return fmt.Errorf("Consumer cancel failed: %s", err)
}
if err := c.conn.Close(); err != nil {
return fmt.Errorf("AMQP connection close error: %s", err)
}
defer log.Printf("AMQP shutdown OK")
// wait for handle() to exit
return <-c.done
}
func handle(deliveries <-chan wabbit.Delivery, done chan error) {
for d := range deliveries {
log.Printf(
"got %dB delivery: [%v] %q",
len(d.Body()),
d.DeliveryTag(),
d.Body(),
)
d.Ack(false)
}
log.Printf("handle: deliveries channel closed")
done <- nil
}
|
