You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ac...@apache.org on 2007/02/13 03:41:16 UTC
svn commit: r506823 [1/2] - in /incubator/qpid/branches/qpid.0-9: cpp/lib/
cpp/lib/broker/ cpp/lib/client/ cpp/lib/common/ cpp/lib/common/framing/
cpp/tests/ gentools/src/org/apache/qpid/gentools/ gentools/templ.cpp/
Author: aconway
Date: Mon Feb 12 18:41:14 2007
New Revision: 506823
URL: http://svn.apache.org/viewvc?view=rev&rev=506823
Log:
* gentools/templ.cpp/*Proxy*, CppGenerator.java: Changes to Proxy
classes to make them directly usable as an API for low-level AMQP access.
- Proxies hold reference to a ChannelAdapter not just an output handler.
- Removed MethodContext parameter, makes no sense on requester end.
- Return RequestId from request methods so caller can correlate
incoming responses.
- Add RequestId parameter to response methods so caller can provide
correlation for outgoing responses.
- No longer inherit from *Operations classes as the signatures no
longer match. Proxy is for caller (client/requester) and Operations
is for callee (server/responder)
* cpp/lib/client/ClientChannel.h: Channel provides a raw proxy to the broker.
Normal users will still use the Channel API to deal with the broker, but
advanced users (incl ourselves!) can use the raw API to directly send
and receive any AMQP message.
* cpp/lib/broker/BrokerChannel,BrokerAdapter: Refactor for new proxies.
broker::Channel is also a ClientProxy
* Sundry files:
- Pass ProtcolVersion by value, it is only two bytes.
- Misc. const correctness fixes.
Added:
incubator/qpid/branches/qpid.0-9/cpp/lib/broker/HandlerImpl.h (with props)
incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/MethodContext.cpp (with props)
incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/Proxy.cpp (with props)
incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/Proxy.h (with props)
incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/amqp_types_full.h (with props)
Modified:
incubator/qpid/branches/qpid.0-9/cpp/lib/Makefile.am
incubator/qpid/branches/qpid.0-9/cpp/lib/broker/Broker.h
incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerAdapter.cpp
incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerAdapter.h
incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerChannel.cpp
incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerChannel.h
incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerMessage.cpp
incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerMessage.h
incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerMessageBase.h
incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerQueue.cpp
incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerQueue.h
incubator/qpid/branches/qpid.0-9/cpp/lib/broker/Connection.cpp
incubator/qpid/branches/qpid.0-9/cpp/lib/broker/Connection.h
incubator/qpid/branches/qpid.0-9/cpp/lib/broker/MessageHandlerImpl.cpp
incubator/qpid/branches/qpid.0-9/cpp/lib/broker/MessageHandlerImpl.h
incubator/qpid/branches/qpid.0-9/cpp/lib/client/ClientChannel.cpp
incubator/qpid/branches/qpid.0-9/cpp/lib/client/ClientChannel.h
incubator/qpid/branches/qpid.0-9/cpp/lib/client/Connection.cpp
incubator/qpid/branches/qpid.0-9/cpp/lib/client/Connection.h
incubator/qpid/branches/qpid.0-9/cpp/lib/client/Connector.cpp
incubator/qpid/branches/qpid.0-9/cpp/lib/client/Connector.h
incubator/qpid/branches/qpid.0-9/cpp/lib/common/Makefile.am
incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/AMQFrame.cpp
incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/AMQFrame.h
incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/ChannelAdapter.cpp
incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/ChannelAdapter.h
incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/FramingContent.h
incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/MethodContext.h
incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/ProtocolInitiation.cpp
incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/ProtocolInitiation.h
incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/ProtocolVersion.cpp
incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/ProtocolVersion.h
incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/ProtocolVersionException.h
incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/amqp_types.h
incubator/qpid/branches/qpid.0-9/cpp/tests/MessageTest.cpp
incubator/qpid/branches/qpid.0-9/gentools/src/org/apache/qpid/gentools/CppGenerator.java
incubator/qpid/branches/qpid.0-9/gentools/templ.cpp/AMQP_ClientOperations.h.tmpl
incubator/qpid/branches/qpid.0-9/gentools/templ.cpp/AMQP_ClientProxy.cpp.tmpl
incubator/qpid/branches/qpid.0-9/gentools/templ.cpp/AMQP_ClientProxy.h.tmpl
incubator/qpid/branches/qpid.0-9/gentools/templ.cpp/AMQP_MethodVersionMap.cpp.tmpl
incubator/qpid/branches/qpid.0-9/gentools/templ.cpp/AMQP_ServerOperations.h.tmpl
incubator/qpid/branches/qpid.0-9/gentools/templ.cpp/AMQP_ServerProxy.cpp.tmpl
incubator/qpid/branches/qpid.0-9/gentools/templ.cpp/AMQP_ServerProxy.h.tmpl
incubator/qpid/branches/qpid.0-9/gentools/templ.cpp/MethodBodyClass.h.tmpl
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/Makefile.am
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/Makefile.am?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/Makefile.am (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/Makefile.am Mon Feb 12 18:41:14 2007
@@ -1 +1 @@
-SUBDIRS = client common broker
+SUBDIRS = common broker client
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/broker/Broker.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/broker/Broker.h?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/broker/Broker.h (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/broker/Broker.h Mon Feb 12 18:41:14 2007
@@ -30,7 +30,6 @@
#include <MessageStore.h>
#include <AutoDelete.h>
#include <ExchangeRegistry.h>
-#include <BrokerChannel.h>
#include <ConnectionToken.h>
#include <DirectExchange.h>
#include <OutputHandler.h>
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerAdapter.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerAdapter.cpp?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerAdapter.cpp (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerAdapter.cpp Mon Feb 12 18:41:14 2007
@@ -18,11 +18,10 @@
#include <boost/format.hpp>
#include "BrokerAdapter.h"
+#include "BrokerChannel.h"
#include "Connection.h"
-#include "Exception.h"
#include "AMQMethodBody.h"
#include "Exception.h"
-#include "MessageHandlerImpl.h"
namespace qpid {
namespace broker {
@@ -33,18 +32,37 @@
typedef std::vector<Queue::shared_ptr> QueueVector;
-void BrokerAdapter::BrokerAdapter::ConnectionHandlerImpl::startOk(
- const MethodContext& context , const FieldTable& /*clientProperties*/,
+
+BrokerAdapter::BrokerAdapter(Channel& ch, Connection& c, Broker& b) :
+ CoreRefs(ch, c, b),
+ connection(c),
+ basicHandler(*this),
+ channelHandler(*this),
+ connectionHandler(*this),
+ exchangeHandler(*this),
+ messageHandler(*this),
+ queueHandler(*this),
+ txHandler(*this)
+{}
+
+
+ProtocolVersion BrokerAdapter::getVersion() const {
+ return connection.getVersion();
+}
+
+void BrokerAdapter::ConnectionHandlerImpl::startOk(
+ const MethodContext&, const FieldTable& /*clientProperties*/,
const string& /*mechanism*/,
- const string& /*response*/, const string& /*locale*/){
- connection.client->getConnection().tune(
- context, 100, connection.getFrameMax(), connection.getHeartbeat());
+ const string& /*response*/, const string& /*locale*/)
+{
+ client.tune(
+ 100, connection.getFrameMax(), connection.getHeartbeat());
}
-void BrokerAdapter::BrokerAdapter::ConnectionHandlerImpl::secureOk(
+void BrokerAdapter::ConnectionHandlerImpl::secureOk(
const MethodContext&, const string& /*response*/){}
-void BrokerAdapter::BrokerAdapter::ConnectionHandlerImpl::tuneOk(
+void BrokerAdapter::ConnectionHandlerImpl::tuneOk(
const MethodContext&, u_int16_t /*channelmax*/,
u_int32_t framemax, u_int16_t heartbeat)
{
@@ -52,50 +70,55 @@
connection.setHeartbeat(heartbeat);
}
-void BrokerAdapter::BrokerAdapter::ConnectionHandlerImpl::open(const MethodContext& context, const string& /*virtualHost*/, const string& /*capabilities*/, bool /*insist*/){
+void BrokerAdapter::ConnectionHandlerImpl::open(
+ const MethodContext& context, const string& /*virtualHost*/,
+ const string& /*capabilities*/, bool /*insist*/)
+{
string knownhosts;
- connection.client->getConnection().openOk(context, knownhosts);
+ client.openOk(
+ knownhosts, context.getRequestId());
}
-void BrokerAdapter::BrokerAdapter::ConnectionHandlerImpl::close(
+void BrokerAdapter::ConnectionHandlerImpl::close(
const MethodContext& context, u_int16_t /*replyCode*/, const string& /*replyText*/,
u_int16_t /*classId*/, u_int16_t /*methodId*/)
{
- connection.client->getConnection().closeOk(context);
+ client.closeOk(context.getRequestId());
connection.getOutput().close();
}
-void BrokerAdapter::BrokerAdapter::ConnectionHandlerImpl::closeOk(const MethodContext&){
+void BrokerAdapter::ConnectionHandlerImpl::closeOk(const MethodContext&){
connection.getOutput().close();
}
-void BrokerAdapter::BrokerAdapter::ChannelHandlerImpl::open(
+void BrokerAdapter::ChannelHandlerImpl::open(
const MethodContext& context, const string& /*outOfBand*/){
channel.open();
// FIXME aconway 2007-01-04: provide valid ID as per ampq 0-9
- connection.client->getChannel().openOk(context, std::string()/* ID */);
+ client.openOk(
+ std::string()/* ID */, context.getRequestId());
}
-void BrokerAdapter::BrokerAdapter::ChannelHandlerImpl::flow(const MethodContext&, bool /*active*/){}
-void BrokerAdapter::BrokerAdapter::ChannelHandlerImpl::flowOk(const MethodContext&, bool /*active*/){}
+void BrokerAdapter::ChannelHandlerImpl::flow(const MethodContext&, bool /*active*/){}
+void BrokerAdapter::ChannelHandlerImpl::flowOk(const MethodContext&, bool /*active*/){}
-void BrokerAdapter::BrokerAdapter::ChannelHandlerImpl::close(
+void BrokerAdapter::ChannelHandlerImpl::close(
const MethodContext& context, u_int16_t /*replyCode*/,
const string& /*replyText*/,
u_int16_t /*classId*/, u_int16_t /*methodId*/)
{
- connection.client->getChannel().closeOk(context);
+ client.closeOk(context.getRequestId());
// FIXME aconway 2007-01-18: Following line will "delete this". Ugly.
connection.closeChannel(channel.getId());
}
-void BrokerAdapter::BrokerAdapter::ChannelHandlerImpl::closeOk(const MethodContext&){}
+void BrokerAdapter::ChannelHandlerImpl::closeOk(const MethodContext&){}
-void BrokerAdapter::BrokerAdapter::ExchangeHandlerImpl::declare(const MethodContext& context, u_int16_t /*ticket*/, const string& exchange, const string& type,
- bool passive, bool /*durable*/, bool /*autoDelete*/, bool /*internal*/, bool nowait,
- const FieldTable& /*arguments*/){
+void BrokerAdapter::ExchangeHandlerImpl::declare(const MethodContext& context, u_int16_t /*ticket*/, const string& exchange, const string& type,
+ bool passive, bool /*durable*/, bool /*autoDelete*/, bool /*internal*/, bool nowait,
+ const FieldTable& /*arguments*/){
if(passive){
if(!broker.getExchanges().get(exchange)) {
@@ -116,27 +139,30 @@
}
}
if(!nowait){
- connection.client->getExchange().declareOk(context);
+ client.declareOk(context.getRequestId());
}
}
-void BrokerAdapter::BrokerAdapter::ExchangeHandlerImpl::delete_(const MethodContext& context, u_int16_t /*ticket*/,
- const string& exchange, bool /*ifUnused*/, bool nowait){
+void BrokerAdapter::ExchangeHandlerImpl::delete_(const MethodContext& context, u_int16_t /*ticket*/,
+ const string& exchange, bool /*ifUnused*/, bool nowait){
//TODO: implement unused
broker.getExchanges().destroy(exchange);
- if(!nowait) connection.client->getExchange().deleteOk(context);
+ if(!nowait) client.deleteOk(context.getRequestId());
}
-void BrokerAdapter::BrokerAdapter::QueueHandlerImpl::declare(const MethodContext& context, u_int16_t /*ticket*/, const string& name,
- bool passive, bool durable, bool exclusive,
- bool autoDelete, bool nowait, const qpid::framing::FieldTable& arguments){
+void BrokerAdapter::QueueHandlerImpl::declare(const MethodContext& context, u_int16_t /*ticket*/, const string& name,
+ bool passive, bool durable, bool exclusive,
+ bool autoDelete, bool nowait, const qpid::framing::FieldTable& arguments){
Queue::shared_ptr queue;
if (passive && !name.empty()) {
queue = connection.getQueue(name, channel.getId());
} else {
std::pair<Queue::shared_ptr, bool> queue_created =
- broker.getQueues().declare(name, durable, autoDelete ? connection.settings.timeout : 0, exclusive ? &connection : 0);
+ broker.getQueues().declare(
+ name, durable,
+ autoDelete ? connection.getTimeout() : 0,
+ exclusive ? &connection : 0);
queue = queue_created.first;
assert(queue);
if (queue_created.second) { // This is a new queue
@@ -161,20 +187,22 @@
% queue->getName());
if (!nowait) {
string queueName = queue->getName();
- connection.client->getQueue().declareOk(context, queueName, queue->getMessageCount(), queue->getConsumerCount());
+ client.declareOk(
+ queueName, queue->getMessageCount(), queue->getConsumerCount(),
+ context.getRequestId());
}
}
-void BrokerAdapter::BrokerAdapter::QueueHandlerImpl::bind(const MethodContext& context, u_int16_t /*ticket*/, const string& queueName,
- const string& exchangeName, const string& routingKey, bool nowait,
- const FieldTable& arguments){
+void BrokerAdapter::QueueHandlerImpl::bind(const MethodContext& context, u_int16_t /*ticket*/, const string& queueName,
+ const string& exchangeName, const string& routingKey, bool nowait,
+ const FieldTable& arguments){
Queue::shared_ptr queue = connection.getQueue(queueName, channel.getId());
Exchange::shared_ptr exchange = broker.getExchanges().get(exchangeName);
if(exchange){
string exchangeRoutingKey = routingKey.empty() && queueName.empty() ? queue->getName() : routingKey;
exchange->bind(queue, exchangeRoutingKey, &arguments);
- if(!nowait) connection.client->getQueue().bindOk(context);
+ if(!nowait) client.bindOk(context.getRequestId());
}else{
throw ChannelException(
404, "Bind failed. No such exchange: " + exchangeName);
@@ -182,7 +210,7 @@
}
void
-BrokerAdapter::BrokerAdapter::QueueHandlerImpl::unbind(
+BrokerAdapter::QueueHandlerImpl::unbind(
const MethodContext& context,
u_int16_t /*ticket*/,
const string& queueName,
@@ -198,18 +226,18 @@
exchange->unbind(queue, routingKey, &arguments);
- connection.client->getQueue().unbindOk(context);
+ client.unbindOk(context.getRequestId());
}
-void BrokerAdapter::BrokerAdapter::QueueHandlerImpl::purge(const MethodContext& context, u_int16_t /*ticket*/, const string& queueName, bool nowait){
+void BrokerAdapter::QueueHandlerImpl::purge(const MethodContext& context, u_int16_t /*ticket*/, const string& queueName, bool nowait){
Queue::shared_ptr queue = connection.getQueue(queueName, channel.getId());
int count = queue->purge();
- if(!nowait) connection.client->getQueue().purgeOk(context, count);
+ if(!nowait) client.purgeOk( count, context.getRequestId());
}
-void BrokerAdapter::BrokerAdapter::QueueHandlerImpl::delete_(const MethodContext& context, u_int16_t /*ticket*/, const string& queue,
- bool ifUnused, bool ifEmpty, bool nowait){
+void BrokerAdapter::QueueHandlerImpl::delete_(const MethodContext& context, u_int16_t /*ticket*/, const string& queue,
+ bool ifUnused, bool ifEmpty, bool nowait){
ChannelException error(0, "");
int count(0);
Queue::shared_ptr q = connection.getQueue(queue, channel.getId());
@@ -228,20 +256,21 @@
broker.getQueues().destroy(queue);
}
- if(!nowait) connection.client->getQueue().deleteOk(context, count);
+ if(!nowait)
+ client.deleteOk(count, context.getRequestId());
}
-void BrokerAdapter::BrokerAdapter::BasicHandlerImpl::qos(const MethodContext& context, u_int32_t prefetchSize, u_int16_t prefetchCount, bool /*global*/){
+void BrokerAdapter::BasicHandlerImpl::qos(const MethodContext& context, u_int32_t prefetchSize, u_int16_t prefetchCount, bool /*global*/){
//TODO: handle global
channel.setPrefetchSize(prefetchSize);
channel.setPrefetchCount(prefetchCount);
- connection.client->getBasic().qosOk(context);
+ client.qosOk(context.getRequestId());
}
-void BrokerAdapter::BrokerAdapter::BasicHandlerImpl::consume(
+void BrokerAdapter::BasicHandlerImpl::consume(
const MethodContext& context, u_int16_t /*ticket*/,
const string& queueName, const string& consumerTag,
bool noLocal, bool noAck, bool exclusive,
@@ -257,19 +286,19 @@
channel.consume(
newTag, queue, !noAck, exclusive, noLocal ? &connection : 0, &fields);
- if(!nowait) connection.client->getBasic().consumeOk(context, newTag);
+ if(!nowait) client.consumeOk(newTag, context.getRequestId());
//allow messages to be dispatched if required as there is now a consumer:
queue->dispatch();
}
-void BrokerAdapter::BrokerAdapter::BasicHandlerImpl::cancel(const MethodContext& context, const string& consumerTag, bool nowait){
+void BrokerAdapter::BasicHandlerImpl::cancel(const MethodContext& context, const string& consumerTag, bool nowait){
channel.cancel(consumerTag);
- if(!nowait) connection.client->getBasic().cancelOk(context, consumerTag);
+ if(!nowait) client.cancelOk(consumerTag, context.getRequestId());
}
-void BrokerAdapter::BrokerAdapter::BasicHandlerImpl::publish(
+void BrokerAdapter::BasicHandlerImpl::publish(
const MethodContext& context, u_int16_t /*ticket*/,
const string& exchangeName, const string& routingKey,
bool mandatory, bool immediate)
@@ -287,16 +316,16 @@
}
}
-void BrokerAdapter::BrokerAdapter::BasicHandlerImpl::get(const MethodContext& context, u_int16_t /*ticket*/, const string& queueName, bool noAck){
+void BrokerAdapter::BasicHandlerImpl::get(const MethodContext& context, u_int16_t /*ticket*/, const string& queueName, bool noAck){
Queue::shared_ptr queue = connection.getQueue(queueName, channel.getId());
if(!connection.getChannel(channel.getId()).get(queue, "", !noAck)){
string clusterId;//not used, part of an imatix hack
- connection.client->getBasic().getEmpty(context, clusterId);
+ client.getEmpty(clusterId, context.getRequestId());
}
}
-void BrokerAdapter::BrokerAdapter::BasicHandlerImpl::ack(const MethodContext&, u_int64_t deliveryTag, bool multiple){
+void BrokerAdapter::BasicHandlerImpl::ack(const MethodContext&, u_int64_t deliveryTag, bool multiple){
try{
channel.ack(deliveryTag, multiple);
}catch(InvalidAckException& e){
@@ -304,31 +333,31 @@
}
}
-void BrokerAdapter::BrokerAdapter::BasicHandlerImpl::reject(const MethodContext&, u_int64_t /*deliveryTag*/, bool /*requeue*/){}
+void BrokerAdapter::BasicHandlerImpl::reject(const MethodContext&, u_int64_t /*deliveryTag*/, bool /*requeue*/){}
-void BrokerAdapter::BrokerAdapter::BasicHandlerImpl::recover(const MethodContext&, bool requeue){
+void BrokerAdapter::BasicHandlerImpl::recover(const MethodContext&, bool requeue){
channel.recover(requeue);
}
-void BrokerAdapter::BrokerAdapter::TxHandlerImpl::select(const MethodContext& context){
+void BrokerAdapter::TxHandlerImpl::select(const MethodContext& context){
channel.begin();
- connection.client->getTx().selectOk(context);
+ client.selectOk(context.getRequestId());
}
-void BrokerAdapter::BrokerAdapter::TxHandlerImpl::commit(const MethodContext& context){
+void BrokerAdapter::TxHandlerImpl::commit(const MethodContext& context){
channel.commit();
- connection.client->getTx().commitOk(context);
+ client.commitOk(context.getRequestId());
}
-void BrokerAdapter::BrokerAdapter::TxHandlerImpl::rollback(const MethodContext& context){
+void BrokerAdapter::TxHandlerImpl::rollback(const MethodContext& context){
channel.rollback();
- connection.client->getTx().rollbackOk(context);
+ client.rollbackOk(context.getRequestId());
channel.recover(false);
}
void
-BrokerAdapter::BrokerAdapter::ChannelHandlerImpl::ok( const MethodContext& )
+BrokerAdapter::ChannelHandlerImpl::ok( const MethodContext& )
{
//no specific action required, generic response handling should be sufficient
}
@@ -338,21 +367,21 @@
// Message class method handlers
//
void
-BrokerAdapter::BrokerAdapter::ChannelHandlerImpl::ping( const MethodContext& context)
+BrokerAdapter::ChannelHandlerImpl::ping( const MethodContext& context)
{
- connection.client->getChannel().ok(context);
- connection.client->getChannel().pong(context);
+ client.ok(context.getRequestId());
+ client.pong();
}
void
-BrokerAdapter::BrokerAdapter::ChannelHandlerImpl::pong( const MethodContext& context)
+BrokerAdapter::ChannelHandlerImpl::pong( const MethodContext& context)
{
- connection.client->getChannel().ok(context);
+ client.ok(context.getRequestId());
}
void
-BrokerAdapter::BrokerAdapter::ChannelHandlerImpl::resume(
+BrokerAdapter::ChannelHandlerImpl::resume(
const MethodContext&,
const string& /*channel*/ )
{
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerAdapter.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerAdapter.h?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerAdapter.h (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerAdapter.h Mon Feb 12 18:41:14 2007
@@ -19,8 +19,9 @@
*
*/
#include "AMQP_ServerOperations.h"
+#include "HandlerImpl.h"
#include "MessageHandlerImpl.h"
-#include "BrokerChannel.h"
+#include "Exception.h"
namespace qpid {
namespace broker {
@@ -28,14 +29,6 @@
class Channel;
class Connection;
class Broker;
-
-/**
- * Per-channel protocol adapter.
- *
- * Translates protocol bodies into calls on the core Channel,
- * Connection and Broker objects.
- */
-
class ChannelHandler;
class ConnectionHandler;
class BasicHandler;
@@ -48,20 +41,23 @@
class StreamHandler;
class DtxHandler;
class TunnelHandler;
+class MessageHandlerImpl;
-class BrokerAdapter : public framing::AMQP_ServerOperations
+/**
+ * Per-channel protocol adapter.
+ *
+ * A container for a collection of AMQP-class adapters that translate
+ * AMQP method bodies into calls on the core Channel, Connection and
+ * Broker objects. Each adapter class also provides a client proxy
+ * to send methods to the peer.
+ *
+ */
+class BrokerAdapter : public CoreRefs, public framing::AMQP_ServerOperations
{
public:
- BrokerAdapter(Channel& ch, Connection& c, Broker& b) :
- basicHandler(ch, c, b),
- channelHandler(ch, c, b),
- connectionHandler(ch, c, b),
- exchangeHandler(ch, c, b),
- messageHandler(ch, c, b),
- queueHandler(ch, c, b),
- txHandler(ch, c, b)
- {}
-
+ BrokerAdapter(Channel& ch, Connection& c, Broker& b);
+
+ framing::ProtocolVersion getVersion() const;
ChannelHandler* getChannelHandler() { return &channelHandler; }
ConnectionHandler* getConnectionHandler() { return &connectionHandler; }
BasicHandler* getBasicHandler() { return &basicHandler; }
@@ -80,19 +76,16 @@
TunnelHandler* getTunnelHandler() {
throw ConnectionException(540, "Tunnel class not implemented"); }
+ framing::AMQP_ClientProxy& getProxy() { return proxy; }
+
private:
- struct CoreRefs {
- CoreRefs(Channel& ch, Connection& c, Broker& b)
- : channel(ch), connection(c), broker(b) {}
-
- Channel& channel;
- Connection& connection;
- Broker& broker;
- };
-
- class ConnectionHandlerImpl : private CoreRefs, public ConnectionHandler {
+
+ class ConnectionHandlerImpl :
+ public ConnectionHandler,
+ public HandlerImpl<framing::AMQP_ClientProxy::Connection>
+ {
public:
- ConnectionHandlerImpl(Channel& ch, Connection& c, Broker& b) : CoreRefs(ch, c, b) {}
+ ConnectionHandlerImpl(BrokerAdapter& parent) : HandlerImplType(parent) {}
void startOk(const framing::MethodContext& context,
const qpid::framing::FieldTable& clientProperties,
@@ -112,9 +105,13 @@
void closeOk(const framing::MethodContext& context);
};
- class ChannelHandlerImpl : private CoreRefs, public ChannelHandler{
+ class ChannelHandlerImpl :
+ public ChannelHandler,
+ public HandlerImpl<framing::AMQP_ClientProxy::Channel>
+ {
public:
- ChannelHandlerImpl(Channel& ch, Connection& c, Broker& b) : CoreRefs(ch, c, b) {}
+ ChannelHandlerImpl(BrokerAdapter& parent) : HandlerImplType(parent) {}
+
void open(const framing::MethodContext& context, const std::string& outOfBand);
void flow(const framing::MethodContext& context, bool active);
void flowOk(const framing::MethodContext& context, bool active);
@@ -127,9 +124,13 @@
void closeOk(const framing::MethodContext& context);
};
- class ExchangeHandlerImpl : private CoreRefs, public ExchangeHandler{
+ class ExchangeHandlerImpl :
+ public ExchangeHandler,
+ public HandlerImpl<framing::AMQP_ClientProxy::Exchange>
+ {
public:
- ExchangeHandlerImpl(Channel& ch, Connection& c, Broker& b) : CoreRefs(ch, c, b) {}
+ ExchangeHandlerImpl(BrokerAdapter& parent) : HandlerImplType(parent) {}
+
void declare(const framing::MethodContext& context, u_int16_t ticket,
const std::string& exchange, const std::string& type,
bool passive, bool durable, bool autoDelete,
@@ -139,9 +140,13 @@
const std::string& exchange, bool ifUnused, bool nowait);
};
- class QueueHandlerImpl : private CoreRefs, public QueueHandler{
+ class QueueHandlerImpl :
+ public QueueHandler,
+ public HandlerImpl<framing::AMQP_ClientProxy::Queue>
+ {
public:
- QueueHandlerImpl(Channel& ch, Connection& c, Broker& b) : CoreRefs(ch, c, b) {}
+ QueueHandlerImpl(BrokerAdapter& parent) : HandlerImplType(parent) {}
+
void declare(const framing::MethodContext& context, u_int16_t ticket, const std::string& queue,
bool passive, bool durable, bool exclusive,
bool autoDelete, bool nowait,
@@ -162,9 +167,13 @@
bool nowait);
};
- class BasicHandlerImpl : private CoreRefs, public BasicHandler{
+ class BasicHandlerImpl :
+ public BasicHandler,
+ public HandlerImpl<framing::AMQP_ClientProxy::Basic>
+ {
public:
- BasicHandlerImpl(Channel& ch, Connection& c, Broker& b) : CoreRefs(ch, c, b) {}
+ BasicHandlerImpl(BrokerAdapter& parent) : HandlerImplType(parent) {}
+
void qos(const framing::MethodContext& context, u_int32_t prefetchSize,
u_int16_t prefetchCount, bool global);
void consume(
@@ -184,14 +193,19 @@
void recover(const framing::MethodContext& context, bool requeue);
};
- class TxHandlerImpl : private CoreRefs, public TxHandler{
+ class TxHandlerImpl :
+ public TxHandler,
+ public HandlerImpl<framing::AMQP_ClientProxy::Tx>
+ {
public:
- TxHandlerImpl(Channel& ch, Connection& c, Broker& b) : CoreRefs(ch, c, b) {}
+ TxHandlerImpl(BrokerAdapter& parent) : HandlerImplType(parent) {}
+
void select(const framing::MethodContext& context);
void commit(const framing::MethodContext& context);
void rollback(const framing::MethodContext& context);
};
+ Connection& connection;
BasicHandlerImpl basicHandler;
ChannelHandlerImpl channelHandler;
ConnectionHandlerImpl connectionHandler;
@@ -199,7 +213,7 @@
MessageHandlerImpl messageHandler;
QueueHandlerImpl queueHandler;
TxHandlerImpl txHandler;
-
+
};
}} // namespace qpid::broker
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerChannel.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerChannel.cpp?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerChannel.cpp (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerChannel.cpp Mon Feb 12 18:41:14 2007
@@ -25,6 +25,8 @@
#include <algorithm>
#include <functional>
+#include <boost/bind.hpp>
+
#include "BrokerChannel.h"
#include "DeletingTxOp.h"
#include "framing/ChannelAdapter.h"
@@ -50,7 +52,7 @@
u_int32_t _framesize, MessageStore* const _store,
u_int64_t _stagingThreshold
) :
- ChannelAdapter(id, &con.getOutput(), con.client->getProtocolVersion()),
+ ChannelAdapter(id, &con.getOutput(), con.getVersion()),
connection(con),
currentDeliveryTag(1),
transactional(false),
@@ -74,46 +76,32 @@
return consumers.find(consumerTag) != consumers.end();
}
-void Channel::consume(string& tag, Queue::shared_ptr queue, bool acks,
+// TODO aconway 2007-02-12: Why is connection token passed in instead
+// of using the channel's parent connection?
+void Channel::consume(string& tagInOut, Queue::shared_ptr queue, bool acks,
bool exclusive, ConnectionToken* const connection,
const FieldTable*)
{
- if(tag.empty()) tag = tagGenerator.generate();
- ConsumerImpl* c(new ConsumerImpl(this, tag, queue, connection, acks));
- try{
- queue->consume(c, exclusive);//may throw exception
- consumers[tag] = c;
- } catch(...) {
- // FIXME aconway 2007-02-06: auto_ptr for exception safe mem. mgmt.
- delete c;
- throw;
- }
-}
-
-void Channel::cancel(consumer_iterator i){
- ConsumerImpl* c = i->second;
- consumers.erase(i);
- if(c){
- c->cancel();
- delete c;
- }
+ if(tagInOut.empty())
+ tagInOut = tagGenerator.generate();
+ std::auto_ptr<ConsumerImpl> c(
+ new ConsumerImpl(this, tagInOut, queue, connection, acks));
+ queue->consume(c.get(), exclusive);//may throw exception
+ consumers.insert(tagInOut, c.release());
}
void Channel::cancel(const string& tag){
- consumer_iterator i = consumers.find(tag);
- if(i != consumers.end()){
- cancel(i);
- }
+ // consumers is a ptr_map so erase will delete the consumer
+ // which will call cancel.
+ ConsumerImplMap::iterator i = consumers.find(tag);
+ if (i != consumers.end())
+ consumers.erase(i);
}
void Channel::close(){
- if (isOpen()) {
- opened = false;
- while (!consumers.empty())
- cancel(consumers.begin());
- //requeue:
- recover(true);
- }
+ opened = false;
+ consumers.clear();
+ recover(true);
}
void Channel::begin(){
@@ -160,14 +148,10 @@
}
Channel::ConsumerImpl::ConsumerImpl(Channel* _parent, const string& _tag,
- Queue::shared_ptr _queue,
- ConnectionToken* const _connection, bool ack) : parent(_parent),
- tag(_tag),
- queue(_queue),
- connection(_connection),
- ackExpected(ack),
- blocked(false){
-}
+ Queue::shared_ptr _queue,
+ ConnectionToken* const _connection, bool ack
+) : parent(_parent), tag(_tag), queue(_queue), connection(_connection),
+ ackExpected(ack), blocked(false) {}
bool Channel::ConsumerImpl::deliver(Message::shared_ptr& msg){
if(!connection || connection != msg->getPublisher()){//check for no_local
@@ -182,12 +166,18 @@
return false;
}
+Channel::ConsumerImpl::~ConsumerImpl() {
+ cancel();
+}
+
void Channel::ConsumerImpl::cancel(){
- if(queue) queue->cancel(this);
+ if(queue)
+ queue->cancel(this);
}
void Channel::ConsumerImpl::requestDispatch(){
- if(blocked) queue->dispatch();
+ if(blocked)
+ queue->dispatch();
}
void Channel::handleInlineTransfer(Message::shared_ptr msg)
@@ -196,11 +186,15 @@
connection.broker.getExchanges().get(msg->getExchange());
if(transactional){
TxPublish* deliverable = new TxPublish(msg);
- exchange->route(*deliverable, msg->getRoutingKey(), &(msg->getApplicationHeaders()));
+ exchange->route(
+ *deliverable, msg->getRoutingKey(),
+ &(msg->getApplicationHeaders()));
txBuffer.enlist(new DeletingTxOp(deliverable));
}else{
DeliverableMessage deliverable(msg);
- exchange->route(deliverable, msg->getRoutingKey(), &(msg->getApplicationHeaders()));
+ exchange->route(
+ deliverable, msg->getRoutingKey(),
+ &(msg->getApplicationHeaders()));
}
}
@@ -244,7 +238,8 @@
ack(getRequestInProgress(), false);
}
-void Channel::ack(u_int64_t deliveryTag, bool multiple){
+void Channel::ack(u_int64_t deliveryTag, bool multiple)
+{
if(transactional){
accumulatedAck.update(deliveryTag, multiple);
//TODO: I think the outstanding prefetch size & count should be updated at this point...
@@ -271,9 +266,8 @@
//if the prefetch limit had previously been reached, there may
//be messages that can be now be delivered
- for(consumer_iterator j = consumers.begin(); j != consumers.end(); j++){
- j->second->requestDispatch();
- }
+ std::for_each(consumers.begin(), consumers.end(),
+ boost::bind(&ConsumerImpl::requestDispatch, _1));
}
}
@@ -328,8 +322,8 @@
method->invoke(*adapter, context);
}
}catch(ChannelException& e){
- connection.client->getChannel().close(
- context, e.code, e.toString(),
+ adapter->getProxy().getChannel().close(
+ e.code, e.toString(),
method->amqpClassId(), method->amqpMethodId());
connection.closeChannel(getId());
}catch(ConnectionException& e){
@@ -338,4 +332,3 @@
connection.close(541/*internal error*/, e.what(), method->amqpClassId(), method->amqpMethodId());
}
}
-
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerChannel.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerChannel.h?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerChannel.h (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerChannel.h Mon Feb 12 18:41:14 2007
@@ -23,10 +23,10 @@
*/
#include <list>
-#include <map>
#include <boost/scoped_ptr.hpp>
#include <boost/shared_ptr.hpp>
+#include <boost/ptr_container/ptr_map.hpp>
#include <AccumulatedAck.h>
#include <Consumer.h>
@@ -56,7 +56,7 @@
class Channel : public framing::ChannelAdapter,
public CompletionHandler
{
- class ConsumerImpl : public virtual Consumer
+ class ConsumerImpl : public Consumer
{
Channel* parent;
const string tag;
@@ -64,23 +64,25 @@
ConnectionToken* const connection;
const bool ackExpected;
bool blocked;
+
public:
ConsumerImpl(Channel* parent, const string& tag,
Queue::shared_ptr queue,
ConnectionToken* const connection, bool ack);
+ ~ConsumerImpl();
virtual bool deliver(Message::shared_ptr& msg);
void cancel();
void requestDispatch();
};
- typedef std::map<string,ConsumerImpl*>::iterator consumer_iterator;
+ typedef boost::ptr_map<string,ConsumerImpl> ConsumerImplMap;
Connection& connection;
u_int16_t id;
u_int64_t currentDeliveryTag;
Queue::shared_ptr defaultQueue;
bool transactional;
- std::map<string, ConsumerImpl*> consumers;
+ ConsumerImplMap consumers;
u_int32_t prefetchSize;
u_int16_t prefetchCount;
Prefetch outstanding;
@@ -93,18 +95,17 @@
MessageStore* const store;
MessageBuilder messageBuilder;//builder for in-progress message
bool opened;
-
boost::scoped_ptr<BrokerAdapter> adapter;
// completion handler for MessageBuilder
void complete(Message::shared_ptr msg);
- void deliver(Message::shared_ptr& msg, const string& tag, Queue::shared_ptr& queue, bool ackExpected);
- void cancel(consumer_iterator consumer);
+ void deliver(Message::shared_ptr& msg, const string& tag,
+ Queue::shared_ptr& queue, bool ackExpected);
bool checkPrefetch(Message::shared_ptr& msg);
public:
- Channel(Connection& channel,
+ Channel(Connection& parent,
framing::ChannelId id,
u_int32_t framesize,
MessageStore* const _store = 0,
@@ -112,8 +113,8 @@
~Channel();
- // For ChannelAdapter
bool isOpen() const { return opened; }
+ BrokerAdapter& getAdatper() { return *adapter; }
void open() { opened = true; }
void setDefaultQueue(Queue::shared_ptr queue){ defaultQueue = queue; }
@@ -122,7 +123,11 @@
u_int16_t setPrefetchCount(u_int16_t n){ return prefetchCount = n; }
bool exists(const string& consumerTag);
- void consume(string& tag, Queue::shared_ptr queue, bool acks,
+
+ /**
+ *@param tagInOut - if empty it is updated with the generated token.
+ */
+ void consume(string& tagInOut, Queue::shared_ptr queue, bool acks,
bool exclusive, ConnectionToken* const connection = 0,
const framing::FieldTable* = 0);
void cancel(const string& tag);
@@ -146,7 +151,6 @@
void handleMethodInContext(
boost::shared_ptr<framing::AMQMethodBody> method,
const framing::MethodContext& context);
-
};
struct InvalidAckException{};
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerMessage.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerMessage.cpp?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerMessage.cpp (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerMessage.cpp Mon Feb 12 18:41:14 2007
@@ -28,6 +28,9 @@
#include <MessageStore.h>
#include <BasicDeliverBody.h>
#include <BasicGetOkBody.h>
+#include <AMQContentBody.h>
+#include <AMQHeaderBody.h>
+#include "AMQMethodBody.h"
#include "AMQFrame.h"
#include "framing/ChannelAdapter.h"
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerMessage.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerMessage.h?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerMessage.h (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerMessage.h Mon Feb 12 18:41:14 2007
@@ -22,12 +22,10 @@
*
*/
-#include <BrokerMessageBase.h>
#include <memory>
#include <boost/shared_ptr.hpp>
-#include <AMQContentBody.h>
-#include <AMQHeaderBody.h>
-#include "AMQMethodBody.h"
+
+#include <BrokerMessageBase.h>
#include <BasicHeaderProperties.h>
#include <ConnectionToken.h>
#include <Content.h>
@@ -39,6 +37,7 @@
namespace framing {
class MethodContext;
class ChannelAdapter;
+class AMQHeaderBody;
}
namespace broker {
@@ -52,7 +51,7 @@
* request.
*/
class BasicMessage : public Message {
- framing::AMQHeaderBody::shared_ptr header;
+ boost::shared_ptr<framing::AMQHeaderBody> header;
std::auto_ptr<Content> content;
sys::Mutex contentLock;
u_int64_t size;
@@ -65,10 +64,10 @@
BasicMessage(const ConnectionToken* const publisher,
const string& exchange, const string& routingKey,
bool mandatory, bool immediate,
- framing::AMQMethodBody::shared_ptr respondTo);
+ boost::shared_ptr<framing::AMQMethodBody> respondTo);
BasicMessage();
~BasicMessage();
- void setHeader(framing::AMQHeaderBody::shared_ptr header);
+ void setHeader(boost::shared_ptr<framing::AMQHeaderBody> header);
void addContent(framing::AMQContentBody::shared_ptr data);
bool isComplete();
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerMessageBase.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerMessageBase.h?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerMessageBase.h (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerMessageBase.h Mon Feb 12 18:41:14 2007
@@ -22,14 +22,10 @@
*
*/
-#include "AMQContentBody.h"
-#include "AMQHeaderBody.h"
-#include "AMQMethodBody.h"
-#include "Content.h"
-#include "framing/amqp_types.h"
-
#include <string>
#include <boost/shared_ptr.hpp>
+#include "Content.h"
+#include "framing/amqp_types.h"
namespace qpid {
@@ -38,6 +34,9 @@
class ChannelAdapter;
class BasicHeaderProperties;
class FieldTable;
+class AMQMethodBody;
+class AMQContentBody;
+class AMQHeaderBody;
}
@@ -50,24 +49,17 @@
* abstracting away the operations
* TODO; AMS: for the moment this is mostly a placeholder
*/
-class Message{
- const ConnectionToken* publisher;
- std::string exchange;
- std::string routingKey;
- const bool mandatory;
- const bool immediate;
- u_int64_t persistenceId;
- bool redelivered;
- framing::AMQMethodBody::shared_ptr respondTo;
-
+class Message {
public:
typedef boost::shared_ptr<Message> shared_ptr;
+ typedef boost::shared_ptr<framing::AMQMethodBody> AMQMethodBodyPtr;
+
Message(const ConnectionToken* publisher_,
const std::string& _exchange,
const std::string& _routingKey,
bool _mandatory, bool _immediate,
- framing::AMQMethodBody::shared_ptr respondTo_) :
+ AMQMethodBodyPtr respondTo_) :
publisher(publisher_),
exchange(_exchange),
routingKey(_routingKey),
@@ -92,9 +84,7 @@
const std::string& getExchange() const { return exchange; }
u_int64_t getPersistenceId() const { return persistenceId; }
bool getRedelivered() const { return redelivered; }
- framing::AMQMethodBody::shared_ptr getRespondTo() const {
- return respondTo;
- }
+ AMQMethodBodyPtr getRespondTo() const { return respondTo; }
void setRouting(const std::string& _exchange, const std::string& _routingKey)
{ exchange = _exchange; routingKey = _routingKey; }
@@ -168,14 +158,24 @@
* it uses).
*/
virtual void setContent(std::auto_ptr<Content>& /*content*/) {};
- virtual void setHeader(framing::AMQHeaderBody::shared_ptr /*header*/) {};
- virtual void addContent(framing::AMQContentBody::shared_ptr /*data*/) {};
+ virtual void setHeader(boost::shared_ptr<framing::AMQHeaderBody>) {};
+ virtual void addContent(boost::shared_ptr<framing::AMQContentBody>) {};
/**
* Releases the in-memory content data held by this
* message. Must pass in a store from which the data can
* be reloaded.
*/
virtual void releaseContent(MessageStore* /*store*/) {};
+
+ private:
+ const ConnectionToken* publisher;
+ std::string exchange;
+ std::string routingKey;
+ const bool mandatory;
+ const bool immediate;
+ u_int64_t persistenceId;
+ bool redelivered;
+ AMQMethodBodyPtr respondTo;
};
}}
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerQueue.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerQueue.cpp?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerQueue.cpp (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerQueue.cpp Mon Feb 12 18:41:14 2007
@@ -149,7 +149,9 @@
void Queue::cancel(Consumer* c){
Mutex::ScopedLock locker(lock);
- consumers.erase(find(consumers.begin(), consumers.end(), c));
+ Consumers::iterator i = std::find(consumers.begin(), consumers.end(), c);
+ if (i != consumers.end())
+ consumers.erase(i);
if(autodelete && consumers.empty()) lastUsed = now()*TIME_MSEC;
if(exclusive == c) exclusive = 0;
}
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerQueue.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerQueue.h?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerQueue.h (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerQueue.h Mon Feb 12 18:41:14 2007
@@ -53,13 +53,17 @@
* or more consumers registers.
*/
class Queue{
+ typedef std::vector<Consumer*> Consumers;
+ typedef std::queue<Binding*> Bindings;
+ typedef std::queue<Message::shared_ptr> Messages;
+
const string name;
const u_int32_t autodelete;
MessageStore* const store;
const ConnectionToken* const owner;
- std::vector<Consumer*> consumers;
- std::queue<Binding*> bindings;
- std::queue<Message::shared_ptr> messages;
+ Consumers consumers;
+ Bindings bindings;
+ Messages messages;
bool queueing;
bool dispatching;
int next;
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/broker/Connection.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/broker/Connection.cpp?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/broker/Connection.cpp (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/broker/Connection.cpp Mon Feb 12 18:41:14 2007
@@ -22,6 +22,9 @@
#include <assert.h>
#include "Connection.h"
+#include "BrokerChannel.h"
+#include "AMQP_ClientProxy.h"
+#include "BrokerAdapter.h"
using namespace boost;
using namespace qpid::sys;
@@ -33,12 +36,15 @@
Connection::Connection(ConnectionOutputHandler* out_, Broker& broker_) :
broker(broker_),
- settings(broker.getTimeout(), broker.getStagingThreshold()),
out(out_),
framemax(65536),
- heartbeat(0)
+ heartbeat(0),
+ client(0),
+ timeout(broker.getTimeout()),
+ stagingThreshold(broker.getStagingThreshold())
{}
+
Queue::shared_ptr Connection::getQueue(const string& name, u_int16_t channel){
Queue::shared_ptr queue;
if (name.empty()) {
@@ -59,31 +65,27 @@
}
-void Connection::received(qpid::framing::AMQFrame* frame){
+void Connection::received(framing::AMQFrame* frame){
getChannel(frame->getChannel()).handleBody(frame->getBody());
}
-void Connection::close(ReplyCode code, const string& text, ClassId classId, MethodId methodId){
- client->getConnection().close(MethodContext(&getChannel(0)), code, text, classId, methodId);
+void Connection::close(
+ ReplyCode code, const string& text, ClassId classId, MethodId methodId)
+{
+ client->close(code, text, classId, methodId);
getOutput().close();
}
-// TODO aconway 2007-02-02: Should be delegated to the BrokerAdapter
-// as it is part of the protocol.
-void Connection::initiated(qpid::framing::ProtocolInitiation* header) {
- if (client.get())
- // TODO aconway 2007-01-16: correct error code.
- throw ConnectionException(0, "Connection initiated twice");
- client.reset(new qpid::framing::AMQP_ClientProxy(
- out, header->getMajor(), header->getMinor()));
+void Connection::initiated(framing::ProtocolInitiation* header) {
+ version = ProtocolVersion(header->getMajor(), header->getMinor());
FieldTable properties;
string mechanisms("PLAIN");
string locales("en_US");
- client->getConnection().start(
- MethodContext(&getChannel(0)),
+ getChannel(0).init(0, *out, getVersion());
+ client = &getChannel(0).getAdatper().getProxy().getConnection();
+ client->start(
header->getMajor(), header->getMinor(),
properties, mechanisms, locales);
- getChannel(0).init(0, *out, client->getProtocolVersion());
}
void Connection::idleOut(){}
@@ -103,9 +105,10 @@
}
}
-void Connection::closeChannel(u_int16_t channel) {
- getChannel(channel).close();
- channels.erase(channels.find(channel));
+void Connection::closeChannel(u_int16_t id) {
+ ChannelMap::iterator i = channels.find(id);
+ if (i != channels.end())
+ i->close();
}
@@ -115,7 +118,7 @@
i = channels.insert(
id, new Channel(
*this, id, framemax, broker.getQueues().getStore(),
- settings.stagingThreshold)).first;
+ broker.getStagingThreshold())).first;
}
return *i;
}
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/broker/Connection.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/broker/Connection.h?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/broker/Connection.h (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/broker/Connection.h Mon Feb 12 18:41:14 2007
@@ -27,66 +27,64 @@
#include <boost/ptr_container/ptr_map.hpp>
#include <AMQFrame.h>
-#include <AMQP_ClientProxy.h>
#include <AMQP_ServerOperations.h>
+#include <AMQP_ClientProxy.h>
#include <sys/ConnectionOutputHandler.h>
#include <sys/ConnectionInputHandler.h>
#include <sys/TimeoutHandler.h>
+#include "framing/ProtocolVersion.h"
#include "Broker.h"
#include "Exception.h"
+#include "BrokerChannel.h"
namespace qpid {
namespace broker {
-class Settings {
- public:
- const u_int32_t timeout;//timeout for auto-deleted queues (in ms)
- const u_int64_t stagingThreshold;
-
- Settings(u_int32_t _timeout, u_int64_t _stagingThreshold) : timeout(_timeout), stagingThreshold(_stagingThreshold) {}
-};
+class Channel;
class Connection : public sys::ConnectionInputHandler,
public ConnectionToken
{
public:
Connection(sys::ConnectionOutputHandler* out, Broker& broker);
- // ConnectionInputHandler methods
- void received(framing::AMQFrame* frame);
- void initiated(framing::ProtocolInitiation* header);
- void idleOut();
- void idleIn();
- void closed();
- sys::ConnectionOutputHandler& getOutput() { return *out; }
+ /** Get a channel. Create if it does not already exist */
+ Channel& getChannel(framing::ChannelId channel);
- const framing::ProtocolVersion& getVersion() {
- return client->getProtocolVersion(); }
+ /** Close a channel */
+ void closeChannel(framing::ChannelId channel);
+
+ /** Close the connection */
+ void close(framing::ReplyCode code, const string& text, framing::ClassId classId, framing::MethodId methodId);
+
+ sys::ConnectionOutputHandler& getOutput() const { return *out; }
+ framing::ProtocolVersion getVersion() const { return version; }
u_int32_t getFrameMax() const { return framemax; }
u_int16_t getHeartbeat() const { return heartbeat; }
+ u_int32_t getTimeout() const { return timeout; }
+ u_int64_t getStagingThreshold() const { return stagingThreshold; }
void setFrameMax(u_int32_t fm) { framemax = fm; }
void setHeartbeat(u_int16_t hb) { heartbeat = hb; }
-
- Broker& broker;
- std::auto_ptr<framing::AMQP_ClientProxy> client;
- Settings settings;
-
- std::vector<Queue::shared_ptr> exclusiveQueues;
-
+
/**
* 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.
+ * @exception: ConnectionException if name="" and channel has no default.
*/
Queue::shared_ptr getQueue(const string& name, u_int16_t channel);
- Channel& newChannel(framing::ChannelId channel);
- Channel& getChannel(framing::ChannelId channel);
- void closeChannel(framing::ChannelId channel);
- void close(framing::ReplyCode code, const string& text, framing::ClassId classId, framing::MethodId methodId);
+ Broker& broker;
+ std::vector<Queue::shared_ptr> exclusiveQueues;
+
+ // ConnectionInputHandler methods
+ void received(framing::AMQFrame* frame);
+ void initiated(framing::ProtocolInitiation* header);
+ void idleOut();
+ void idleIn();
+ void closed();
private:
typedef boost::ptr_map<framing::ChannelId, Channel> ChannelMap;
@@ -94,10 +92,15 @@
typedef std::vector<Queue::shared_ptr>::iterator queue_iterator;
Exchange::shared_ptr findExchange(const string& name);
+ framing::ProtocolVersion version;
ChannelMap channels;
sys::ConnectionOutputHandler* out;
u_int32_t framemax;
u_int16_t heartbeat;
+ framing::AMQP_ClientProxy::Connection* client;
+ const u_int32_t timeout; //timeout for auto-deleted queues (in ms)
+ const u_int64_t stagingThreshold;
+
};
}}
Added: incubator/qpid/branches/qpid.0-9/cpp/lib/broker/HandlerImpl.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/broker/HandlerImpl.h?view=auto&rev=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/broker/HandlerImpl.h (added)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/broker/HandlerImpl.h Mon Feb 12 18:41:14 2007
@@ -0,0 +1,71 @@
+#ifndef _broker_HandlerImpl_h
+#define _broker_HandlerImpl_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 "BrokerChannel.h"
+#include "AMQP_ClientProxy.h"
+
+namespace qpid {
+
+namespace framing {
+class AMQP_ClientProxy;
+}
+
+namespace broker {
+
+class Broker;
+class Channel;
+class Connection;
+
+/**
+ * A collection of references to the core objects required by an adapter,
+ * and a client proxy.
+ */
+struct CoreRefs
+{
+ CoreRefs(Channel& ch, Connection& c, Broker& b)
+ : channel(ch), connection(c), broker(b), proxy(ch) {}
+
+ Channel& channel;
+ Connection& connection;
+ Broker& broker;
+ framing::AMQP_ClientProxy proxy;
+};
+
+
+/**
+ * Base template for protocol handler implementations.
+ * Provides the core references and appropriate AMQP class proxy.
+ */
+template <class ProxyType>
+struct HandlerImpl : public CoreRefs {
+ typedef HandlerImpl<ProxyType> HandlerImplType;
+ HandlerImpl(CoreRefs& parent)
+ : CoreRefs(parent), client(ProxyType::get(proxy)) {}
+ ProxyType client;
+};
+
+
+
+}} // namespace qpid::broker
+
+
+
+#endif /*!_broker_HandlerImpl_h*/
Propchange: incubator/qpid/branches/qpid.0-9/cpp/lib/broker/HandlerImpl.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/branches/qpid.0-9/cpp/lib/broker/HandlerImpl.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/broker/MessageHandlerImpl.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/broker/MessageHandlerImpl.cpp?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/broker/MessageHandlerImpl.cpp (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/broker/MessageHandlerImpl.cpp Mon Feb 12 18:41:14 2007
@@ -25,16 +25,15 @@
#include "BrokerMessageMessage.h"
#include "MessageAppendBody.h"
#include "MessageTransferBody.h"
+#include "BrokerAdapter.h"
namespace qpid {
namespace broker {
using namespace framing;
-MessageHandlerImpl::MessageHandlerImpl(Channel& ch, Connection& c, Broker& b)
- : channel(ch), connection(c), broker(b), references(ch),
- client(connection.client->getMessage())
-{}
+MessageHandlerImpl::MessageHandlerImpl(CoreRefs& parent)
+ : HandlerImplType(parent), references(channel) {}
//
// Message class method handlers
@@ -47,7 +46,7 @@
references.get(reference).append(
boost::shared_polymorphic_downcast<MessageAppendBody>(
context.methodBody));
- client.ok(context);
+ client.ok(context.getRequestId());
}
@@ -56,7 +55,7 @@
const string& destination )
{
channel.cancel(destination);
- client.ok(context);
+ client.ok(context.getRequestId());
}
void
@@ -73,7 +72,7 @@
const string& reference)
{
references.get(reference).close();
- client.ok(context);
+ client.ok(context.getRequestId());
}
void
@@ -84,7 +83,7 @@
bool noLocal,
bool noAck,
bool exclusive,
- const qpid::framing::FieldTable& filter )
+ const framing::FieldTable& filter )
{
Queue::shared_ptr queue = connection.getQueue(queueName, channel.getId());
if(!destination.empty() && channel.exists(destination))
@@ -93,7 +92,7 @@
channel.consume(
tag, queue, !noAck, exclusive,
noLocal ? &connection : 0, &filter);
- client.ok(context);
+ client.ok(context.getRequestId());
// Dispatch messages as there is now a consumer.
queue->dispatch();
}
@@ -117,9 +116,9 @@
connection.getQueue(queueName, context.channel->getId());
if(channel.get(queue, destination, !noAck))
- client.ok(context);
+ client.ok(context.getRequestId());
else
- client.empty(context);
+ client.empty(context.getRequestId());
}
void
@@ -141,7 +140,7 @@
const string& reference)
{
references.open(reference);
- client.ok(context);
+ client.ok(context.getRequestId());
}
void
@@ -153,7 +152,7 @@
//TODO: handle global
channel.setPrefetchSize(prefetchSize);
channel.setPrefetchCount(prefetchCount);
- client.ok(context);
+ client.ok(context.getRequestId());
}
void
@@ -161,7 +160,7 @@
bool requeue)
{
channel.recover(requeue);
- client.ok(context);
+ client.ok(context.getRequestId());
}
void
@@ -204,8 +203,8 @@
const string& /*appId*/,
const string& /*transactionId*/,
const string& /*securityToken*/,
- const qpid::framing::FieldTable& /*applicationHeaders*/,
- qpid::framing::Content body,
+ const framing::FieldTable& /*applicationHeaders*/,
+ const framing::Content& body,
bool /*mandatory*/)
{
MessageTransferBody::shared_ptr transfer(
@@ -218,7 +217,7 @@
channel.handleInlineTransfer(message);
else
references.get(body.getValue()).addMessage(message);
- client.ok(context);
+ client.ok(context.getRequestId());
}
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/broker/MessageHandlerImpl.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/broker/MessageHandlerImpl.h?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/broker/MessageHandlerImpl.h (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/broker/MessageHandlerImpl.h Mon Feb 12 18:41:14 2007
@@ -24,7 +24,7 @@
#include "AMQP_ServerOperations.h"
#include "AMQP_ClientProxy.h"
#include "Reference.h"
-#include "BrokerChannel.h"
+#include "HandlerImpl.h"
namespace qpid {
namespace broker {
@@ -34,10 +34,11 @@
class MessageMessage;
class MessageHandlerImpl :
- public framing::AMQP_ServerOperations::MessageHandler
+ public framing::AMQP_ServerOperations::MessageHandler,
+ public HandlerImpl<framing::AMQP_ClientProxy::Message>
{
public:
- MessageHandlerImpl(Channel& ch, Connection& c, Broker& b);
+ MessageHandlerImpl(CoreRefs& parent);
void append(const framing::MethodContext&,
const std::string& reference,
@@ -116,14 +117,10 @@
const std::string& transactionId,
const std::string& securityToken,
const framing::FieldTable& applicationHeaders,
- framing::Content body,
+ const framing::Content& body,
bool mandatory );
private:
- Channel& channel;
- Connection& connection;
- Broker& broker;
ReferenceRegistry references;
- framing::AMQP_ClientProxy::Message& client;
};
}} // namespace qpid::broker
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/client/ClientChannel.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/client/ClientChannel.cpp?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/client/ClientChannel.cpp (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/client/ClientChannel.cpp Mon Feb 12 18:41:14 2007
@@ -24,6 +24,7 @@
#include <QpidError.h>
#include <MethodBodyInstances.h>
#include "Connection.h"
+#include "AMQP_ServerProxy.h"
// FIXME aconway 2007-01-26: Evaluate all throws, ensure consistent
// handling of errors that should close the connection or the channel.
@@ -48,12 +49,23 @@
close();
}
+AMQP_ServerProxy& Channel::brokerProxy() {
+ assert(proxy.get());
+ return *proxy;
+}
+
+AMQMethodBody::shared_ptr Channel::brokerResponse() {
+ // FIXME aconway 2007-02-08: implement responses.
+ return AMQMethodBody::shared_ptr();
+}
+
void Channel::open(ChannelId id, Connection& con)
{
if (isOpen())
THROW_QPID_ERROR(INTERNAL_ERROR, "Attempt to re-open channel "+id);
connection = &con;
init(id, con, con.getVersion()); // ChannelAdapter initialization.
+ proxy.reset(new AMQP_ServerProxy(*this));
string oob;
if (id != 0)
sendAndReceive<ChannelOpenOkBody>(new ChannelOpenBody(version, oob));
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/client/ClientChannel.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/client/ClientChannel.h?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/client/ClientChannel.h (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/client/ClientChannel.h Mon Feb 12 18:41:14 2007
@@ -24,6 +24,7 @@
#include <map>
#include <string>
#include <queue>
+#include <boost/scoped_ptr.hpp>
#include "sys/types.h"
#include <framing/amqp_framing.h>
@@ -39,8 +40,10 @@
#include "Thread.h"
namespace qpid {
+
namespace framing {
class ChannelCloseBody;
+class AMQP_ServerProxy;
}
namespace client {
@@ -102,6 +105,7 @@
u_int16_t prefetch;
const bool transactional;
framing::ProtocolVersion version;
+ boost::scoped_ptr<framing::AMQP_ServerProxy> proxy;
void enqueue();
void retrieve(Message& msg);
@@ -151,8 +155,6 @@
public:
- bool isOpen() const;
-
/**
* Creates a channel object.
*
@@ -358,9 +360,21 @@
* @see publish()
*/
void setReturnedMessageHandler(ReturnedMessageHandler* handler);
+
+ bool isOpen() const;
+
+ /**
+ * Returns a proxy for the "raw" AMQP broker protocol. Only for use by
+ * protocol experts.
+ */
+
+ framing::AMQP_ServerProxy& brokerProxy();
+ /**
+ * Wait for the next method from the broker.
+ */
+ framing::AMQMethodBody::shared_ptr brokerResponse();
};
-}
-}
+}}
#endif /*!_client_ClientChannel_h*/
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/client/Connection.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/client/Connection.cpp?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/client/Connection.cpp (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/client/Connection.cpp Mon Feb 12 18:41:14 2007
@@ -43,7 +43,7 @@
Connection::Connection(
bool _debug, u_int32_t _max_frame_size,
- const framing::ProtocolVersion& _version
+ framing::ProtocolVersion _version
) : version(_version), max_frame_size(_max_frame_size),
defaultConnector(version, _debug, _max_frame_size),
isOpen(false), debug(_debug)
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/client/Connection.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/client/Connection.h?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/client/Connection.h (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/client/Connection.h Mon Feb 12 18:41:14 2007
@@ -96,12 +96,12 @@
Connector* connector;
framing::OutputHandler* out;
volatile bool isOpen;
+ Channel channel0;
+ bool debug;
void erase(framing::ChannelId);
void channelException(
Channel&, framing::AMQMethodBody*, const QpidError&);
- Channel channel0;
- bool debug;
// TODO aconway 2007-01-26: too many friendships, untagle these classes.
friend class Channel;
@@ -120,9 +120,8 @@
* @param max_frame_size the maximum frame size that the
* client will accept. Optional and defaults to 65536.
*/
- Connection(
- bool debug = false, u_int32_t max_frame_size = 65536,
- const framing::ProtocolVersion& = framing::highestProtocolVersion);
+ Connection(bool debug = false, u_int32_t max_frame_size = 65536,
+ framing::ProtocolVersion=framing::highestProtocolVersion);
~Connection();
/**
@@ -185,7 +184,7 @@
inline u_int32_t getMaxFrameSize(){ return max_frame_size; }
/** @return protocol version in use on this connection. */
- const framing::ProtocolVersion& getVersion() const { return version; }
+ framing::ProtocolVersion getVersion() const { return version; }
};
}} // namespace qpid::client
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/client/Connector.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/client/Connector.cpp?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/client/Connector.cpp (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/client/Connector.cpp Mon Feb 12 18:41:14 2007
@@ -23,17 +23,19 @@
#include <sys/Time.h>
#include "Connector.h"
+namespace qpid {
+namespace client {
+
using namespace qpid::sys;
-using namespace qpid::client;
using namespace qpid::framing;
using qpid::QpidError;
-Connector::Connector(const qpid::framing::ProtocolVersion& pVersion,
- bool _debug, u_int32_t buffer_size) :
- debug(_debug),
+Connector::Connector(
+ ProtocolVersion ver, bool _debug, u_int32_t buffer_size
+) : debug(_debug),
receive_buffer_size(buffer_size),
send_buffer_size(buffer_size),
- version(pVersion),
+ version(ver),
closed(true),
lastIn(0), lastOut(0),
timeout(0),
@@ -180,3 +182,5 @@
handleClosed();
}
}
+
+}} // namespace qpid::client
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/client/Connector.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/client/Connector.h?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/client/Connector.h (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/client/Connector.h Mon Feb 12 18:41:14 2007
@@ -77,7 +77,7 @@
friend class Channel;
public:
- Connector(const framing::ProtocolVersion& pVersion,
+ Connector(framing::ProtocolVersion pVersion,
bool debug = false, u_int32_t buffer_size = 1024);
virtual ~Connector();
virtual void connect(const std::string& host, int port);
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/common/Makefile.am
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/common/Makefile.am?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/common/Makefile.am (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/common/Makefile.am Mon Feb 12 18:41:14 2007
@@ -63,6 +63,7 @@
$(framing)/AMQHeaderBody.cpp \
$(framing)/AMQHeartbeatBody.cpp \
$(framing)/AMQMethodBody.cpp \
+ $(framing)/MethodContext.cpp \
$(framing)/BasicHeaderProperties.cpp \
$(framing)/BodyHandler.cpp \
$(framing)/ChannelAdapter.cpp \
@@ -76,8 +77,8 @@
$(framing)/Requester.cpp \
$(framing)/Responder.cpp \
$(framing)/Value.cpp \
+ $(framing)/Proxy.cpp \
$(gen)/AMQP_ClientProxy.cpp \
- $(gen)/AMQP_HighestVersion.h \
$(gen)/AMQP_MethodVersionMap.cpp \
$(gen)/AMQP_ServerProxy.cpp \
Exception.cpp \
@@ -87,6 +88,7 @@
sys/Time.cpp
nobase_pkginclude_HEADERS = \
+ $(gen)/AMQP_HighestVersion.h \
$(platform_hdr) \
$(framing)/AMQBody.h \
$(framing)/AMQContentBody.h \
@@ -95,6 +97,7 @@
$(framing)/AMQHeaderBody.h \
$(framing)/AMQHeartbeatBody.h \
$(framing)/AMQMethodBody.h \
+ $(framing)/MethodContext.h \
$(framing)/BasicHeaderProperties.h \
$(framing)/BodyHandler.h \
$(framing)/ChannelAdapter.h \
@@ -111,6 +114,7 @@
$(framing)/Value.h \
$(framing)/amqp_framing.h \
$(framing)/amqp_types.h \
+ $(framing)/Proxy.h \
Exception.h \
ExceptionHolder.h \
QpidError.h \
@@ -121,9 +125,9 @@
sys/Monitor.h \
sys/Mutex.h \
sys/Runnable.h \
- sys/ConnectionOutputHandler.h \
- sys/ConnectionInputHandler.h \
- sys/ConnectionInputHandlerFactory.h \
+ sys/ConnectionOutputHandler.h \
+ sys/ConnectionInputHandler.h \
+ sys/ConnectionInputHandlerFactory.h \
sys/ShutdownHandler.h \
sys/Socket.h \
sys/Thread.h \
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/AMQFrame.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/AMQFrame.cpp?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/AMQFrame.cpp (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/AMQFrame.cpp Mon Feb 12 18:41:14 2007
@@ -26,19 +26,24 @@
#include "AMQRequestBody.h"
#include "AMQResponseBody.h"
-using namespace qpid::framing;
+
+namespace qpid {
+namespace framing {
+
AMQP_MethodVersionMap AMQFrame::versionMap;
-AMQFrame::AMQFrame(const qpid::framing::ProtocolVersion& _version):
+AMQFrame::AMQFrame(ProtocolVersion _version):
version(_version)
- {}
+ {
+ assert(version != ProtocolVersion(0,0));
+ }
-AMQFrame::AMQFrame(const qpid::framing::ProtocolVersion& _version, u_int16_t _channel, AMQBody* _body) :
+AMQFrame::AMQFrame(ProtocolVersion _version, u_int16_t _channel, AMQBody* _body) :
version(_version), channel(_channel), body(_body)
{}
-AMQFrame::AMQFrame(const qpid::framing::ProtocolVersion& _version, u_int16_t _channel, const AMQBody::shared_ptr& _body) :
+AMQFrame::AMQFrame(ProtocolVersion _version, u_int16_t _channel, const AMQBody::shared_ptr& _body) :
version(_version), channel(_channel), body(_body)
{}
@@ -119,7 +124,7 @@
body->decode(buffer, size);
}
-std::ostream& qpid::framing::operator<<(std::ostream& out, const AMQFrame& t)
+std::ostream& operator<<(std::ostream& out, const AMQFrame& t)
{
out << "Frame[channel=" << t.channel << "; ";
if (t.body.get() == 0)
@@ -130,3 +135,5 @@
return out;
}
+
+}} // namespace qpid::framing
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/AMQFrame.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/AMQFrame.h?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/AMQFrame.h (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/AMQFrame.h Mon Feb 12 18:41:14 2007
@@ -41,9 +41,9 @@
class AMQFrame : public AMQDataBlock
{
public:
- AMQFrame(const qpid::framing::ProtocolVersion& _version = highestProtocolVersion);
- AMQFrame(const qpid::framing::ProtocolVersion& _version, u_int16_t channel, AMQBody* body);
- AMQFrame(const qpid::framing::ProtocolVersion& _version, u_int16_t channel, const AMQBody::shared_ptr& body);
+ AMQFrame(ProtocolVersion _version = highestProtocolVersion);
+ AMQFrame(ProtocolVersion _version, u_int16_t channel, AMQBody* body);
+ AMQFrame(ProtocolVersion _version, u_int16_t channel, const AMQBody::shared_ptr& body);
virtual ~AMQFrame();
virtual void encode(Buffer& buffer);
virtual bool decode(Buffer& buffer);
@@ -62,7 +62,7 @@
private:
static AMQP_MethodVersionMap versionMap;
- qpid::framing::ProtocolVersion version;
+ ProtocolVersion version;
u_int16_t channel;
u_int8_t type;
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/ChannelAdapter.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/ChannelAdapter.cpp?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/ChannelAdapter.cpp (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/ChannelAdapter.cpp Mon Feb 12 18:41:14 2007
@@ -19,6 +19,7 @@
#include "ChannelAdapter.h"
#include "AMQFrame.h"
+#include "Exception.h"
using boost::format;
@@ -26,7 +27,7 @@
namespace framing {
void ChannelAdapter::init(
- ChannelId i, OutputHandler& o, const ProtocolVersion& v)
+ ChannelId i, OutputHandler& o, ProtocolVersion v)
{
assertChannelNotOpen();
id = i;
@@ -34,13 +35,15 @@
version = v;
}
-void ChannelAdapter::send(AMQBody::shared_ptr body) {
+RequestId ChannelAdapter::send(AMQBody::shared_ptr body) {
+ RequestId result = 0;
assertChannelOpen();
switch (body->type()) {
case REQUEST_BODY: {
AMQRequestBody::shared_ptr request =
boost::shared_polymorphic_downcast<AMQRequestBody>(body);
requester.sending(request->getData());
+ result = request->getData().requestId;
break;
}
case RESPONSE_BODY: {
@@ -51,6 +54,7 @@
}
}
out->send(new AMQFrame(getVersion(), getId(), body));
+ return result;
}
void ChannelAdapter::handleRequest(AMQRequestBody::shared_ptr request) {
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/ChannelAdapter.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/ChannelAdapter.h?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/ChannelAdapter.h (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/ChannelAdapter.h Mon Feb 12 18:41:14 2007
@@ -35,9 +35,8 @@
class MethodContext;
/**
- * Base class for client and broker channel adapters.
+ * Base class for client and broker channels.
*
- * BodyHandler::handl*
* - receives frame bodies from the network.
* - Updates request/response data.
* - Dispatches requests with a MethodContext for responses.
@@ -55,21 +54,21 @@
*@param output Processed frames are forwarded to this handler.
*/
ChannelAdapter(ChannelId id_=0, OutputHandler* out_=0,
- const ProtocolVersion& ver=ProtocolVersion())
+ ProtocolVersion ver=ProtocolVersion())
: id(id_), out(out_), version(ver) {}
/** Initialize the channel adapter. */
- void init(ChannelId, OutputHandler&, const ProtocolVersion&);
+ void init(ChannelId, OutputHandler&, ProtocolVersion);
ChannelId getId() const { return id; }
- const ProtocolVersion& getVersion() const { return version; }
+ ProtocolVersion getVersion() const { return version; }
/**
* Wrap body in a frame and send the frame.
* Takes ownership of body.
*/
- void send(AMQBody::shared_ptr body);
- void send(AMQBody* body) { send(AMQBody::shared_ptr(body)); }
+ RequestId send(AMQBody::shared_ptr body);
+ RequestId send(AMQBody* body) { return send(AMQBody::shared_ptr(body)); }
void handleMethod(boost::shared_ptr<qpid::framing::AMQMethodBody>);
void handleRequest(boost::shared_ptr<qpid::framing::AMQRequestBody>);
@@ -95,7 +94,7 @@
ProtocolVersion version;
Requester requester;
Responder responder;
- RequestId requestInProgress; // TODO aconway 2007-01-24: use it.
+ RequestId requestInProgress;
};
}}
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/FramingContent.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/FramingContent.h?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/FramingContent.h (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/FramingContent.h Mon Feb 12 18:41:14 2007
@@ -27,9 +27,9 @@
void encode(Buffer& buffer) const;
void decode(Buffer& buffer);
size_t size() const;
- bool isInline() { return discriminator == INLINE; }
- bool isReference() { return discriminator == REFERENCE; }
- const string& getValue() { return value; }
+ bool isInline() const { return discriminator == INLINE; }
+ bool isReference() const { return discriminator == REFERENCE; }
+ const string& getValue() const { return value; }
friend std::ostream& operator<<(std::ostream&, const Content&);
};
Added: incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/MethodContext.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/MethodContext.cpp?view=auto&rev=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/MethodContext.cpp (added)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/MethodContext.cpp Mon Feb 12 18:41:14 2007
@@ -0,0 +1,31 @@
+/*
+ *
+ * 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 "MethodContext.h"
+#include "amqp_types.h"
+#include "AMQRequestBody.h"
+
+namespace qpid {
+namespace framing {
+
+RequestId MethodContext::getRequestId() const {
+ return boost::shared_polymorphic_downcast<AMQRequestBody>(methodBody)
+ ->getRequestId();
+}
+
+}} // namespace qpid::framing
Propchange: incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/MethodContext.cpp
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/MethodContext.cpp
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/MethodContext.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/MethodContext.h?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/MethodContext.h (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/MethodContext.h Mon Feb 12 18:41:14 2007
@@ -24,8 +24,6 @@
#include "OutputHandler.h"
#include "ProtocolVersion.h"
-#include <boost/shared_ptr.hpp>
-
namespace qpid {
namespace framing {
@@ -61,6 +59,12 @@
* It's also provides the request ID when constructing a response.
*/
BodyPtr methodBody;
+
+ /**
+ * Return methodBody's request ID.
+ * It is an error to call this if methodBody is not a request.
+ */
+ RequestId getRequestId() const;
};
// FIXME aconway 2007-02-01: Method context only required on Handler
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/ProtocolInitiation.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/ProtocolInitiation.cpp?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/ProtocolInitiation.cpp (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/ProtocolInitiation.cpp Mon Feb 12 18:41:14 2007
@@ -20,15 +20,18 @@
*/
#include <ProtocolInitiation.h>
-qpid::framing::ProtocolInitiation::ProtocolInitiation(){}
+namespace qpid {
+namespace framing {
-qpid::framing::ProtocolInitiation::ProtocolInitiation(u_int8_t _major, u_int8_t _minor) : version(_major, _minor) {}
+ProtocolInitiation::ProtocolInitiation(){}
-qpid::framing::ProtocolInitiation::ProtocolInitiation(const qpid::framing::ProtocolVersion& p) : version(p) {}
+ProtocolInitiation::ProtocolInitiation(u_int8_t _major, u_int8_t _minor) : version(_major, _minor) {}
-qpid::framing::ProtocolInitiation::~ProtocolInitiation(){}
+ProtocolInitiation::ProtocolInitiation(ProtocolVersion p) : version(p) {}
-void qpid::framing::ProtocolInitiation::encode(Buffer& buffer){
+ProtocolInitiation::~ProtocolInitiation(){}
+
+void ProtocolInitiation::encode(Buffer& buffer){
buffer.putOctet('A');
buffer.putOctet('M');
buffer.putOctet('Q');
@@ -39,7 +42,7 @@
buffer.putOctet(version.getMinor());
}
-bool qpid::framing::ProtocolInitiation::decode(Buffer& buffer){
+bool ProtocolInitiation::decode(Buffer& buffer){
if(buffer.available() >= 8){
buffer.getOctet();//A
buffer.getOctet();//M
@@ -56,3 +59,5 @@
}
//TODO: this should prbably be generated from the spec at some point to keep the version numbers up to date
+
+}} // namespace qpid::framing
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/ProtocolInitiation.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/ProtocolInitiation.h?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/ProtocolInitiation.h (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/ProtocolInitiation.h Mon Feb 12 18:41:14 2007
@@ -37,14 +37,14 @@
public:
ProtocolInitiation();
ProtocolInitiation(u_int8_t major, u_int8_t minor);
- ProtocolInitiation(const ProtocolVersion& p);
+ ProtocolInitiation(ProtocolVersion p);
virtual ~ProtocolInitiation();
virtual void encode(Buffer& buffer);
virtual bool decode(Buffer& buffer);
inline virtual u_int32_t size() const { return 8; }
inline u_int8_t getMajor() const { return version.getMajor(); }
inline u_int8_t getMinor() const { return version.getMinor(); }
- inline const ProtocolVersion& getVersion() const { return version; }
+ inline ProtocolVersion getVersion() const { return version; }
};
}
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/ProtocolVersion.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/ProtocolVersion.cpp?view=diff&rev=506823&r1=506822&r2=506823
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/ProtocolVersion.cpp (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/common/framing/ProtocolVersion.cpp Mon Feb 12 18:41:14 2007
@@ -20,37 +20,9 @@
*/
#include <ProtocolVersion.h>
#include <sstream>
-#include "AMQP_HighestVersion.h"
using namespace qpid::framing;
-ProtocolVersion::ProtocolVersion() {
- *this = highestProtocolVersion;
-}
-
-ProtocolVersion::ProtocolVersion(u_int8_t _major, u_int8_t _minor) :
- major_(_major),
- minor_(_minor)
-{}
-
-ProtocolVersion::ProtocolVersion(const ProtocolVersion::ProtocolVersion& p):
- major_(p.major_),
- minor_(p.minor_)
-{}
-
-ProtocolVersion::~ProtocolVersion()
-{}
-
-bool ProtocolVersion::equals(u_int8_t _major, u_int8_t _minor) const
-{
- return major_ == _major && minor_ == _minor;
-}
-
-bool ProtocolVersion::equals(const ProtocolVersion::ProtocolVersion& p) const
-{
- return major_ == p.major_ && minor_ == p.minor_;
-}
-
const std::string ProtocolVersion::toString() const
{
std::stringstream ss;
@@ -58,10 +30,15 @@
return ss.str();
}
-ProtocolVersion::ProtocolVersion ProtocolVersion::operator=(const ProtocolVersion& p)
+ProtocolVersion& ProtocolVersion::operator=(ProtocolVersion p)
{
major_ = p.major_;
minor_ = p.minor_;
return *this;
+}
+
+bool ProtocolVersion::operator==(ProtocolVersion p) const
+{
+ return major_ == p.major_ && minor_ == p.minor_;
}