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
176
177
178
179
180
181
182
183
|
/*
*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*
*/
/**
* topic_publisher.cpp:
*
* This program is one of three programs designed to be used
* together. These programs implement a publish-subscribe example
* using the "amq.topic" exchange. In the example multiple listeners
* can subscribe to the same queues for TTL messages.
* The TTL messages are all ticker price data. Messages are
* browsed and therefore shared among the multiple listeners.
* Messages timeout using TTL so that they don't stay in the queue
* for too long and fill it up.
* Local exclusive LVQ are also declared for market data.
*
* declare_queues.cpp
*
* Declares several non-exclusive queues bound to the amq:topic exchange
*
* topic_publisher.cpp
*
* Sends messages to the "amq.topic" exchange, using the
* multipart routing keys for ticker price and market data
* Ticker messages are sent using a TTL value.
*
* topic_listener.cpp (this program)
*
* Subscribes to non-exclusive queues in NOT_ACQUIRE mode for
* ticker price data and declares two LVQs for market data.
*
* Multiple listeners can be run at the same time.
*
*/
#include <qpid/client/Connection.h>
#include <qpid/client/Session.h>
#include <qpid/client/Message.h>
#include <qpid/client/MessageListener.h>
#include <qpid/client/SubscriptionManager.h>
#include "qpid/client/QueueOptions.h"
#include <cstdlib>
#include <iostream>
using namespace qpid::client;
using namespace qpid::framing;
class Listener : public MessageListener {
private:
Session& session;
SubscriptionManager subscriptions;
public:
Listener(Session& session);
virtual void subscribeTTLQueue(std::string queue);
virtual void subscribeLVQQueue(std::string queue);
virtual void received(Message& message);
virtual void listen();
~Listener() { };
};
/*
* Listener::Listener
*
* Subscribe to the queue, route it to a client destination for the
* listener. (The destination name merely identifies the destination
* in the listener, you can use any name as long as you use the same
* name for the listener).
*/
Listener::Listener(Session& session) :
session(session),
subscriptions(session)
{
}
void Listener::subscribeTTLQueue(std::string queue) {
/*
* Subscribe to the queue using the subscription manager.
* The queues were declared elsewhere alog with their bindings.
*/
std::cout << "Subscribing to queue " << queue << std::endl;
subscriptions.subscribe(*this, queue);
// Will not acquire messages but instead browse them.
subscriptions.setAcquireMode(message::ACQUIRE_MODE_NOT_ACQUIRED);
}
void Listener::subscribeLVQQueue(std::string queue) {
/*
* Declare and subscribe to the queue using the subscription manager.
*/
QueueOptions qo;
qo.setOrdering(LVQ);
std::string binding = queue + ".#";
queue += session.getId().getName();
session.queueDeclare(arg::queue=queue, arg::exclusive=true, arg::arguments=qo);
session.exchangeBind(arg::exchange="amq.topic", arg::queue=queue, arg::bindingKey=binding);
std::cout << "Declared queue " << queue << " non-exclusive with amq:topic binding " << binding << std::endl;
std::cout << "Subscribing to queue " << queue << std::endl;
subscriptions.subscribe(*this, queue, SubscriptionSettings(FlowControl::unlimited(), ACCEPT_MODE_NONE));
}
void Listener::received(Message& message) {
// If you want to see the destination you can swap the following lines.
// std::cout << message.getDestination() << "\t" << message.getData() << std::endl;
std::cout << message.getData() << std::endl;
}
void Listener::listen() {
// Receive messages
subscriptions.run();
}
int main(int argc, char** argv) {
const char* host = argc>1 ? argv[1] : "127.0.0.1";
int port = argc>2 ? atoi(argv[2]) : 5672;
Connection connection;
try {
connection.open(host, port);
Session session = connection.newSession();
//--------- Main body of program --------------------------------------------
// Create a listener for the session
Listener listener(session);
// Subscribe to messages on the queues we are interested in
listener.subscribeTTLQueue("TICKER.NASDAQ");
listener.subscribeTTLQueue("TICKER.NYSE");
listener.subscribeLVQQueue("MRKT.NASDAQ");
listener.subscribeLVQQueue("MRKT.NYSE");
std::cout << "Starting Listener <Ctrl>-C to exit." << std::endl;
std::cout << "Listening for messages ..." << std::endl;
// Give up control and receive messages
listener.listen();
//-----------------------------------------------------------------------------
connection.close();
return 0;
} catch(const std::exception& error) {
std::cout << error.what() << std::endl;
}
return 1;
}
|
