00001 #ifndef _broker_SessionAdapter_h
00002 #define _broker_SessionAdapter_h
00003
00004
00005
00006
00007
00008
00009
00010
00011
00012
00013
00014
00015
00016
00017
00018
00019
00020
00021
00022 #include "HandlerImpl.h"
00023
00024 #include "ConnectionToken.h"
00025 #include "OwnershipToken.h"
00026 #include "qpid/Exception.h"
00027 #include "qpid/framing/AMQP_ServerOperations.h"
00028 #include "qpid/framing/reply_exceptions.h"
00029 #include "qpid/framing/StructHelper.h"
00030
00031 #include <vector>
00032 #include <boost/function.hpp>
00033 #include <boost/shared_ptr.hpp>
00034
00035 namespace qpid {
00036 namespace broker {
00037
00038 class Channel;
00039 class Connection;
00040 class Broker;
00041 class Queue;
00042
00052 class SessionAdapter : public HandlerImpl, public framing::AMQP_ServerOperations
00053 {
00054 public:
00055 SessionAdapter(SemanticState& session);
00056
00057 framing::ProtocolVersion getVersion() const { return session.getConnection().getVersion();}
00058
00059 MessageHandler* getMessageHandler(){ return &messageImpl; }
00060 ExchangeHandler* getExchangeHandler(){ return &exchangeImpl; }
00061 QueueHandler* getQueueHandler(){ return &queueImpl; }
00062 ExecutionHandler* getExecutionHandler(){ return &executionImpl; }
00063 TxHandler* getTxHandler(){ return &txImpl; }
00064 DtxHandler* getDtxHandler(){ return &dtxImpl; }
00065
00066 ConnectionHandler* getConnectionHandler() { throw framing::NotImplementedException("Class not implemented"); }
00067 SessionHandler* getSessionHandler() { throw framing::NotImplementedException("Class not implemented"); }
00068 FileHandler* getFileHandler() { throw framing::NotImplementedException("Class not implemented"); }
00069 StreamHandler* getStreamHandler() { throw framing::NotImplementedException("Class not implemented"); }
00070
00071 private:
00072
00073 struct HandlerHelper : public HandlerImpl
00074 {
00075 HandlerHelper(SemanticState& s) : HandlerImpl(s) {}
00076
00077 Queue::shared_ptr getQueue(const string& name) const;
00078 };
00079
00080
00081 class ExchangeHandlerImpl :
00082 public ExchangeHandler,
00083 public HandlerHelper
00084 {
00085 public:
00086 ExchangeHandlerImpl(SemanticState& session) : HandlerHelper(session) {}
00087
00088 void declare(const std::string& exchange, const std::string& type,
00089 const std::string& alternateExchange,
00090 bool passive, bool durable, bool autoDelete,
00091 const qpid::framing::FieldTable& arguments);
00092 void delete_(const std::string& exchange, bool ifUnused);
00093 framing::ExchangeQueryResult query(const std::string& name);
00094 void bind(const std::string& queue,
00095 const std::string& exchange, const std::string& routingKey,
00096 const qpid::framing::FieldTable& arguments);
00097 void unbind(const std::string& queue,
00098 const std::string& exchange,
00099 const std::string& routingKey);
00100 framing::ExchangeBoundResult bound(const std::string& exchange,
00101 const std::string& queue,
00102 const std::string& routingKey,
00103 const framing::FieldTable& arguments);
00104 private:
00105 void checkType(shared_ptr<Exchange> exchange, const std::string& type);
00106
00107 void checkAlternate(shared_ptr<Exchange> exchange,
00108 shared_ptr<Exchange> alternate);
00109 };
00110
00111 class QueueHandlerImpl : public QueueHandler,
00112 public HandlerHelper, public OwnershipToken
00113 {
00114 Broker& broker;
00115 std::vector< boost::shared_ptr<Queue> > exclusiveQueues;
00116
00117 public:
00118 QueueHandlerImpl(SemanticState& session);
00119 ~QueueHandlerImpl();
00120
00121 void declare(const std::string& queue,
00122 const std::string& alternateExchange,
00123 bool passive, bool durable, bool exclusive,
00124 bool autoDelete,
00125 const qpid::framing::FieldTable& arguments);
00126 void delete_(const std::string& queue,
00127 bool ifUnused, bool ifEmpty);
00128 void purge(const std::string& queue);
00129 framing::QueueQueryResult query(const std::string& queue);
00130 bool isLocal(const ConnectionToken* t) const;
00131
00132 void destroyExclusiveQueues();
00133 };
00134
00135 class MessageHandlerImpl :
00136 public MessageHandler,
00137 public HandlerHelper
00138 {
00139 typedef boost::function<void(DeliveryId, DeliveryId)> RangedOperation;
00140 RangedOperation releaseRedeliveredOp;
00141 RangedOperation releaseOp;
00142 RangedOperation rejectOp;
00143 RangedOperation acceptOp;
00144
00145 public:
00146 MessageHandlerImpl(SemanticState& session);
00147 void transfer(const string& destination,
00148 uint8_t acceptMode,
00149 uint8_t acquireMode);
00150
00151 void accept(const framing::SequenceSet& commands);
00152
00153 void reject(const framing::SequenceSet& commands,
00154 uint16_t code,
00155 const string& text);
00156
00157 void release(const framing::SequenceSet& commands,
00158 bool setRedelivered);
00159
00160 framing::MessageAcquireResult acquire(const framing::SequenceSet&);
00161
00162 void subscribe(const string& queue,
00163 const string& destination,
00164 uint8_t acceptMode,
00165 uint8_t acquireMode,
00166 bool exclusive,
00167 const string& resumeId,
00168 uint64_t resumeTtl,
00169 const framing::FieldTable& arguments);
00170
00171 void cancel(const string& destination);
00172
00173 void setFlowMode(const string& destination,
00174 uint8_t flowMode);
00175
00176 void flow(const string& destination,
00177 uint8_t unit,
00178 uint32_t value);
00179
00180 void flush(const string& destination);
00181
00182 void stop(const string& destination);
00183
00184 framing::MessageResumeResult resume(const std::string& destination,
00185 const std::string& resumeId);
00186
00187 };
00188
00189 class ExecutionHandlerImpl : public ExecutionHandler, public HandlerHelper
00190 {
00191 public:
00192 ExecutionHandlerImpl(SemanticState& session) : HandlerHelper(session) {}
00193
00194 void sync();
00195 void result(const framing::SequenceNumber& commandId, const string& value);
00196 void exception(uint16_t errorCode,
00197 const framing::SequenceNumber& commandId,
00198 uint8_t classCode,
00199 uint8_t commandCode,
00200 uint8_t fieldIndex,
00201 const std::string& description,
00202 const framing::FieldTable& errorInfo);
00203
00204 };
00205
00206 class TxHandlerImpl : public TxHandler, public HandlerHelper
00207 {
00208 public:
00209 TxHandlerImpl(SemanticState& session) : HandlerHelper(session) {}
00210
00211 void select();
00212 void commit();
00213 void rollback();
00214 };
00215
00216 class DtxHandlerImpl : public DtxHandler, public HandlerHelper, private framing::StructHelper
00217 {
00218 std::string convert(const framing::Xid& xid);
00219
00220 public:
00221 DtxHandlerImpl(SemanticState& session) : HandlerHelper(session) {}
00222
00223 void select();
00224
00225 framing::XaResult start(const framing::Xid& xid,
00226 bool join,
00227 bool resume);
00228
00229 framing::XaResult end(const framing::Xid& xid,
00230 bool fail,
00231 bool suspend);
00232
00233 framing::XaResult commit(const framing::Xid& xid,
00234 bool onePhase);
00235
00236 void forget(const framing::Xid& xid);
00237
00238 framing::DtxGetTimeoutResult getTimeout(const framing::Xid& xid);
00239
00240 framing::XaResult prepare(const framing::Xid& xid);
00241
00242 framing::DtxRecoverResult recover();
00243
00244 framing::XaResult rollback(const framing::Xid& xid);
00245
00246 void setTimeout(const framing::Xid& xid, uint32_t timeout);
00247 };
00248
00249 ExchangeHandlerImpl exchangeImpl;
00250 QueueHandlerImpl queueImpl;
00251 MessageHandlerImpl messageImpl;
00252 ExecutionHandlerImpl executionImpl;
00253 TxHandlerImpl txImpl;
00254 DtxHandlerImpl dtxImpl;
00255 };
00256 }}
00257
00258
00259
00260 #endif