You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by cc...@apache.org on 2006/12/20 23:29:39 UTC
svn commit: r489212 - in /incubator/qpid/trunk/qpid/cpp: lib/broker/
lib/client/ lib/common/framing/ tests/
Author: cctrieloff
Date: Wed Dec 20 14:29:38 2006
New Revision: 489212
URL: http://svn.apache.org/viewvc?view=rev&rev=489212
Log:
Support for multi version, merge part 1. - can still refactor out dup use of
version object in client and server opperations.
Modified:
incubator/qpid/trunk/qpid/cpp/lib/broker/BrokerChannel.cpp
incubator/qpid/trunk/qpid/cpp/lib/broker/BrokerChannel.h
incubator/qpid/trunk/qpid/cpp/lib/broker/BrokerMessage.cpp
incubator/qpid/trunk/qpid/cpp/lib/broker/BrokerMessage.h
incubator/qpid/trunk/qpid/cpp/lib/broker/SessionHandlerImpl.cpp
incubator/qpid/trunk/qpid/cpp/lib/client/ClientChannel.h
incubator/qpid/trunk/qpid/cpp/lib/client/Connection.cpp
incubator/qpid/trunk/qpid/cpp/lib/client/Connection.h
incubator/qpid/trunk/qpid/cpp/lib/common/framing/AMQFrame.h
incubator/qpid/trunk/qpid/cpp/tests/ChannelTest.cpp
incubator/qpid/trunk/qpid/cpp/tests/FramingTest.cpp
incubator/qpid/trunk/qpid/cpp/tests/MessageTest.cpp
incubator/qpid/trunk/qpid/cpp/tests/client_test.cpp
Modified: incubator/qpid/trunk/qpid/cpp/lib/broker/BrokerChannel.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/lib/broker/BrokerChannel.cpp?view=diff&rev=489212&r1=489211&r2=489212
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/lib/broker/BrokerChannel.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/lib/broker/BrokerChannel.cpp Wed Dec 20 14:29:38 2006
@@ -31,7 +31,7 @@
using namespace qpid::sys;
-Channel::Channel(OutputHandler* _out, int _id, u_int32_t _framesize, MessageStore* const _store, u_int64_t _stagingThreshold) :
+Channel::Channel(qpid::framing::ProtocolVersion& _version, OutputHandler* _out, int _id, u_int32_t _framesize, MessageStore* const _store, u_int64_t _stagingThreshold) :
id(_id),
out(_out),
currentDeliveryTag(1),
@@ -41,7 +41,8 @@
framesize(_framesize),
tagGenerator("sgen"),
store(_store),
- messageBuilder(this, _store, _stagingThreshold){
+ messageBuilder(this, _store, _stagingThreshold),
+ version(_version){
outstanding.reset();
}
@@ -118,7 +119,7 @@
outstanding.count++;
}
//send deliver method, header and content(s)
- msg->deliver(out, id, consumerTag, deliveryTag, framesize);
+ msg->deliver(out, id, consumerTag, deliveryTag, framesize, &version);
}
bool Channel::checkPrefetch(Message::shared_ptr& msg){
@@ -242,7 +243,7 @@
if(msg){
Mutex::ScopedLock locker(deliveryLock);
u_int64_t myDeliveryTag = currentDeliveryTag++;
- msg->sendGetOk(out, id, queue->getMessageCount() + 1, myDeliveryTag, framesize);
+ msg->sendGetOk(out, id, queue->getMessageCount() + 1, myDeliveryTag, framesize, &version);
if(ackExpected){
unacked.push_back(DeliveryRecord(msg, queue, myDeliveryTag));
}
@@ -253,5 +254,5 @@
}
void Channel::deliver(Message::shared_ptr& msg, const string& consumerTag, u_int64_t deliveryTag){
- msg->deliver(out, id, consumerTag, deliveryTag, framesize);
+ msg->deliver(out, id, consumerTag, deliveryTag, framesize, &version);
}
Modified: incubator/qpid/trunk/qpid/cpp/lib/broker/BrokerChannel.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/lib/broker/BrokerChannel.h?view=diff&rev=489212&r1=489211&r2=489212
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/lib/broker/BrokerChannel.h (original)
+++ incubator/qpid/trunk/qpid/cpp/lib/broker/BrokerChannel.h Wed Dec 20 14:29:38 2006
@@ -88,6 +88,7 @@
MessageStore* const store;
MessageBuilder messageBuilder;//builder for in-progress message
Exchange::shared_ptr exchange;//exchange to which any in-progress message was published to
+ qpid::framing::ProtocolVersion version; // version used for this channel
virtual void complete(Message::shared_ptr& msg);
void deliver(Message::shared_ptr& msg, const string& tag, Queue::shared_ptr& queue, bool ackExpected);
@@ -95,7 +96,7 @@
bool checkPrefetch(Message::shared_ptr& msg);
public:
- Channel(qpid::framing::OutputHandler* out, int id, u_int32_t framesize,
+ Channel(qpid::framing::ProtocolVersion& _version, qpid::framing::OutputHandler* out, int id, u_int32_t framesize,
MessageStore* const _store = 0, u_int64_t stagingThreshold = 0);
~Channel();
inline void setDefaultQueue(Queue::shared_ptr queue){ defaultQueue = queue; }
Modified: incubator/qpid/trunk/qpid/cpp/lib/broker/BrokerMessage.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/lib/broker/BrokerMessage.cpp?view=diff&rev=489212&r1=489211&r2=489212
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/lib/broker/BrokerMessage.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/lib/broker/BrokerMessage.cpp Wed Dec 20 14:29:38 2006
@@ -24,8 +24,6 @@
#include <InMemoryContent.h>
#include <LazyLoadedContent.h>
#include <MessageStore.h>
-// AMQP version change - kpvdr 2006-11-17
-#include <ProtocolVersion.h>
#include <BasicDeliverBody.h>
#include <BasicGetOkBody.h>
@@ -79,11 +77,10 @@
void Message::deliver(OutputHandler* out, int channel,
const string& consumerTag, u_int64_t deliveryTag,
- u_int32_t framesize){
-
- // 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)));
+ u_int32_t framesize,
+ ProtocolVersion* version){
+ // CCT -- TODO - Update code generator to take pointer/ not instance to avoid extra contruction
+ out->send(new AMQFrame(channel, new BasicDeliverBody(*version, consumerTag, deliveryTag, redelivered, exchange, routingKey)));
sendContent(out, channel, framesize);
}
@@ -91,11 +88,10 @@
int channel,
u_int32_t messageCount,
u_int64_t deliveryTag,
- u_int32_t framesize){
-
- // 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)));
+ u_int32_t framesize,
+ ProtocolVersion* version){
+ // CCT -- TODO - Update code generator to take pointer/ not instance to avoid extra contruction
+ out->send(new AMQFrame(channel, new BasicGetOkBody(*version, deliveryTag, redelivered, exchange, routingKey, messageCount)));
sendContent(out, channel, framesize);
}
Modified: incubator/qpid/trunk/qpid/cpp/lib/broker/BrokerMessage.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/lib/broker/BrokerMessage.h?view=diff&rev=489212&r1=489211&r2=489212
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/lib/broker/BrokerMessage.h (original)
+++ incubator/qpid/trunk/qpid/cpp/lib/broker/BrokerMessage.h Wed Dec 20 14:29:38 2006
@@ -25,6 +25,7 @@
#include <boost/shared_ptr.hpp>
#include <AMQContentBody.h>
#include <AMQHeaderBody.h>
+#include <ProtocolVersion.h>
#include <BasicHeaderProperties.h>
#include <ConnectionToken.h>
#include <Content.h>
@@ -37,6 +38,7 @@
class MessageStore;
using qpid::framing::string;
+
/**
* Represents an AMQP message, i.e. a header body, a list of
* content bodies and some details about the publication
@@ -76,12 +78,14 @@
int channel,
const string& consumerTag,
u_int64_t deliveryTag,
- u_int32_t framesize);
+ u_int32_t framesize,
+ qpid::framing::ProtocolVersion* version);
void sendGetOk(qpid::framing::OutputHandler* out,
int channel,
u_int32_t messageCount,
u_int64_t deliveryTag,
- u_int32_t framesize);
+ u_int32_t framesize,
+ qpid::framing::ProtocolVersion* version);
void redeliver();
qpid::framing::BasicHeaderProperties* getHeaderProperties();
Modified: incubator/qpid/trunk/qpid/cpp/lib/broker/SessionHandlerImpl.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/lib/broker/SessionHandlerImpl.cpp?view=diff&rev=489212&r1=489211&r2=489212
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/lib/broker/SessionHandlerImpl.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/lib/broker/SessionHandlerImpl.cpp Wed Dec 20 14:29:38 2006
@@ -131,7 +131,10 @@
if (client == NULL)
{
client = new qpid::framing::AMQP_ClientProxy(context, header->getMajor(), header->getMinor());
-
+
+
+ std::cout << "---------------" << this << std::endl;
+
//send connection start
FieldTable properties;
string mechanisms("PLAIN");
@@ -212,7 +215,9 @@
void SessionHandlerImpl::ChannelHandlerImpl::open(u_int16_t channel, const string& /*outOfBand*/){
- parent->channels[channel] = new Channel(parent->context, channel, parent->framemax,
+
+
+ parent->channels[channel] = new Channel(parent->client->getProtocolVersion() , parent->context, channel, parent->framemax,
parent->queues->getStore(), parent->settings.stagingThreshold);
parent->client->getChannel().openOk(channel);
}
Modified: incubator/qpid/trunk/qpid/cpp/lib/client/ClientChannel.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/lib/client/ClientChannel.h?view=diff&rev=489212&r1=489211&r2=489212
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/lib/client/ClientChannel.h (original)
+++ incubator/qpid/trunk/qpid/cpp/lib/client/ClientChannel.h Wed Dec 20 14:29:38 2006
@@ -92,7 +92,7 @@
u_int16_t prefetch;
const bool transactional;
- qpid::framing::ProtocolVersion version;
+ qpid::framing::ProtocolVersion version;
void enqueue();
void retrieve(Message& msg);
Modified: incubator/qpid/trunk/qpid/cpp/lib/client/Connection.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/lib/client/Connection.cpp?view=diff&rev=489212&r1=489211&r2=489212
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/lib/client/Connection.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/lib/client/Connection.cpp Wed Dec 20 14:29:38 2006
@@ -32,10 +32,8 @@
u_int16_t Connection::channelIdCounter;
-Connection::Connection(bool debug, u_int32_t _max_frame_size) : max_frame_size(_max_frame_size), closed(true),
-// AMQP version management change - kpvdr 2006-11-20
-// TODO: Make this class version-aware and link these hard-wired numbers to that version
- version(8, 0)
+Connection::Connection( bool debug, u_int32_t _max_frame_size, qpid::framing::ProtocolVersion* _version) : max_frame_size(_max_frame_size), closed(true),
+ version(_version->getMajor(),_version->getMinor())
{
connector = new Connector(debug, _max_frame_size);
}
@@ -53,7 +51,7 @@
out = connector->getOutputHandler();
connector->connect(host, port);
- ProtocolInitiation* header = new ProtocolInitiation(8, 0);
+ ProtocolInitiation* header = new ProtocolInitiation(version);
responses.expect();
connector->init(header);
responses.receive(method_bodies.connection_start);
Modified: incubator/qpid/trunk/qpid/cpp/lib/client/Connection.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/lib/client/Connection.h?view=diff&rev=489212&r1=489211&r2=489212
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/lib/client/Connection.h (original)
+++ incubator/qpid/trunk/qpid/cpp/lib/client/Connection.h Wed Dec 20 14:29:38 2006
@@ -38,6 +38,7 @@
#include <ResponseHandler.h>
namespace qpid {
+
/**
* The client namespace contains all classes that make up a client
* implementation of the AMQP protocol. The key classes that form
@@ -93,6 +94,8 @@
* Creates a connection object, but does not open the
* connection.
*
+ * @param _version the version of the protocol to connect with
+ *
* @param debug turns on tracing for the connection
* (i.e. prints details of the frames sent and received to std
* out). Optional and defaults to false.
@@ -100,7 +103,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);
+ Connection( bool debug = false, u_int32_t max_frame_size = 65536,
+ qpid::framing::ProtocolVersion* _version = &(qpid::framing::highestVersion));
~Connection();
/**
Modified: incubator/qpid/trunk/qpid/cpp/lib/common/framing/AMQFrame.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/lib/common/framing/AMQFrame.h?view=diff&rev=489212&r1=489211&r2=489212
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/lib/common/framing/AMQFrame.h (original)
+++ incubator/qpid/trunk/qpid/cpp/lib/common/framing/AMQFrame.h Wed Dec 20 14:29:38 2006
@@ -35,6 +35,11 @@
namespace qpid {
namespace framing {
+
+ // TODO - replace with generated file
+ // CCT
+ static ProtocolVersion highestVersion(8,0);
+
class AMQFrame : virtual public AMQDataBlock
{
static AMQP_MethodVersionMap versionMap;
Modified: incubator/qpid/trunk/qpid/cpp/tests/ChannelTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/tests/ChannelTest.cpp?view=diff&rev=489212&r1=489211&r2=489212
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/tests/ChannelTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/tests/ChannelTest.cpp Wed Dec 20 14:29:38 2006
@@ -132,7 +132,7 @@
void testConsumerMgmt(){
Queue::shared_ptr queue(new Queue("my_queue"));
- Channel channel(0, 0, 0);
+ Channel channel(qpid::framing::highestVersion, 0, 0, 0);
CPPUNIT_ASSERT(!channel.exists("my_consumer"));
ConnectionToken* owner = 0;
@@ -157,7 +157,7 @@
void testDeliveryNoAck(){
DummyHandler handler;
- Channel channel(&handler, 7, 10000);
+ Channel channel(qpid::framing::highestVersion, &handler, 7, 10000);
const string data("abcdefghijklmn");
@@ -184,7 +184,7 @@
void testDeliveryAndRecovery(){
DummyHandler handler;
- Channel channel(&handler, 7, 10000);
+ Channel channel(qpid::framing::highestVersion, &handler, 7, 10000);
const string data("abcdefghijklmn");
Message::shared_ptr msg(createMessage("test", "my_routing_key", "my_message_id", 14));
@@ -212,7 +212,7 @@
void testStaging(){
MockMessageStore store;
DummyHandler handler;
- Channel channel(&handler, 1, 1000/*framesize*/, &store, 10/*staging threshold*/);
+ Channel channel(qpid::framing::highestVersion, &handler, 1, 1000/*framesize*/, &store, 10/*staging threshold*/);
const string data[] = {"abcde", "fghij", "klmno"};
Message* msg = new Message(0, "my_exchange", "my_routing_key", false, false);
Modified: incubator/qpid/trunk/qpid/cpp/tests/FramingTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/tests/FramingTest.cpp?view=diff&rev=489212&r1=489211&r2=489212
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/tests/FramingTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/tests/FramingTest.cpp Wed Dec 20 14:29:38 2006
@@ -54,9 +54,7 @@
public:
-// AMQP version management change - kpvdr 2006-11-17
-// TODO: Make this class version-aware and link these hard-wired numbers to that version
- FramingTest() : buffer(100), v(8, 0) {}
+ FramingTest() : buffer(100), v(qpid::framing::highestVersion){}
void testBasicQosBody()
{
Modified: incubator/qpid/trunk/qpid/cpp/tests/MessageTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/tests/MessageTest.cpp?view=diff&rev=489212&r1=489211&r2=489212
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/tests/MessageTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/tests/MessageTest.cpp Wed Dec 20 14:29:38 2006
@@ -76,7 +76,7 @@
CPPUNIT_ASSERT_EQUAL((u_int64_t) 14, msg->contentSize());
DummyHandler handler;
- msg->deliver(&handler, 0, "ignore", 0, 100);
+ msg->deliver(&handler, 0, "ignore", 0, 100, &(qpid::framing::highestVersion));
CPPUNIT_ASSERT_EQUAL((size_t) 3, handler.frames.size());
AMQContentBody::shared_ptr contentBody(dynamic_pointer_cast<AMQContentBody, AMQBody>(handler.frames[2]->getBody()));
CPPUNIT_ASSERT(contentBody);
Modified: incubator/qpid/trunk/qpid/cpp/tests/client_test.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/tests/client_test.cpp?view=diff&rev=489212&r1=489211&r2=489212
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/tests/client_test.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/tests/client_test.cpp Wed Dec 20 14:29:38 2006
@@ -65,7 +65,7 @@
//Use a named, temporary queue
Queue queue("MyQueue", true);
- //Create and open a connection
+
Connection con(argc > 1);
string host("localhost");
con.open(host);