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
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
|
/*
*
* 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.
*
*/
#ifndef _SessionHandlerImpl_
#define _SessionHandlerImpl_
#include <map>
#include <sstream>
#include <vector>
#include <exception>
#include "AMQFrame.h"
#include "AMQP_ClientProxy.h"
#include "AMQP_ServerOperations.h"
#include "AutoDelete.h"
#include "ExchangeRegistry.h"
#include "Channel.h"
#include "ConnectionToken.h"
#include "DirectExchange.h"
#include "OutputHandler.h"
#include "ProtocolInitiation.h"
#include "QueueRegistry.h"
#include "SessionContext.h"
#include "SessionHandler.h"
#include "TimeoutHandler.h"
#include "TopicExchange.h"
namespace qpid {
namespace broker {
struct ChannelException : public std::exception {
u_int16_t code;
string text;
ChannelException(u_int16_t _code, string _text) : code(_code), text(_text) {}
~ChannelException() throw() {}
const char* what() const throw() { return text.c_str(); }
};
struct ConnectionException : public std::exception {
u_int16_t code;
string text;
ConnectionException(u_int16_t _code, string _text) : code(_code), text(_text) {}
~ConnectionException() throw() {}
const char* what() const throw() { return text.c_str(); }
};
class SessionHandlerImpl : public virtual qpid::io::SessionHandler,
public virtual qpid::framing::AMQP_ServerOperations,
public virtual ConnectionToken
{
typedef std::map<u_int16_t, Channel*>::iterator channel_iterator;
typedef std::vector<Queue::shared_ptr>::iterator queue_iterator;
qpid::io::SessionContext* context;
qpid::framing::AMQP_ClientProxy client;
QueueRegistry* queues;
ExchangeRegistry* const exchanges;
AutoDelete* const cleaner;
const u_int32_t timeout;//timeout for auto-deleted queues (in ms)
ConnectionHandler* connectionHandler;
ChannelHandler* channelHandler;
BasicHandler* basicHandler;
ExchangeHandler* exchangeHandler;
QueueHandler* queueHandler;
std::map<u_int16_t, Channel*> channels;
std::vector<Queue::shared_ptr> exclusiveQueues;
u_int32_t framemax;
u_int16_t heartbeat;
void handleHeader(u_int16_t channel, qpid::framing::AMQHeaderBody::shared_ptr body);
void handleContent(u_int16_t channel, qpid::framing::AMQContentBody::shared_ptr body);
void handleHeartbeat(qpid::framing::AMQHeartbeatBody::shared_ptr body);
/**
* Get named queue, never returns 0.
* @return: named queue or default queue for channel if name=""
* @exception: ChannelException if no queue of that name is found.
* @exception: ConnectionException if no queue specified and channel has not declared one.
*/
Queue::shared_ptr getQueue(const string& name, u_int16_t channel);
Exchange* findExchange(const string& name);
public:
SessionHandlerImpl(qpid::io::SessionContext* context, QueueRegistry* queues,
ExchangeRegistry* exchanges, AutoDelete* cleaner, const u_int32_t timeout);
virtual void received(qpid::framing::AMQFrame* frame);
virtual void initiated(qpid::framing::ProtocolInitiation* header);
virtual void idleOut();
virtual void idleIn();
virtual void closed();
virtual ~SessionHandlerImpl();
class ConnectionHandlerImpl : public virtual ConnectionHandler{
SessionHandlerImpl* parent;
public:
inline ConnectionHandlerImpl(SessionHandlerImpl* _parent) : parent(_parent) {}
virtual void startOk(u_int16_t channel, qpid::framing::FieldTable& clientProperties, string& mechanism,
string& response, string& locale);
virtual void secureOk(u_int16_t channel, string& response);
virtual void tuneOk(u_int16_t channel, u_int16_t channelMax, u_int32_t frameMax, u_int16_t heartbeat);
virtual void open(u_int16_t channel, string& virtualHost, string& capabilities, bool insist);
virtual void close(u_int16_t channel, u_int16_t replyCode, string& replyText, u_int16_t classId,
u_int16_t methodId);
virtual void closeOk(u_int16_t channel);
virtual ~ConnectionHandlerImpl(){}
};
class ChannelHandlerImpl : public virtual ChannelHandler{
SessionHandlerImpl* parent;
public:
inline ChannelHandlerImpl(SessionHandlerImpl* _parent) : parent(_parent) {}
virtual void open(u_int16_t channel, string& outOfBand);
virtual void flow(u_int16_t channel, bool active);
virtual void flowOk(u_int16_t channel, bool active);
virtual void close(u_int16_t channel, u_int16_t replyCode, string& replyText,
u_int16_t classId, u_int16_t methodId);
virtual void closeOk(u_int16_t channel);
virtual ~ChannelHandlerImpl(){}
};
class ExchangeHandlerImpl : public virtual ExchangeHandler{
SessionHandlerImpl* parent;
public:
inline ExchangeHandlerImpl(SessionHandlerImpl* _parent) : parent(_parent) {}
virtual void declare(u_int16_t channel, u_int16_t ticket, string& exchange, string& type,
bool passive, bool durable, bool autoDelete, bool internal, bool nowait,
qpid::framing::FieldTable& arguments);
virtual void delete_(u_int16_t channel, u_int16_t ticket, string& exchange, bool ifUnused, bool nowait);
virtual ~ExchangeHandlerImpl(){}
};
class QueueHandlerImpl : public virtual QueueHandler{
SessionHandlerImpl* parent;
public:
inline QueueHandlerImpl(SessionHandlerImpl* _parent) : parent(_parent) {}
virtual void declare(u_int16_t channel, u_int16_t ticket, string& queue,
bool passive, bool durable, bool exclusive,
bool autoDelete, bool nowait, qpid::framing::FieldTable& arguments);
virtual void bind(u_int16_t channel, u_int16_t ticket, string& queue,
string& exchange, string& routingKey, bool nowait,
qpid::framing::FieldTable& arguments);
virtual void purge(u_int16_t channel, u_int16_t ticket, string& queue,
bool nowait);
virtual void delete_(u_int16_t channel, u_int16_t ticket, string& queue, bool ifUnused, bool ifEmpty,
bool nowait);
virtual ~QueueHandlerImpl(){}
};
class BasicHandlerImpl : public virtual BasicHandler{
SessionHandlerImpl* parent;
public:
inline BasicHandlerImpl(SessionHandlerImpl* _parent) : parent(_parent) {}
virtual void qos(u_int16_t channel, u_int32_t prefetchSize, u_int16_t prefetchCount, bool global);
virtual void consume(u_int16_t channel, u_int16_t ticket, string& queue, string& consumerTag,
bool noLocal, bool noAck, bool exclusive, bool nowait);
virtual void cancel(u_int16_t channel, string& consumerTag, bool nowait);
virtual void publish(u_int16_t channel, u_int16_t ticket, string& exchange, string& routingKey,
bool mandatory, bool immediate);
virtual void get(u_int16_t channel, u_int16_t ticket, string& queue, bool noAck);
virtual void ack(u_int16_t channel, u_int64_t deliveryTag, bool multiple);
virtual void reject(u_int16_t channel, u_int64_t deliveryTag, bool requeue);
virtual void recover(u_int16_t channel, bool requeue);
virtual ~BasicHandlerImpl(){}
};
inline virtual ChannelHandler* getChannelHandler(){ return channelHandler; }
inline virtual ConnectionHandler* getConnectionHandler(){ return connectionHandler; }
inline virtual BasicHandler* getBasicHandler(){ return basicHandler; }
inline virtual ExchangeHandler* getExchangeHandler(){ return exchangeHandler; }
inline virtual QueueHandler* getQueueHandler(){ return queueHandler; }
inline virtual AccessHandler* getAccessHandler(){ return 0; }
inline virtual FileHandler* getFileHandler(){ return 0; }
inline virtual StreamHandler* getStreamHandler(){ return 0; }
inline virtual TxHandler* getTxHandler(){ return 0; }
inline virtual DtxHandler* getDtxHandler(){ return 0; }
inline virtual TunnelHandler* getTunnelHandler(){ return 0; }
};
}
}
#endif
|
