summaryrefslogtreecommitdiff
path: root/cpp/src/qpid/broker
diff options
context:
space:
mode:
authorKim van der Riet <kpvdr@apache.org>2006-11-22 16:57:35 +0000
committerKim van der Riet <kpvdr@apache.org>2006-11-22 16:57:35 +0000
commitd46ac2955c4871c9f22067f47490095e2c5f1806 (patch)
tree7e76ef7e4ca47e4cc57c83f7950bf97c3eceb210 /cpp/src/qpid/broker
parent018723f3889e9a1f63585dddba8eecff1d168501 (diff)
downloadqpid-python-d46ac2955c4871c9f22067f47490095e2c5f1806.tar.gz
Merged AMQP version-sensitive generated files with C++ trunk. Phase 1 of merge complete - all locations where version info is required in the framing, broker and client code, the version has been hard-coded to mahor=8, minor=0. Next step: make broker and client version-aware.
git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@478237 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'cpp/src/qpid/broker')
-rw-r--r--cpp/src/qpid/broker/Channel.cpp7
-rw-r--r--cpp/src/qpid/broker/Channel.h6
-rw-r--r--cpp/src/qpid/broker/DirectExchange.cpp6
-rw-r--r--cpp/src/qpid/broker/DirectExchange.h6
-rw-r--r--cpp/src/qpid/broker/Exchange.h6
-rw-r--r--cpp/src/qpid/broker/ExchangeBinding.cpp2
-rw-r--r--cpp/src/qpid/broker/ExchangeBinding.h4
-rw-r--r--cpp/src/qpid/broker/FanOutExchange.cpp6
-rw-r--r--cpp/src/qpid/broker/FanOutExchange.h6
-rw-r--r--cpp/src/qpid/broker/HeadersExchange.cpp6
-rw-r--r--cpp/src/qpid/broker/HeadersExchange.h6
-rw-r--r--cpp/src/qpid/broker/Message.cpp13
-rw-r--r--cpp/src/qpid/broker/Message.h1
-rw-r--r--cpp/src/qpid/broker/SessionHandlerImpl.cpp64
-rw-r--r--cpp/src/qpid/broker/SessionHandlerImpl.h70
-rw-r--r--cpp/src/qpid/broker/TopicExchange.cpp6
-rw-r--r--cpp/src/qpid/broker/TopicExchange.h6
17 files changed, 127 insertions, 94 deletions
diff --git a/cpp/src/qpid/broker/Channel.cpp b/cpp/src/qpid/broker/Channel.cpp
index f3624b4f3d..2894e294e0 100644
--- a/cpp/src/qpid/broker/Channel.cpp
+++ b/cpp/src/qpid/broker/Channel.cpp
@@ -54,8 +54,7 @@ bool Channel::exists(const string& consumerTag){
}
void Channel::consume(string& tag, Queue::shared_ptr queue, bool acks, bool exclusive, ConnectionToken* const connection){
- if(tag.empty()) tag = tagGenerator.generate();
-
+ if(tag.empty()) tag = tagGenerator.generate();
ConsumerImpl* c(new ConsumerImpl(this, tag, queue, connection, acks));
try{
queue->consume(c, exclusive);//may throw exception
@@ -109,7 +108,7 @@ void Channel::rollback(){
accumulatedAck.clear();
}
-void Channel::deliver(Message::shared_ptr& msg, string& consumerTag, Queue::shared_ptr& queue, bool ackExpected){
+void Channel::deliver(Message::shared_ptr& msg, const string& consumerTag, Queue::shared_ptr& queue, bool ackExpected){
Mutex::ScopedLock locker(deliveryLock);
u_int64_t deliveryTag = currentDeliveryTag++;
@@ -129,7 +128,7 @@ bool Channel::checkPrefetch(Message::shared_ptr& msg){
return countOk && sizeOk;
}
-Channel::ConsumerImpl::ConsumerImpl(Channel* _parent, string& _tag,
+Channel::ConsumerImpl::ConsumerImpl(Channel* _parent, const string& _tag,
Queue::shared_ptr _queue,
ConnectionToken* const _connection, bool ack) : parent(_parent),
tag(_tag),
diff --git a/cpp/src/qpid/broker/Channel.h b/cpp/src/qpid/broker/Channel.h
index 6ed867284c..42f65f2c7c 100644
--- a/cpp/src/qpid/broker/Channel.h
+++ b/cpp/src/qpid/broker/Channel.h
@@ -57,13 +57,13 @@ namespace qpid {
class Channel : private MessageBuilder::CompletionHandler{
class ConsumerImpl : public virtual Consumer{
Channel* parent;
- string tag;
+ const string tag;
Queue::shared_ptr queue;
ConnectionToken* const connection;
const bool ackExpected;
bool blocked;
public:
- ConsumerImpl(Channel* parent, string& tag, Queue::shared_ptr queue, ConnectionToken* const connection, bool ack);
+ ConsumerImpl(Channel* parent, const string& tag, Queue::shared_ptr queue, ConnectionToken* const connection, bool ack);
virtual bool deliver(Message::shared_ptr& msg);
void cancel();
void requestDispatch();
@@ -90,7 +90,7 @@ namespace qpid {
Exchange::shared_ptr exchange;//exchange to which any in-progress message was published to
virtual void complete(Message::shared_ptr& msg);
- void deliver(Message::shared_ptr& msg, string& tag, Queue::shared_ptr& queue, bool ackExpected);
+ void deliver(Message::shared_ptr& msg, const string& tag, Queue::shared_ptr& queue, bool ackExpected);
void cancel(consumer_iterator consumer);
bool checkPrefetch(Message::shared_ptr& msg);
diff --git a/cpp/src/qpid/broker/DirectExchange.cpp b/cpp/src/qpid/broker/DirectExchange.cpp
index fe6117ce18..2713fb9482 100644
--- a/cpp/src/qpid/broker/DirectExchange.cpp
+++ b/cpp/src/qpid/broker/DirectExchange.cpp
@@ -30,7 +30,7 @@ DirectExchange::DirectExchange(const string& _name) : Exchange(_name) {
}
-void DirectExchange::bind(Queue::shared_ptr queue, const string& routingKey, FieldTable* args){
+void DirectExchange::bind(Queue::shared_ptr queue, const string& routingKey, const FieldTable* args){
Mutex::ScopedLock l(lock);
std::vector<Queue::shared_ptr>& queues(bindings[routingKey]);
std::vector<Queue::shared_ptr>::iterator i = find(queues.begin(), queues.end(), queue);
@@ -40,7 +40,7 @@ void DirectExchange::bind(Queue::shared_ptr queue, const string& routingKey, Fie
}
}
-void DirectExchange::unbind(Queue::shared_ptr queue, const string& routingKey, FieldTable* /*args*/){
+void DirectExchange::unbind(Queue::shared_ptr queue, const string& routingKey, const FieldTable* /*args*/){
Mutex::ScopedLock l(lock);
std::vector<Queue::shared_ptr>& queues(bindings[routingKey]);
@@ -53,7 +53,7 @@ void DirectExchange::unbind(Queue::shared_ptr queue, const string& routingKey, F
}
}
-void DirectExchange::route(Deliverable& msg, const string& routingKey, FieldTable* /*args*/){
+void DirectExchange::route(Deliverable& msg, const string& routingKey, const FieldTable* /*args*/){
Mutex::ScopedLock l(lock);
std::vector<Queue::shared_ptr>& queues(bindings[routingKey]);
int count(0);
diff --git a/cpp/src/qpid/broker/DirectExchange.h b/cpp/src/qpid/broker/DirectExchange.h
index 0904facaef..09909a8383 100644
--- a/cpp/src/qpid/broker/DirectExchange.h
+++ b/cpp/src/qpid/broker/DirectExchange.h
@@ -42,11 +42,11 @@ namespace broker {
virtual std::string getType(){ return typeName; }
- virtual void bind(Queue::shared_ptr queue, const std::string& routingKey, qpid::framing::FieldTable* args);
+ virtual void bind(Queue::shared_ptr queue, const std::string& routingKey, const qpid::framing::FieldTable* args);
- virtual void unbind(Queue::shared_ptr queue, const std::string& routingKey, qpid::framing::FieldTable* args);
+ virtual void unbind(Queue::shared_ptr queue, const std::string& routingKey, const qpid::framing::FieldTable* args);
- virtual void route(Deliverable& msg, const std::string& routingKey, qpid::framing::FieldTable* args);
+ virtual void route(Deliverable& msg, const std::string& routingKey, const qpid::framing::FieldTable* args);
virtual ~DirectExchange();
};
diff --git a/cpp/src/qpid/broker/Exchange.h b/cpp/src/qpid/broker/Exchange.h
index 2cca942290..d66c4a9e0d 100644
--- a/cpp/src/qpid/broker/Exchange.h
+++ b/cpp/src/qpid/broker/Exchange.h
@@ -39,9 +39,9 @@ namespace qpid {
virtual ~Exchange(){}
string getName() { return name; }
virtual string getType() = 0;
- virtual void bind(Queue::shared_ptr queue, const string& routingKey, qpid::framing::FieldTable* args) = 0;
- virtual void unbind(Queue::shared_ptr queue, const string& routingKey, qpid::framing::FieldTable* args) = 0;
- virtual void route(Deliverable& msg, const string& routingKey, qpid::framing::FieldTable* args) = 0;
+ virtual void bind(Queue::shared_ptr queue, const string& routingKey, const qpid::framing::FieldTable* args) = 0;
+ virtual void unbind(Queue::shared_ptr queue, const string& routingKey, const qpid::framing::FieldTable* args) = 0;
+ virtual void route(Deliverable& msg, const string& routingKey, const qpid::framing::FieldTable* args) = 0;
};
}
}
diff --git a/cpp/src/qpid/broker/ExchangeBinding.cpp b/cpp/src/qpid/broker/ExchangeBinding.cpp
index 38c60cad0f..375fbe165e 100644
--- a/cpp/src/qpid/broker/ExchangeBinding.cpp
+++ b/cpp/src/qpid/broker/ExchangeBinding.cpp
@@ -24,7 +24,7 @@
using namespace qpid::broker;
using namespace qpid::framing;
-ExchangeBinding::ExchangeBinding(Exchange* _e, Queue::shared_ptr _q, const string& _key, FieldTable* _args) : e(_e), q(_q), key(_key), args(_args){}
+ExchangeBinding::ExchangeBinding(Exchange* _e, Queue::shared_ptr _q, const string& _key, const FieldTable* _args) : e(_e), q(_q), key(_key), args(_args){}
void ExchangeBinding::cancel(){
e->unbind(q, key, args);
diff --git a/cpp/src/qpid/broker/ExchangeBinding.h b/cpp/src/qpid/broker/ExchangeBinding.h
index 1133d14e51..fd1a05391a 100644
--- a/cpp/src/qpid/broker/ExchangeBinding.h
+++ b/cpp/src/qpid/broker/ExchangeBinding.h
@@ -34,9 +34,9 @@ namespace qpid {
Exchange* e;
Queue::shared_ptr q;
const string key;
- qpid::framing::FieldTable* args;
+ const qpid::framing::FieldTable* args;
public:
- ExchangeBinding(Exchange* _e, Queue::shared_ptr _q, const string& _key, qpid::framing::FieldTable* _args);
+ ExchangeBinding(Exchange* _e, Queue::shared_ptr _q, const string& _key, const qpid::framing::FieldTable* _args);
virtual void cancel();
virtual ~ExchangeBinding();
};
diff --git a/cpp/src/qpid/broker/FanOutExchange.cpp b/cpp/src/qpid/broker/FanOutExchange.cpp
index de0310495d..0f3223d3a6 100644
--- a/cpp/src/qpid/broker/FanOutExchange.cpp
+++ b/cpp/src/qpid/broker/FanOutExchange.cpp
@@ -28,7 +28,7 @@ using namespace qpid::sys;
FanOutExchange::FanOutExchange(const std::string& _name) : Exchange(_name) {}
-void FanOutExchange::bind(Queue::shared_ptr queue, const string& routingKey, FieldTable* args){
+void FanOutExchange::bind(Queue::shared_ptr queue, const string& routingKey, const FieldTable* args){
Mutex::ScopedLock locker(lock);
// Add if not already present.
Queue::vector::iterator i = std::find(bindings.begin(), bindings.end(), queue);
@@ -38,7 +38,7 @@ void FanOutExchange::bind(Queue::shared_ptr queue, const string& routingKey, Fie
}
}
-void FanOutExchange::unbind(Queue::shared_ptr queue, const string& /*routingKey*/, FieldTable* /*args*/){
+void FanOutExchange::unbind(Queue::shared_ptr queue, const string& /*routingKey*/, const FieldTable* /*args*/){
Mutex::ScopedLock locker(lock);
Queue::vector::iterator i = std::find(bindings.begin(), bindings.end(), queue);
if (i != bindings.end()) {
@@ -47,7 +47,7 @@ void FanOutExchange::unbind(Queue::shared_ptr queue, const string& /*routingKey*
}
}
-void FanOutExchange::route(Deliverable& msg, const string& /*routingKey*/, FieldTable* /*args*/){
+void FanOutExchange::route(Deliverable& msg, const string& /*routingKey*/, const FieldTable* /*args*/){
Mutex::ScopedLock locker(lock);
for(Queue::vector::iterator i = bindings.begin(); i != bindings.end(); ++i){
msg.deliverTo(*i);
diff --git a/cpp/src/qpid/broker/FanOutExchange.h b/cpp/src/qpid/broker/FanOutExchange.h
index 671f93819d..0e309a0e79 100644
--- a/cpp/src/qpid/broker/FanOutExchange.h
+++ b/cpp/src/qpid/broker/FanOutExchange.h
@@ -43,11 +43,11 @@ class FanOutExchange : public virtual Exchange {
virtual std::string getType(){ return typeName; }
- virtual void bind(Queue::shared_ptr queue, const std::string& routingKey, qpid::framing::FieldTable* args);
+ virtual void bind(Queue::shared_ptr queue, const std::string& routingKey, const qpid::framing::FieldTable* args);
- virtual void unbind(Queue::shared_ptr queue, const std::string& routingKey, qpid::framing::FieldTable* args);
+ virtual void unbind(Queue::shared_ptr queue, const std::string& routingKey, const qpid::framing::FieldTable* args);
- virtual void route(Deliverable& msg, const std::string& routingKey, qpid::framing::FieldTable* args);
+ virtual void route(Deliverable& msg, const std::string& routingKey, const qpid::framing::FieldTable* args);
virtual ~FanOutExchange();
};
diff --git a/cpp/src/qpid/broker/HeadersExchange.cpp b/cpp/src/qpid/broker/HeadersExchange.cpp
index 18676691e6..bf1a89c7e8 100644
--- a/cpp/src/qpid/broker/HeadersExchange.cpp
+++ b/cpp/src/qpid/broker/HeadersExchange.cpp
@@ -43,7 +43,7 @@ namespace {
HeadersExchange::HeadersExchange(const string& _name) : Exchange(_name) { }
-void HeadersExchange::bind(Queue::shared_ptr queue, const string& routingKey, FieldTable* args){
+void HeadersExchange::bind(Queue::shared_ptr queue, const string& routingKey, const FieldTable* args){
Mutex::ScopedLock locker(lock);
std::string what = args->getString("x-match");
if (what != all && what != any) {
@@ -53,7 +53,7 @@ void HeadersExchange::bind(Queue::shared_ptr queue, const string& routingKey, Fi
queue->bound(new ExchangeBinding(this, queue, routingKey, args));
}
-void HeadersExchange::unbind(Queue::shared_ptr queue, const string& /*routingKey*/, FieldTable* args){
+void HeadersExchange::unbind(Queue::shared_ptr queue, const string& /*routingKey*/, const FieldTable* args){
Mutex::ScopedLock locker(lock);
Bindings::iterator i =
std::find(bindings.begin(),bindings.end(), Binding(*args, queue));
@@ -61,7 +61,7 @@ void HeadersExchange::unbind(Queue::shared_ptr queue, const string& /*routingKey
}
-void HeadersExchange::route(Deliverable& msg, const string& /*routingKey*/, FieldTable* args){
+void HeadersExchange::route(Deliverable& msg, const string& /*routingKey*/, const FieldTable* args){
Mutex::ScopedLock locker(lock);;
for (Bindings::iterator i = bindings.begin(); i != bindings.end(); ++i) {
if (match(i->first, *args)) msg.deliverTo(i->second);
diff --git a/cpp/src/qpid/broker/HeadersExchange.h b/cpp/src/qpid/broker/HeadersExchange.h
index 8669329ba0..5d1a51747f 100644
--- a/cpp/src/qpid/broker/HeadersExchange.h
+++ b/cpp/src/qpid/broker/HeadersExchange.h
@@ -46,11 +46,11 @@ class HeadersExchange : public virtual Exchange {
virtual std::string getType(){ return typeName; }
- virtual void bind(Queue::shared_ptr queue, const string& routingKey, qpid::framing::FieldTable* args);
+ virtual void bind(Queue::shared_ptr queue, const string& routingKey, const qpid::framing::FieldTable* args);
- virtual void unbind(Queue::shared_ptr queue, const string& routingKey, qpid::framing::FieldTable* args);
+ virtual void unbind(Queue::shared_ptr queue, const string& routingKey, const qpid::framing::FieldTable* args);
- virtual void route(Deliverable& msg, const string& routingKey, qpid::framing::FieldTable* args);
+ virtual void route(Deliverable& msg, const string& routingKey, const qpid::framing::FieldTable* args);
virtual ~HeadersExchange();
diff --git a/cpp/src/qpid/broker/Message.cpp b/cpp/src/qpid/broker/Message.cpp
index d398fd7e31..f71324f3fa 100644
--- a/cpp/src/qpid/broker/Message.cpp
+++ b/cpp/src/qpid/broker/Message.cpp
@@ -20,6 +20,10 @@
*/
#include <qpid/broker/Message.h>
#include <iostream>
+// AMQP version change - kpvdr 2006-11-17
+#include <qpid/framing/ProtocolVersion.h>
+#include <qpid/framing/BasicDeliverBody.h>
+#include <qpid/framing/BasicGetOkBody.h>
using namespace boost;
using namespace qpid::broker;
@@ -76,7 +80,10 @@ void Message::redeliver(){
void Message::deliver(OutputHandler* out, int channel,
const string& consumerTag, u_int64_t deliveryTag,
u_int32_t framesize){
- out->send(new AMQFrame(channel, new BasicDeliverBody(consumerTag, deliveryTag, redelivered, exchange, routingKey)));
+ // AMQP version change - kpvdr 2006-11-17
+ // TODO: Make this class version-aware and link these hard-wired numbers to that version
+ out->send(new AMQFrame(channel, new BasicDeliverBody(ProtocolVersion(8,0), consumerTag, deliveryTag, redelivered, exchange, routingKey)));
+// out->send(new AMQFrame(channel, new BasicDeliverBody(ProtocolVersion(8,0), consumerTag, deliveryTag, redelivered, exchange, routingKey)));
sendContent(out, channel, framesize);
}
@@ -86,7 +93,9 @@ void Message::sendGetOk(OutputHandler* out,
u_int64_t deliveryTag,
u_int32_t framesize){
- out->send(new AMQFrame(channel, new BasicGetOkBody(deliveryTag, redelivered, exchange, routingKey, messageCount)));
+ // AMQP version change - kpvdr 2006-11-17
+ // TODO: Make this class version-aware and link these hard-wired numbers to that version
+ out->send(new AMQFrame(channel, new BasicGetOkBody(ProtocolVersion(8,0), deliveryTag, redelivered, exchange, routingKey, messageCount)));
sendContent(out, channel, framesize);
}
diff --git a/cpp/src/qpid/broker/Message.h b/cpp/src/qpid/broker/Message.h
index f6b9e19209..3d0a0d358b 100644
--- a/cpp/src/qpid/broker/Message.h
+++ b/cpp/src/qpid/broker/Message.h
@@ -27,7 +27,6 @@
#include <qpid/framing/AMQContentBody.h>
#include <qpid/framing/AMQHeaderBody.h>
#include <qpid/framing/BasicHeaderProperties.h>
-#include <qpid/framing/BasicPublishBody.h>
#include <qpid/framing/OutputHandler.h>
namespace qpid {
diff --git a/cpp/src/qpid/broker/SessionHandlerImpl.cpp b/cpp/src/qpid/broker/SessionHandlerImpl.cpp
index bbb5d22c8d..c8c7b440aa 100644
--- a/cpp/src/qpid/broker/SessionHandlerImpl.cpp
+++ b/cpp/src/qpid/broker/SessionHandlerImpl.cpp
@@ -37,7 +37,9 @@ SessionHandlerImpl::SessionHandlerImpl(SessionContext* _context,
AutoDelete* _cleaner,
const u_int32_t _timeout) :
context(_context),
- client(context),
+// AMQP version management change - kpvdr 2006-11-17
+// TODO: Make this class version-aware and link these hard-wired numbers to that version
+ client(context, 8, 0),
queues(_queues),
exchanges(_exchanges),
cleaner(_cleaner),
@@ -165,26 +167,26 @@ void SessionHandlerImpl::handleHeartbeat(AMQHeartbeatBody::shared_ptr /*body*/){
}
void SessionHandlerImpl::ConnectionHandlerImpl::startOk(
- u_int16_t /*channel*/, FieldTable& /*clientProperties*/, string& /*mechanism*/,
- string& /*response*/, string& /*locale*/){
+ u_int16_t /*channel*/, const FieldTable& /*clientProperties*/, const string& /*mechanism*/,
+ const string& /*response*/, const string& /*locale*/){
parent->client.getConnection().tune(0, 100, parent->framemax, parent->heartbeat);
}
-void SessionHandlerImpl::ConnectionHandlerImpl::secureOk(u_int16_t /*channel*/, string& /*response*/){}
+void SessionHandlerImpl::ConnectionHandlerImpl::secureOk(u_int16_t /*channel*/, const string& /*response*/){}
void SessionHandlerImpl::ConnectionHandlerImpl::tuneOk(u_int16_t /*channel*/, u_int16_t /*channelmax*/, u_int32_t framemax, u_int16_t heartbeat){
parent->framemax = framemax;
parent->heartbeat = heartbeat;
}
-void SessionHandlerImpl::ConnectionHandlerImpl::open(u_int16_t /*channel*/, string& /*virtualHost*/, string& /*capabilities*/, bool /*insist*/){
+void SessionHandlerImpl::ConnectionHandlerImpl::open(u_int16_t /*channel*/, const string& /*virtualHost*/, const string& /*capabilities*/, bool /*insist*/){
string knownhosts;
parent->client.getConnection().openOk(0, knownhosts);
}
void SessionHandlerImpl::ConnectionHandlerImpl::close(
- u_int16_t /*channel*/, u_int16_t /*replyCode*/, string& /*replyText*/,
+ u_int16_t /*channel*/, u_int16_t /*replyCode*/, const string& /*replyText*/,
u_int16_t /*classId*/, u_int16_t /*methodId*/)
{
parent->client.getConnection().closeOk(0);
@@ -197,7 +199,7 @@ void SessionHandlerImpl::ConnectionHandlerImpl::closeOk(u_int16_t /*channel*/){
-void SessionHandlerImpl::ChannelHandlerImpl::open(u_int16_t channel, string& /*outOfBand*/){
+void SessionHandlerImpl::ChannelHandlerImpl::open(u_int16_t channel, const string& /*outOfBand*/){
parent->channels[channel] = new Channel(parent->context, channel, parent->framemax);
parent->client.getChannel().openOk(channel);
}
@@ -205,7 +207,7 @@ void SessionHandlerImpl::ChannelHandlerImpl::open(u_int16_t channel, string& /*o
void SessionHandlerImpl::ChannelHandlerImpl::flow(u_int16_t /*channel*/, bool /*active*/){}
void SessionHandlerImpl::ChannelHandlerImpl::flowOk(u_int16_t /*channel*/, bool /*active*/){}
-void SessionHandlerImpl::ChannelHandlerImpl::close(u_int16_t channel, u_int16_t /*replyCode*/, string& /*replyText*/,
+void SessionHandlerImpl::ChannelHandlerImpl::close(u_int16_t channel, u_int16_t /*replyCode*/, const string& /*replyText*/,
u_int16_t /*classId*/, u_int16_t /*methodId*/){
Channel* c = parent->getChannel(channel);
if(c){
@@ -220,9 +222,9 @@ void SessionHandlerImpl::ChannelHandlerImpl::closeOk(u_int16_t /*channel*/){}
-void SessionHandlerImpl::ExchangeHandlerImpl::declare(u_int16_t channel, u_int16_t /*ticket*/, string& exchange, string& type,
+void SessionHandlerImpl::ExchangeHandlerImpl::declare(u_int16_t channel, u_int16_t /*ticket*/, const string& exchange, const string& type,
bool passive, bool /*durable*/, bool /*autoDelete*/, bool /*internal*/, bool nowait,
- FieldTable& /*arguments*/){
+ const FieldTable& /*arguments*/){
if(passive){
if(!parent->exchanges->get(exchange)){
@@ -244,17 +246,17 @@ void SessionHandlerImpl::ExchangeHandlerImpl::declare(u_int16_t channel, u_int16
parent->client.getExchange().declareOk(channel);
}
}
-
+
void SessionHandlerImpl::ExchangeHandlerImpl::delete_(u_int16_t channel, u_int16_t /*ticket*/,
- string& exchange, bool /*ifUnused*/, bool nowait){
+ const string& exchange, bool /*ifUnused*/, bool nowait){
//TODO: implement unused
parent->exchanges->destroy(exchange);
if(!nowait) parent->client.getExchange().deleteOk(channel);
}
-void SessionHandlerImpl::QueueHandlerImpl::declare(u_int16_t channel, u_int16_t /*ticket*/, string& name,
+void SessionHandlerImpl::QueueHandlerImpl::declare(u_int16_t channel, u_int16_t /*ticket*/, const string& name,
bool passive, bool durable, bool exclusive,
- bool autoDelete, bool nowait, FieldTable& /*arguments*/){
+ bool autoDelete, bool nowait, const qpid::framing::FieldTable& /*arguments*/){
Queue::shared_ptr queue;
if (passive && !name.empty()) {
queue = parent->getQueue(name, channel);
@@ -282,34 +284,37 @@ void SessionHandlerImpl::QueueHandlerImpl::declare(u_int16_t channel, u_int16_t
throw ChannelException(405, "Cannot grant exclusive access to queue");
}
if (!nowait) {
- name = queue->getName();
- parent->client.getQueue().declareOk(channel, name, queue->getMessageCount(), queue->getConsumerCount());
+ string queueName = queue->getName();
+ parent->client.getQueue().declareOk(channel, queueName, queue->getMessageCount(), queue->getConsumerCount());
}
}
-void SessionHandlerImpl::QueueHandlerImpl::bind(u_int16_t channel, u_int16_t /*ticket*/, string& queueName,
- string& exchangeName, string& routingKey, bool nowait,
- FieldTable& arguments){
+void SessionHandlerImpl::QueueHandlerImpl::bind(u_int16_t channel, u_int16_t /*ticket*/, const string& queueName,
+ const string& exchangeName, const string& routingKey, bool nowait,
+ const FieldTable& arguments){
Queue::shared_ptr queue = parent->getQueue(queueName, channel);
Exchange::shared_ptr exchange = parent->exchanges->get(exchangeName);
if(exchange){
- if(routingKey.empty() && queueName.empty()) routingKey = queue->getName();
- exchange->bind(queue, routingKey, &arguments);
+// kpvdr - cannot use this any longer as routingKey is now const
+// if(routingKey.empty() && queueName.empty()) routingKey = queue->getName();
+// exchange->bind(queue, routingKey, &arguments);
+ string exchangeRoutingKey = routingKey.empty() && queueName.empty() ? queue->getName() : routingKey;
+ exchange->bind(queue, exchangeRoutingKey, &arguments);
if(!nowait) parent->client.getQueue().bindOk(channel);
}else{
throw ChannelException(404, "Bind failed. No such exchange: " + exchangeName);
}
}
-void SessionHandlerImpl::QueueHandlerImpl::purge(u_int16_t channel, u_int16_t /*ticket*/, string& queueName, bool nowait){
+void SessionHandlerImpl::QueueHandlerImpl::purge(u_int16_t channel, u_int16_t /*ticket*/, const string& queueName, bool nowait){
Queue::shared_ptr queue = parent->getQueue(queueName, channel);
int count = queue->purge();
if(!nowait) parent->client.getQueue().purgeOk(channel, count);
}
-void SessionHandlerImpl::QueueHandlerImpl::delete_(u_int16_t channel, u_int16_t /*ticket*/, string& queue,
+void SessionHandlerImpl::QueueHandlerImpl::delete_(u_int16_t channel, u_int16_t /*ticket*/, const string& queue,
bool ifUnused, bool ifEmpty, bool nowait){
ChannelException error(0, "");
int count(0);
@@ -342,7 +347,7 @@ void SessionHandlerImpl::BasicHandlerImpl::qos(u_int16_t channel, u_int32_t pref
}
void SessionHandlerImpl::BasicHandlerImpl::consume(u_int16_t channelId, u_int16_t /*ticket*/,
- string& queueName, string& consumerTag,
+ const string& queueName, const string& consumerTag,
bool noLocal, bool noAck, bool exclusive,
bool nowait){
@@ -353,8 +358,9 @@ void SessionHandlerImpl::BasicHandlerImpl::consume(u_int16_t channelId, u_int16_
}
try{
- channel->consume(consumerTag, queue, !noAck, exclusive, noLocal ? parent : 0);
- if(!nowait) parent->client.getBasic().consumeOk(channelId, consumerTag);
+ string newTag = consumerTag;
+ channel->consume(newTag, queue, !noAck, exclusive, noLocal ? parent : 0);
+ if(!nowait) parent->client.getBasic().consumeOk(channelId, newTag);
//allow messages to be dispatched if required as there is now a consumer:
queue->dispatch();
@@ -365,13 +371,13 @@ void SessionHandlerImpl::BasicHandlerImpl::consume(u_int16_t channelId, u_int16_
}
-void SessionHandlerImpl::BasicHandlerImpl::cancel(u_int16_t channel, string& consumerTag, bool nowait){
+void SessionHandlerImpl::BasicHandlerImpl::cancel(u_int16_t channel, const string& consumerTag, bool nowait){
parent->getChannel(channel)->cancel(consumerTag);
if(!nowait) parent->client.getBasic().cancelOk(channel, consumerTag);
}
void SessionHandlerImpl::BasicHandlerImpl::publish(u_int16_t channel, u_int16_t /*ticket*/,
- string& exchangeName, string& routingKey,
+ const string& exchangeName, const string& routingKey,
bool mandatory, bool immediate){
Exchange::shared_ptr exchange = exchangeName.empty() ? parent->exchanges->getDefault() : parent->exchanges->get(exchangeName);
@@ -383,7 +389,7 @@ void SessionHandlerImpl::BasicHandlerImpl::publish(u_int16_t channel, u_int16_t
}
}
-void SessionHandlerImpl::BasicHandlerImpl::get(u_int16_t channelId, u_int16_t /*ticket*/, string& queueName, bool noAck){
+void SessionHandlerImpl::BasicHandlerImpl::get(u_int16_t channelId, u_int16_t /*ticket*/, const string& queueName, bool noAck){
Queue::shared_ptr queue = parent->getQueue(queueName, channelId);
if(!parent->getChannel(channelId)->get(queue, !noAck)){
string clusterId;//not used, part of an imatix hack
diff --git a/cpp/src/qpid/broker/SessionHandlerImpl.h b/cpp/src/qpid/broker/SessionHandlerImpl.h
index cec7d8ada4..2df4b10f9b 100644
--- a/cpp/src/qpid/broker/SessionHandlerImpl.h
+++ b/cpp/src/qpid/broker/SessionHandlerImpl.h
@@ -117,18 +117,22 @@ class SessionHandlerImpl : public virtual qpid::sys::SessionHandler,
public:
inline ConnectionHandlerImpl(SessionHandlerImpl* _parent) : parent(_parent) {}
- virtual void startOk(u_int16_t channel, qpid::framing::FieldTable& clientProperties, string& mechanism,
- string& response, string& locale);
+ // Change to match new code generator function signature (adding const to string& and FieldTable&) - kpvdr 2006-11-20
+ virtual void startOk(u_int16_t channel, const qpid::framing::FieldTable& clientProperties, const string& mechanism,
+ const string& response, const string& locale);
- virtual void secureOk(u_int16_t channel, string& response);
+ // Change to match new code generator function signature (adding const to string&) - kpvdr 2006-11-20
+ virtual void secureOk(u_int16_t channel, const 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);
+ // Change to match new code generator function signature (adding const to string&) - kpvdr 2006-11-20
+ virtual void open(u_int16_t channel, const string& virtualHost, const string& capabilities, bool insist);
- virtual void close(u_int16_t channel, u_int16_t replyCode, string& replyText, u_int16_t classId,
+ // Change to match new code generator function signature (adding const to string&) - kpvdr 2006-11-20
+ virtual void close(u_int16_t channel, u_int16_t replyCode, const string& replyText, u_int16_t classId,
u_int16_t methodId);
-
+
virtual void closeOk(u_int16_t channel);
virtual ~ConnectionHandlerImpl(){}
@@ -139,13 +143,15 @@ class SessionHandlerImpl : public virtual qpid::sys::SessionHandler,
public:
inline ChannelHandlerImpl(SessionHandlerImpl* _parent) : parent(_parent) {}
- virtual void open(u_int16_t channel, string& outOfBand);
+ // Change to match new code generator function signature (adding const to string&) - kpvdr 2006-11-20
+ virtual void open(u_int16_t channel, const 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,
+ // Change to match new code generator function signature (adding const to string&) - kpvdr 2006-11-20
+ virtual void close(u_int16_t channel, u_int16_t replyCode, const string& replyText,
u_int16_t classId, u_int16_t methodId);
virtual void closeOk(u_int16_t channel);
@@ -158,11 +164,13 @@ class SessionHandlerImpl : public virtual qpid::sys::SessionHandler,
public:
inline ExchangeHandlerImpl(SessionHandlerImpl* _parent) : parent(_parent) {}
- virtual void declare(u_int16_t channel, u_int16_t ticket, string& exchange, string& type,
+ // Change to match new code generator function signature (adding const to string& and FieldTable&) - kpvdr 2006-11-20
+ virtual void declare(u_int16_t channel, u_int16_t ticket, const string& exchange, const string& type,
bool passive, bool durable, bool autoDelete, bool internal, bool nowait,
- qpid::framing::FieldTable& arguments);
+ const qpid::framing::FieldTable& arguments);
- virtual void delete_(u_int16_t channel, u_int16_t ticket, string& exchange, bool ifUnused, bool nowait);
+ // Change to match new code generator function signature (adding const to string&) - kpvdr 2006-11-20
+ virtual void delete_(u_int16_t channel, u_int16_t ticket, const string& exchange, bool ifUnused, bool nowait);
virtual ~ExchangeHandlerImpl(){}
};
@@ -173,20 +181,24 @@ class SessionHandlerImpl : public virtual qpid::sys::SessionHandler,
public:
inline QueueHandlerImpl(SessionHandlerImpl* _parent) : parent(_parent) {}
- virtual void declare(u_int16_t channel, u_int16_t ticket, string& queue,
+ // Change to match new code generator function signature (adding const to string& and FieldTable&) - kpvdr 2006-11-20
+ virtual void declare(u_int16_t channel, u_int16_t ticket, const string& queue,
bool passive, bool durable, bool exclusive,
- bool autoDelete, bool nowait, qpid::framing::FieldTable& arguments);
+ bool autoDelete, bool nowait, const 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);
+ // Change to match new code generator function signature (adding const to string& and FieldTable&) - kpvdr 2006-11-20
+ virtual void bind(u_int16_t channel, u_int16_t ticket, const string& queue,
+ const string& exchange, const string& routingKey, bool nowait,
+ const qpid::framing::FieldTable& arguments);
- virtual void purge(u_int16_t channel, u_int16_t ticket, string& queue,
+ // Change to match new code generator function signature (adding const to string&) - kpvdr 2006-11-20
+ virtual void purge(u_int16_t channel, u_int16_t ticket, const string& queue,
bool nowait);
- virtual void delete_(u_int16_t channel, u_int16_t ticket, string& queue, bool ifUnused, bool ifEmpty,
+ // Change to match new code generator function signature (adding const to string&) - kpvdr 2006-11-20
+ virtual void delete_(u_int16_t channel, u_int16_t ticket, const string& queue, bool ifUnused, bool ifEmpty,
bool nowait);
-
+
virtual ~QueueHandlerImpl(){}
};
@@ -197,15 +209,19 @@ class SessionHandlerImpl : public virtual qpid::sys::SessionHandler,
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,
+ // Change to match new code generator function signature (adding const to string&) - kpvdr 2006-11-20
+ virtual void consume(u_int16_t channel, u_int16_t ticket, const string& queue, const string& consumerTag,
bool noLocal, bool noAck, bool exclusive, bool nowait);
+
+ // Change to match new code generator function signature (adding const to string&) - kpvdr 2006-11-20
+ virtual void cancel(u_int16_t channel, const string& consumerTag, 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,
+ // Change to match new code generator function signature (adding const to string&) - kpvdr 2006-11-20
+ virtual void publish(u_int16_t channel, u_int16_t ticket, const string& exchange, const string& routingKey,
bool mandatory, bool immediate);
- virtual void get(u_int16_t channel, u_int16_t ticket, string& queue, bool noAck);
+ // Change to match new code generator function signature (adding const to string&) - kpvdr 2006-11-20
+ virtual void get(u_int16_t channel, u_int16_t ticket, const string& queue, bool noAck);
virtual void ack(u_int16_t channel, u_int64_t deliveryTag, bool multiple);
@@ -238,7 +254,11 @@ class SessionHandlerImpl : public virtual qpid::sys::SessionHandler,
inline virtual FileHandler* getFileHandler(){ throw ConnectionException(540, "File class not implemented"); }
inline virtual StreamHandler* getStreamHandler(){ throw ConnectionException(540, "Stream class not implemented"); }
inline virtual DtxHandler* getDtxHandler(){ throw ConnectionException(540, "Dtx class not implemented"); }
- inline virtual TunnelHandler* getTunnelHandler(){ throw ConnectionException(540, "Tunnel class not implemented"); }
+ inline virtual TunnelHandler* getTunnelHandler(){ throw ConnectionException(540, "Tunnel class not implemented"); }
+
+ // Temporary add-in to resolve version conflicts: AMQP v8.0 still defines class Test;
+ // however v0.9 will not - kpvdr 2006-11-17
+ inline virtual TestHandler* getTestHandler(){ throw ConnectionException(540, "Test class not implemented"); }
};
}
diff --git a/cpp/src/qpid/broker/TopicExchange.cpp b/cpp/src/qpid/broker/TopicExchange.cpp
index 2ff4a42070..34fb25781e 100644
--- a/cpp/src/qpid/broker/TopicExchange.cpp
+++ b/cpp/src/qpid/broker/TopicExchange.cpp
@@ -117,14 +117,14 @@ bool TopicPattern::match(const Tokens& target) const
TopicExchange::TopicExchange(const string& _name) : Exchange(_name) { }
-void TopicExchange::bind(Queue::shared_ptr queue, const string& routingKey, FieldTable* args){
+void TopicExchange::bind(Queue::shared_ptr queue, const string& routingKey, const FieldTable* args){
Monitor::ScopedLock l(lock);
TopicPattern routingPattern(routingKey);
bindings[routingPattern].push_back(queue);
queue->bound(new ExchangeBinding(this, queue, routingKey, args));
}
-void TopicExchange::unbind(Queue::shared_ptr queue, const string& routingKey, FieldTable* /*args*/){
+void TopicExchange::unbind(Queue::shared_ptr queue, const string& routingKey, const FieldTable* /*args*/){
Monitor::ScopedLock l(lock);
BindingMap::iterator bi = bindings.find(TopicPattern(routingKey));
Queue::vector& qv(bi->second);
@@ -136,7 +136,7 @@ void TopicExchange::unbind(Queue::shared_ptr queue, const string& routingKey, Fi
}
-void TopicExchange::route(Deliverable& msg, const string& routingKey, FieldTable* /*args*/){
+void TopicExchange::route(Deliverable& msg, const string& routingKey, const FieldTable* /*args*/){
Monitor::ScopedLock l(lock);
for (BindingMap::iterator i = bindings.begin(); i != bindings.end(); ++i) {
if (i->first.match(routingKey)) {
diff --git a/cpp/src/qpid/broker/TopicExchange.h b/cpp/src/qpid/broker/TopicExchange.h
index 96f0283fdd..05fe871114 100644
--- a/cpp/src/qpid/broker/TopicExchange.h
+++ b/cpp/src/qpid/broker/TopicExchange.h
@@ -83,11 +83,11 @@ class TopicExchange : public virtual Exchange{
virtual std::string getType(){ return typeName; }
- virtual void bind(Queue::shared_ptr queue, const string& routingKey, qpid::framing::FieldTable* args);
+ virtual void bind(Queue::shared_ptr queue, const string& routingKey, const qpid::framing::FieldTable* args);
- virtual void unbind(Queue::shared_ptr queue, const string& routingKey, qpid::framing::FieldTable* args);
+ virtual void unbind(Queue::shared_ptr queue, const string& routingKey, const qpid::framing::FieldTable* args);
- virtual void route(Deliverable& msg, const string& routingKey, qpid::framing::FieldTable* args);
+ virtual void route(Deliverable& msg, const string& routingKey, const qpid::framing::FieldTable* args);
virtual ~TopicExchange();
};