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
184
185
186
187
188
189
190
191
192
193
194
195
|
#ifndef _client_Basic_h
#define _client_Basic_h
/*
*
* Copyright (c) 2006 The Apache Software Foundation
*
* Licensed 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.
*
*/
#include "IncomingMessage.h"
#include "sys/Runnable.h"
namespace qpid {
namespace framing {
class AMQMethodBody;
class FieldTable;
}
namespace client {
class Channel;
class Message;
class Queue;
class Exchange;
class MessageListener;
class ReturnedMessageHandler;
/**
* The available acknowledgements modes.
*
* \ingroup clientapi
*/
enum AckMode {
/** No acknowledgement will be sent, broker can
discard messages as soon as they are delivered
to a consumer using this mode. **/
NO_ACK = 0,
/** Each message will be automatically
acknowledged as soon as it is delivered to the
application **/
AUTO_ACK = 1,
/** Acknowledgements will be sent automatically,
but not for each message. **/
LAZY_ACK = 2,
/** The application is responsible for explicitly
acknowledging messages. **/
CLIENT_ACK = 3
};
/**
* Represents the AMQP Basic class for sending and receiving messages.
*/
class Basic : public sys::Runnable
{
public:
Basic(Channel& parent);
/**
* Creates a 'consumer' for a queue. Messages in (or arriving
* at) that queue will be delivered to consumers
* asynchronously.
*
* @param queue a Queue instance representing the queue to
* consume from
*
* @param tag an identifier to associate with the consumer
* that can be used to cancel its subscription (if empty, this
* will be assigned by the broker)
*
* @param listener a pointer to an instance of an
* implementation of the MessageListener interface. Messages
* received from this queue for this consumer will result in
* invocation of the received() method on the listener, with
* the message itself passed in.
*
* @param ackMode the mode of acknowledgement that the broker
* should assume for this consumer. @see AckMode
*
* @param noLocal if true, this consumer will not be sent any
* message published by this connection
*
* @param synch if true this call will block until a response
* is received from the broker
*/
void consume(
Queue& queue, std::string& tag, MessageListener* listener,
AckMode ackMode = NO_ACK, bool noLocal = false, bool synch = true,
const framing::FieldTable* fields = 0);
/**
* Cancels a subscription previously set up through a call to consume().
*
* @param tag the identifier used (or assigned) in the consume
* request that set up the subscription to be cancelled.
*
* @param synch if true this call will block until a response
* is received from the broker
*/
void cancel(const std::string& tag, bool synch = true);
/**
* Synchronous pull of a message from a queue.
*
* @param msg a message object that will contain the message
* headers and content if the call completes.
*
* @param queue the queue to consume from
*
* @param ackMode the acknowledgement mode to use (@see
* AckMode)
*
* @return true if a message was succcessfully dequeued from
* the queue, false if the queue was empty.
*/
bool get(Message& msg, const Queue& queue, AckMode ackMode = NO_ACK);
/**
* Publishes (i.e. sends a message to the broker).
*
* @param msg the message to publish
*
* @param exchange the exchange to publish the message to
*
* @param routingKey the routing key to publish with
*
* @param mandatory if true and the exchange to which this
* publish is directed has no matching bindings, the message
* will be returned (see setReturnedMessageHandler()).
*
* @param immediate if true and there is no consumer to
* receive this message on publication, the message will be
* returned (see setReturnedMessageHandler()).
*/
void publish(const Message& msg, const Exchange& exchange,
const std::string& routingKey,
bool mandatory = false, bool immediate = false);
/**
* Set a handler for this channel that will process any
* returned messages
*
* @see publish()
*/
void setReturnedMessageHandler(ReturnedMessageHandler* handler);
/**
* Deliver messages from the broker to the appropriate MessageListener.
*/
void run();
private:
struct Consumer{
MessageListener* listener;
AckMode ackMode;
int count;
uint64_t lastDeliveryTag;
};
typedef std::map<std::string, Consumer> ConsumerMap;
void handle(boost::shared_ptr<framing::AMQMethodBody>);
void setQos();
void cancelAll();
void deliver(Consumer& consumer, Message& msg);
sys::Mutex lock;
Channel& channel;
IncomingMessage incoming;
ConsumerMap consumers;
ReturnedMessageHandler* returnsHandler;
// FIXME aconway 2007-02-22: Remove friendship.
friend class Channel;
};
}} // namespace qpid::client
#endif /*!_client_Basic_h*/
|
