You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by gs...@apache.org on 2007/07/27 17:44:56 UTC
svn commit: r560285 [1/2] - in /incubator/qpid/trunk/qpid:
cpp/gentools/src/org/apache/qpid/gentools/ cpp/gentools/templ.cpp/ cpp/src/
cpp/src/qpid/broker/ cpp/src/qpid/client/ cpp/src/qpid/cluster/
cpp/src/qpid/framing/ cpp/src/tests/ python/ python/q...
Author: gsim
Date: Fri Jul 27 08:44:52 2007
New Revision: 560285
URL: http://svn.apache.org/viewvc?view=rev&rev=560285
Log:
Use execution layer to acknowledge messages.
Turn off 0-9 framing of requests and responses.
Some refactoring around message delivery.
Added:
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliveryToken.h (with props)
incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumberSet.cpp (with props)
incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumberSet.h (with props)
Removed:
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/ConsumeAdapter.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/ConsumeAdapter.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/GetAdapter.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/GetAdapter.h
Modified:
incubator/qpid/trunk/qpid/cpp/gentools/src/org/apache/qpid/gentools/AmqpMethod.java
incubator/qpid/trunk/qpid/cpp/gentools/src/org/apache/qpid/gentools/CppGenerator.java
incubator/qpid/trunk/qpid/cpp/gentools/templ.cpp/MethodBodyClass.h.tmpl
incubator/qpid/trunk/qpid/cpp/src/Makefile.am
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerChannel.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerChannel.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessage.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessage.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessageBase.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessageMessage.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessageMessage.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliveryAdapter.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageHandlerImpl.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.h
incubator/qpid/trunk/qpid/cpp/src/qpid/client/ClientChannel.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/SessionManager.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQMethodBody.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQMethodBody.h
incubator/qpid/trunk/qpid/cpp/src/qpid/framing/amqp_types.h
incubator/qpid/trunk/qpid/cpp/src/tests/BrokerChannelTest.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/FramingTest.cpp
incubator/qpid/trunk/qpid/python/cpp_failing_0-10.txt
incubator/qpid/trunk/qpid/python/qpid/codec.py
incubator/qpid/trunk/qpid/python/qpid/message.py
incubator/qpid/trunk/qpid/python/qpid/peer.py
incubator/qpid/trunk/qpid/python/qpid/spec.py
incubator/qpid/trunk/qpid/python/tests_0-10/broker.py
incubator/qpid/trunk/qpid/python/tests_0-10/dtx.py
incubator/qpid/trunk/qpid/python/tests_0-10/example.py
incubator/qpid/trunk/qpid/python/tests_0-10/message.py
incubator/qpid/trunk/qpid/python/tests_0-10/tx.py
incubator/qpid/trunk/qpid/specs/amqp-transitional.0-10.xml
Modified: incubator/qpid/trunk/qpid/cpp/gentools/src/org/apache/qpid/gentools/AmqpMethod.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/gentools/src/org/apache/qpid/gentools/AmqpMethod.java?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/gentools/src/org/apache/qpid/gentools/AmqpMethod.java (original)
+++ incubator/qpid/trunk/qpid/cpp/gentools/src/org/apache/qpid/gentools/AmqpMethod.java Fri Jul 27 08:44:52 2007
@@ -51,6 +51,7 @@
}
public boolean isResponse(AmqpVersion version) {
+ if (!CppGenerator.USE_RELIABLE_FRAMING) return false;
return (version == null) ? isResponseFlagMap.isSet() : isResponseFlagMap.isSet(version);
}
Modified: incubator/qpid/trunk/qpid/cpp/gentools/src/org/apache/qpid/gentools/CppGenerator.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/gentools/src/org/apache/qpid/gentools/CppGenerator.java?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/gentools/src/org/apache/qpid/gentools/CppGenerator.java (original)
+++ incubator/qpid/trunk/qpid/cpp/gentools/src/org/apache/qpid/gentools/CppGenerator.java Fri Jul 27 08:44:52 2007
@@ -29,6 +29,8 @@
public class CppGenerator extends Generator
{
+ protected static final boolean USE_RELIABLE_FRAMING = false;
+
protected static final String versionNamespaceStartToken = "${version_namespace_start}";
protected static final String versionNamespaceEndToken = "${version_namespace_end}";
@@ -150,12 +152,11 @@
"buffer.putLongString(#)", // encodeExpression
"buffer.getLongString(#)")); // decodeExpression
- //NB: this is WRONG! but is here as a transitional aid
typeMap.put("rfc1982-long-set", new DomainInfo(
- "u_int16_t", // type
- "2", // size
- "buffer.putShort(#)", // encodeExpression
- "# = buffer.getShort()")); // decodeExpression
+ "SequenceNumberSet", // type
+ "#.encodedSize()", // size
+ "#.encode(buffer)", // encodeExpression
+ "#.decode(buffer)")); // decodeExpression
}
public boolean isQuietFlag()
@@ -378,6 +379,7 @@
}
private String baseClass(AmqpMethod method, AmqpVersion version) {
+ if (!USE_RELIABLE_FRAMING) return "AMQMethodBody";
String base = method.isResponse(version) ? "AMQResponseBody":"AMQRequestBody";
return base;
}
@@ -787,16 +789,6 @@
sb.append(generateMethodParameterList(thisFieldMap, indentSize + (5*tabSize), false, true, true));
}
- //if (abstractMethodFlag) sb.append("const MethodContext& context");
- //boolean leadingComma = abstractMethodFlag;
- //int paramIndent = indentSize + (5*tabSize);
- // sb.append(generateMethodParameterList(thisFieldMap, paramIndent, leadingComma, true, true));
- /*
- if (!abstractMethodFlag && method.isResponse(null)) {
- if (!thisFieldMap.isEmpty()) sb.append(", \n"+Utils.createSpaces(paramIndent));
- sb.append(" RequestId responseTo");
- }
- */
sb.append(" )");
if (abstractMethodFlag)
sb.append(" = 0");
Modified: incubator/qpid/trunk/qpid/cpp/gentools/templ.cpp/MethodBodyClass.h.tmpl
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/gentools/templ.cpp/MethodBodyClass.h.tmpl?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/gentools/templ.cpp/MethodBodyClass.h.tmpl (original)
+++ incubator/qpid/trunk/qpid/cpp/gentools/templ.cpp/MethodBodyClass.h.tmpl Fri Jul 27 08:44:52 2007
@@ -37,6 +37,7 @@
#include "qpid/framing/Buffer.h"
#include "qpid/framing/FieldTable.h"
#include "qpid/framing/FramingContent.h"
+#include "qpid/framing/SequenceNumberSet.h"
namespace qpid
{
Modified: incubator/qpid/trunk/qpid/cpp/src/Makefile.am
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/Makefile.am?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/Makefile.am (original)
+++ incubator/qpid/trunk/qpid/cpp/src/Makefile.am Fri Jul 27 08:44:52 2007
@@ -155,6 +155,7 @@
qpid/framing/Requester.cpp \
qpid/framing/Responder.cpp \
qpid/framing/SequenceNumber.cpp \
+ qpid/framing/SequenceNumberSet.cpp \
qpid/framing/Correlator.cpp \
qpid/framing/Value.cpp \
qpid/framing/Proxy.cpp \
@@ -200,7 +201,6 @@
qpid/broker/Connection.cpp \
qpid/broker/ConnectionAdapter.cpp \
qpid/broker/ConnectionFactory.cpp \
- qpid/broker/ConsumeAdapter.cpp \
qpid/broker/Daemon.cpp \
qpid/broker/DeliverableMessage.cpp \
qpid/broker/DeliveryRecord.cpp \
@@ -213,7 +213,6 @@
qpid/broker/DtxWorkRecord.cpp \
qpid/broker/ExchangeRegistry.cpp \
qpid/broker/FanOutExchange.cpp \
- qpid/broker/GetAdapter.cpp \
qpid/broker/HeadersExchange.cpp \
qpid/broker/InMemoryContent.cpp \
qpid/broker/LazyLoadedContent.cpp \
@@ -258,11 +257,11 @@
qpid/broker/BrokerMessageBase.h \
qpid/broker/BrokerQueue.h \
qpid/broker/CompletionHandler.h \
- qpid/broker/ConsumeAdapter.h \
qpid/broker/Consumer.h \
qpid/broker/Deliverable.h \
qpid/broker/DeliverableMessage.h \
qpid/broker/DeliveryAdapter.h \
+ qpid/broker/DeliveryToken.h \
qpid/broker/DirectExchange.h \
qpid/broker/DtxAck.h \
qpid/broker/DtxBuffer.h \
@@ -272,7 +271,6 @@
qpid/broker/DtxWorkRecord.h \
qpid/broker/ExchangeRegistry.h \
qpid/broker/FanOutExchange.h \
- qpid/broker/GetAdapter.h \
qpid/broker/HandlerImpl.h \
qpid/broker/InMemoryContent.h \
qpid/broker/MessageBuilder.h \
@@ -362,6 +360,7 @@
qpid/framing/Responder.h \
qpid/framing/SerializeHandler.h \
qpid/framing/SequenceNumber.h \
+ qpid/framing/SequenceNumberSet.h \
qpid/framing/Value.h \
qpid/framing/Uuid.h \
qpid/framing/amqp_framing.h \
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.cpp?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.cpp Fri Jul 27 08:44:52 2007
@@ -20,8 +20,7 @@
#include "BrokerAdapter.h"
#include "BrokerChannel.h"
#include "Connection.h"
-#include "ConsumeAdapter.h"
-#include "GetAdapter.h"
+#include "DeliveryToken.h"
#include "qpid/framing/AMQMethodBody.h"
#include "qpid/Exception.h"
@@ -325,8 +324,8 @@
//need to generate name here, so we have it for the adapter (it is
//also version specific behaviour now)
if (newTag.empty()) newTag = tagGenerator.generate();
- channel.consume(std::auto_ptr<DeliveryAdapter>(new ConsumeAdapter(adapter, newTag, connection.getFrameMax())),
- newTag, queue, !noAck, exclusive, noLocal ? &connection : 0, &fields);
+ DeliveryToken::shared_ptr token(BasicMessage::createConsumeToken(newTag));
+ channel.consume(token, newTag, queue, !noAck, exclusive, noLocal ? &connection : 0, &fields);
if(!nowait) client.consumeOk(newTag);
@@ -357,8 +356,8 @@
void BrokerAdapter::BasicHandlerImpl::get(uint16_t /*ticket*/, const string& queueName, bool noAck){
Queue::shared_ptr queue = getQueue(queueName);
- GetAdapter out(adapter, queue, "", connection.getFrameMax());
- if(!channel.get(out, queue, !noAck)){
+ DeliveryToken::shared_ptr token(BasicMessage::createGetToken(queue));
+ if(!channel.get(token, queue, !noAck)){
string clusterId;//not used, part of an imatix hack
client.getEmpty(clusterId);
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerChannel.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerChannel.cpp?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerChannel.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerChannel.cpp Fri Jul 27 08:44:52 2007
@@ -49,9 +49,10 @@
using namespace qpid::sys;
-Channel::Channel(Connection& con, ChannelId _id, MessageStore* const _store) :
+Channel::Channel(Connection& con, DeliveryAdapter& _out, ChannelId _id, MessageStore* const _store) :
id(_id),
connection(con),
+ out(_out),
currentDeliveryTag(1),
prefetchSize(0),
prefetchCount(0),
@@ -76,7 +77,7 @@
// TODO aconway 2007-02-12: Why is connection token passed in instead
// of using the channel's parent connection?
-void Channel::consume(std::auto_ptr<DeliveryAdapter> adapter, string& tagInOut,
+void Channel::consume(DeliveryToken::shared_ptr token, string& tagInOut,
Queue::shared_ptr queue, bool acks,
bool exclusive, ConnectionToken* const connection,
const FieldTable*)
@@ -84,7 +85,7 @@
if(tagInOut.empty())
tagInOut = tagGenerator.generate();
std::auto_ptr<ConsumerImpl> c(
- new ConsumerImpl(this, adapter, tagInOut, queue, connection, acks));
+ new ConsumerImpl(this, token, tagInOut, queue, connection, acks));
queue->consume(c.get(), exclusive);//may throw exception
consumers.insert(tagInOut, c.release());
}
@@ -97,7 +98,8 @@
consumers.erase(i);
}
-void Channel::close(){
+void Channel::close()
+{
opened = false;
consumers.clear();
if (dtxBuffer.get()) {
@@ -106,11 +108,15 @@
recover(true);
}
-void Channel::startTx(){
+void Channel::startTx()
+{
txBuffer = TxBuffer::shared_ptr(new TxBuffer());
}
-void Channel::commit(){
+void Channel::commit()
+{
+ if (!txBuffer) throw ConnectionException(503, "Channel has not been selected for use with transactions");
+
TxOp::shared_ptr txAck(new TxAck(accumulatedAck, unacked));
txBuffer->enlist(txAck);
if (txBuffer->commitLocal(store)) {
@@ -118,16 +124,21 @@
}
}
-void Channel::rollback(){
+void Channel::rollback()
+{
+ if (!txBuffer) throw ConnectionException(503, "Channel has not been selected for use with transactions");
+
txBuffer->rollback();
accumulatedAck.clear();
}
-void Channel::selectDtx(){
+void Channel::selectDtx()
+{
dtxSelected = true;
}
-void Channel::startDtx(const std::string& xid, DtxManager& mgr, bool join){
+void Channel::startDtx(const std::string& xid, DtxManager& mgr, bool join)
+{
if (!dtxSelected) {
throw ConnectionException(503, "Channel has not been selected for use with dtx");
}
@@ -140,7 +151,8 @@
}
}
-void Channel::endDtx(const std::string& xid, bool fail){
+void Channel::endDtx(const std::string& xid, bool fail)
+{
if (!dtxBuffer) {
throw ConnectionException(503, boost::format("xid %1% not associated with this channel") % xid);
}
@@ -160,7 +172,8 @@
dtxBuffer.reset();
}
-void Channel::suspendDtx(const std::string& xid){
+void Channel::suspendDtx(const std::string& xid)
+{
if (dtxBuffer->getXid() != xid) {
throw ConnectionException(503, boost::format("xid specified on start was %1%, but %2% specified on suspend")
% dtxBuffer->getXid() % xid);
@@ -171,7 +184,8 @@
dtxBuffer->setSuspended(true);
}
-void Channel::resumeDtx(const std::string& xid){
+void Channel::resumeDtx(const std::string& xid)
+{
if (dtxBuffer->getXid() != xid) {
throw ConnectionException(503, boost::format("xid specified on start was %1%, but %2% specified on resume")
% dtxBuffer->getXid() % xid);
@@ -199,20 +213,22 @@
delivery.addTo(&outstanding);
}
-bool Channel::checkPrefetch(Message::shared_ptr& msg){
+bool Channel::checkPrefetch(Message::shared_ptr& msg)
+{
Mutex::ScopedLock locker(deliveryLock);
bool countOk = !prefetchCount || prefetchCount > unacked.size();
bool sizeOk = !prefetchSize || prefetchSize > msg->contentSize() + outstanding.size || unacked.empty();
return countOk && sizeOk;
}
-Channel::ConsumerImpl::ConsumerImpl(Channel* _parent, std::auto_ptr<DeliveryAdapter> _adapter,
+Channel::ConsumerImpl::ConsumerImpl(Channel* _parent, DeliveryToken::shared_ptr _token,
const string& _tag, Queue::shared_ptr _queue,
ConnectionToken* const _connection, bool ack
- ) : parent(_parent), adapter(_adapter), tag(_tag), queue(_queue), connection(_connection),
+ ) : parent(_parent), token(_token), tag(_tag), queue(_queue), connection(_connection),
ackExpected(ack), blocked(false) {}
-bool Channel::ConsumerImpl::deliver(Message::shared_ptr& msg){
+bool Channel::ConsumerImpl::deliver(Message::shared_ptr& msg)
+{
if(!connection || connection != msg->getPublisher()){//check for no_local
if(!parent->flowActive || (ackExpected && !parent->checkPrefetch(msg))){
blocked = true;
@@ -220,11 +236,10 @@
blocked = false;
Mutex::ScopedLock locker(parent->deliveryLock);
- uint64_t deliveryTag = adapter->getNextDeliveryTag();
+ uint64_t deliveryTag = parent->out.deliver(msg, token);
if(ackExpected){
parent->record(DeliveryRecord(msg, queue, tag, deliveryTag));
}
- adapter->deliver(msg, deliveryTag);
return true;
}
@@ -234,14 +249,15 @@
void Channel::ConsumerImpl::redeliver(Message::shared_ptr& msg, uint64_t deliveryTag) {
Mutex::ScopedLock locker(parent->deliveryLock);
- adapter->deliver(msg, deliveryTag);
+ parent->out.redeliver(msg, token, deliveryTag);
}
Channel::ConsumerImpl::~ConsumerImpl() {
cancel();
}
-void Channel::ConsumerImpl::cancel(){
+void Channel::ConsumerImpl::cancel()
+{
if(queue) {
queue->cancel(this);
if (queue->canAutoDelete()) {
@@ -251,27 +267,32 @@
}
}
-void Channel::ConsumerImpl::requestDispatch(){
+void Channel::ConsumerImpl::requestDispatch()
+{
if(blocked)
queue->requestDispatch();
}
-void Channel::handleInlineTransfer(Message::shared_ptr msg){
+void Channel::handleInlineTransfer(Message::shared_ptr msg)
+{
complete(msg);
}
-void Channel::handlePublish(Message* _message){
+void Channel::handlePublish(Message* _message)
+{
Message::shared_ptr message(_message);
messageBuilder.initialise(message);
}
-void Channel::handleHeader(AMQHeaderBody::shared_ptr header){
+void Channel::handleHeader(AMQHeaderBody::shared_ptr header)
+{
messageBuilder.setHeader(header);
//at this point, decide based on the size of the message whether we want
//to stage it by saving content directly to disk as it arrives
}
-void Channel::handleContent(AMQContentBody::shared_ptr content){
+void Channel::handleContent(AMQContentBody::shared_ptr content)
+{
messageBuilder.addContent(content);
}
@@ -306,14 +327,16 @@
}
// Used by Basic
-void Channel::ack(uint64_t deliveryTag, bool multiple){
+void Channel::ack(uint64_t deliveryTag, bool multiple)
+{
if (multiple)
ack(0, deliveryTag);
else
ack(deliveryTag, deliveryTag);
}
-void Channel::ack(uint64_t firstTag, uint64_t lastTag){
+void Channel::ack(uint64_t firstTag, uint64_t lastTag)
+{
if (txBuffer.get()) {
accumulatedAck.update(firstTag, lastTag);
//TODO: I think the outstanding prefetch size & count should be updated at this point...
@@ -355,7 +378,8 @@
}
}
-void Channel::recover(bool requeue){
+void Channel::recover(bool requeue)
+{
Mutex::ScopedLock locker(deliveryLock);//need to synchronize with possible concurrent delivery
if(requeue){
@@ -368,12 +392,12 @@
}
}
-bool Channel::get(DeliveryAdapter& adapter, Queue::shared_ptr queue, bool ackExpected){
+bool Channel::get(DeliveryToken::shared_ptr token, Queue::shared_ptr queue, bool ackExpected)
+{
Message::shared_ptr msg = queue->dequeue();
if(msg){
Mutex::ScopedLock locker(deliveryLock);
- uint64_t myDeliveryTag = adapter.getNextDeliveryTag();
- adapter.deliver(msg, myDeliveryTag);
+ uint64_t myDeliveryTag = out.deliver(msg, token);
if(ackExpected){
unacked.push_back(DeliveryRecord(msg, queue, myDeliveryTag));
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerChannel.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerChannel.h?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerChannel.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerChannel.h Fri Jul 27 08:44:52 2007
@@ -33,6 +33,7 @@
#include "Consumer.h"
#include "DeliveryAdapter.h"
#include "DeliveryRecord.h"
+#include "DeliveryToken.h"
#include "Deliverable.h"
#include "DtxBuffer.h"
#include "DtxManager.h"
@@ -64,7 +65,7 @@
class ConsumerImpl : public Consumer
{
Channel* parent;
- std::auto_ptr<DeliveryAdapter> adapter;
+ DeliveryToken::shared_ptr token;
const string tag;
Queue::shared_ptr queue;
ConnectionToken* const connection;
@@ -72,7 +73,7 @@
bool blocked;
public:
- ConsumerImpl(Channel* parent, std::auto_ptr<DeliveryAdapter> adapter,
+ ConsumerImpl(Channel* parent, DeliveryToken::shared_ptr token,
const string& tag, Queue::shared_ptr queue,
ConnectionToken* const connection, bool ack);
~ConsumerImpl();
@@ -86,6 +87,7 @@
framing::ChannelId id;
Connection& connection;
+ DeliveryAdapter& out;
uint64_t currentDeliveryTag;
Queue::shared_ptr defaultQueue;
ConsumerImplMap consumers;
@@ -110,7 +112,7 @@
void checkDtxTimeout();
public:
- Channel(Connection& parent, framing::ChannelId id, MessageStore* const store = 0);
+ Channel(Connection& parent, DeliveryAdapter& out, framing::ChannelId id, MessageStore* const store = 0);
~Channel();
bool isOpen() const { return opened; }
@@ -127,11 +129,11 @@
/**
*@param tagInOut - if empty it is updated with the generated token.
*/
- void consume(std::auto_ptr<DeliveryAdapter> adapter, string& tagInOut, Queue::shared_ptr queue, bool acks,
+ void consume(DeliveryToken::shared_ptr token, string& tagInOut, Queue::shared_ptr queue, bool acks,
bool exclusive, ConnectionToken* const connection = 0,
const framing::FieldTable* = 0);
void cancel(const string& tag);
- bool get(DeliveryAdapter& adapter, Queue::shared_ptr queue, bool ackExpected);
+ bool get(DeliveryToken::shared_ptr token, Queue::shared_ptr queue, bool ackExpected);
void close();
void startTx();
void commit();
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessage.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessage.cpp?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessage.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessage.cpp Fri Jul 27 08:44:52 2007
@@ -26,6 +26,7 @@
#include "InMemoryContent.h"
#include "LazyLoadedContent.h"
#include "MessageStore.h"
+#include "BrokerQueue.h"
#include "qpid/log/Statement.h"
#include "qpid/framing/BasicDeliverBody.h"
#include "qpid/framing/BasicGetOkBody.h"
@@ -37,6 +38,30 @@
#include "qpid/framing/ChannelAdapter.h"
#include "RecoveryManagerImpl.h"
+namespace qpid{
+namespace broker{
+
+struct BasicGetToken : DeliveryToken
+{
+ typedef boost::shared_ptr<BasicGetToken> shared_ptr;
+
+ Queue::shared_ptr queue;
+
+ BasicGetToken(Queue::shared_ptr q) : queue(q) {}
+};
+
+struct BasicConsumeToken : DeliveryToken
+{
+ typedef boost::shared_ptr<BasicConsumeToken> shared_ptr;
+
+ const string consumer;
+
+ BasicConsumeToken(const string c) : consumer(c) {}
+};
+
+}
+}
+
using namespace boost;
using namespace qpid::broker;
using namespace qpid::framing;
@@ -74,6 +99,16 @@
return header.get() && (header->getContentSize() == contentSize());
}
+DeliveryToken::shared_ptr BasicMessage::createGetToken(Queue::shared_ptr queue)
+{
+ return DeliveryToken::shared_ptr(new BasicGetToken(queue));
+}
+
+DeliveryToken::shared_ptr BasicMessage::createConsumeToken(const string& consumer)
+{
+ return DeliveryToken::shared_ptr(new BasicConsumeToken(consumer));
+}
+
void BasicMessage::deliver(ChannelAdapter& channel,
const string& consumerTag, uint64_t deliveryTag,
uint32_t framesize)
@@ -86,23 +121,39 @@
}
void BasicMessage::sendGetOk(ChannelAdapter& channel,
- const std::string& /*destination*/,
uint32_t messageCount,
- uint64_t responseTo,
+ uint64_t /*responseTo*/,
uint64_t deliveryTag,
uint32_t framesize)
{
channel.send(make_shared_ptr(
new BasicGetOkBody(
channel.getVersion(),
- responseTo,
+ //responseTo,
deliveryTag, getRedelivered(), getExchange(),
getRoutingKey(), messageCount)));
sendContent(channel, framesize);
}
-void BasicMessage::sendContent(
- ChannelAdapter& channel, uint32_t framesize)
+void BasicMessage::deliver(framing::ChannelAdapter& channel, uint64_t deliveryTag, DeliveryToken::shared_ptr token, uint32_t framesize)
+{
+ BasicConsumeToken::shared_ptr consume = dynamic_pointer_cast<BasicConsumeToken>(token);
+ if (consume) {
+ deliver(channel, consume->consumer, deliveryTag, framesize);
+ } else {
+ BasicGetToken::shared_ptr get = dynamic_pointer_cast<BasicGetToken>(token);
+ if (get) {
+ uint64_t request(1/*actual value doesn't affect anything at present*/);
+ sendGetOk(channel, get->queue->getMessageCount(), request, deliveryTag, framesize);
+ } else {
+ //TODO:
+ //either need to be able to convert to a message transfer or
+ //throw error of some kind to allow this to be handled higher up
+ }
+ }
+}
+
+void BasicMessage::sendContent(ChannelAdapter& channel, uint32_t framesize)
{
channel.send(header);
Mutex::ScopedLock locker(contentLock);
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessage.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessage.h?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessage.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessage.h Fri Jul 27 08:44:52 2007
@@ -43,6 +43,7 @@
namespace broker {
class MessageStore;
+class Queue;
using framing::string;
/**
@@ -70,13 +71,16 @@
void addContent(framing::AMQContentBody::shared_ptr data);
bool isComplete();
+ static DeliveryToken::shared_ptr createGetToken(boost::shared_ptr<Queue> queue);
+ static DeliveryToken::shared_ptr createConsumeToken(const string& consumer);
+ void deliver(framing::ChannelAdapter& channel, uint64_t deliveryTag, DeliveryToken::shared_ptr token, uint32_t framesize);
+
void deliver(framing::ChannelAdapter&,
const string& consumerTag,
uint64_t deliveryTag,
uint32_t framesize);
void sendGetOk(framing::ChannelAdapter& channel,
- const std::string& destination,
uint32_t messageCount,
uint64_t responseTo,
uint64_t deliveryTag,
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessageBase.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessageBase.h?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessageBase.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessageBase.h Fri Jul 27 08:44:52 2007
@@ -25,6 +25,7 @@
#include <string>
#include <boost/shared_ptr.hpp>
#include "Content.h"
+#include "DeliveryToken.h"
#include "PersistableMessage.h"
#include "qpid/framing/amqp_types.h"
@@ -91,23 +92,9 @@
void setPersistenceId(uint64_t _persistenceId) const { persistenceId = _persistenceId; }
void redeliver() { redelivered = true; }
- /**
- * Used to deliver the message from the queue
- */
- virtual void deliver(framing::ChannelAdapter& channel,
- const std::string& consumerTag,
- uint64_t deliveryTag,
- uint32_t framesize) = 0;
- /**
- * Used to return a message in response to a get from a queue
- */
- virtual void sendGetOk(framing::ChannelAdapter& channel,
- const std::string& destination,
- uint32_t messageCount,
- uint64_t responseTo,
- uint64_t deliveryTag,
- uint32_t framesize) = 0;
-
+ virtual void deliver(framing::ChannelAdapter& channel, uint64_t deliveryTag/*only needed for basic class*/,
+ DeliveryToken::shared_ptr token, uint32_t framesize) = 0;
+
virtual bool isComplete() = 0;
virtual uint64_t contentSize() const = 0;
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessageMessage.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessageMessage.cpp?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessageMessage.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessageMessage.cpp Fri Jul 27 08:44:52 2007
@@ -34,10 +34,18 @@
#include <algorithm>
using namespace std;
+using namespace boost;
using namespace qpid::framing;
namespace qpid {
namespace broker {
+
+struct MessageDeliveryToken : public DeliveryToken
+{
+ const std::string destination;
+
+ MessageDeliveryToken(const std::string& d) : destination(d) {}
+};
MessageMessage::MessageMessage(
ConnectionToken* publisher, RequestId requestId_, TransferPtr transfer_
@@ -179,22 +187,13 @@
channel.send(make_shared_ptr(new MessageCloseBody(channel.getVersion(), ref->getId())));
}
-void MessageMessage::deliver(
- framing::ChannelAdapter& channel,
- const std::string& consumerTag,
- uint64_t /*deliveryTag*/,
- uint32_t framesize)
+
+void MessageMessage::deliver(ChannelAdapter& channel, uint64_t, DeliveryToken::shared_ptr token, uint32_t framesize)
{
- transferMessage(channel, consumerTag, framesize);
+ transferMessage(channel, shared_polymorphic_cast<MessageDeliveryToken>(token)->destination, framesize);
}
-void MessageMessage::sendGetOk(
- framing::ChannelAdapter& channel,
- const std::string& destination,
- uint32_t /*messageCount*/,
- uint64_t /*responseTo*/,
- uint64_t /*deliveryTag*/,
- uint32_t framesize)
+void MessageMessage::deliver(ChannelAdapter& channel, const std::string& destination, uint32_t framesize)
{
transferMessage(channel, destination, framesize);
}
@@ -321,6 +320,10 @@
return reference;
}
+DeliveryToken::shared_ptr MessageMessage::getToken(const std::string& destination)
+{
+ return DeliveryToken::shared_ptr(new MessageDeliveryToken(destination));
+}
}} // namespace qpid::broker
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessageMessage.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessageMessage.h?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessageMessage.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessageMessage.h Fri Jul 27 08:44:52 2007
@@ -53,17 +53,8 @@
TransferPtr getTransfer() const { return transfer; }
ReferencePtr getReference() const ;
- void deliver(framing::ChannelAdapter& channel,
- const std::string& consumerTag,
- uint64_t deliveryTag,
- uint32_t framesize);
-
- void sendGetOk(framing::ChannelAdapter& channel,
- const std::string& destination,
- uint32_t messageCount,
- uint64_t responseTo,
- uint64_t deliveryTag,
- uint32_t framesize);
+ void deliver(framing::ChannelAdapter& channel, uint64_t deliveryTag, DeliveryToken::shared_ptr token, uint32_t framesize);
+ void deliver(framing::ChannelAdapter&, const std::string& destination, uint32_t framesize);
bool isComplete();
@@ -80,6 +71,8 @@
uint64_t expectedContentSize();
void decodeHeader(framing::Buffer& buffer);
void decodeContent(framing::Buffer& buffer, uint32_t contentChunkSize = 0);
+
+ static DeliveryToken::shared_ptr getToken(const std::string& destination);
private:
void transferMessage(
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliveryAdapter.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliveryAdapter.h?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliveryAdapter.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliveryAdapter.h Fri Jul 27 08:44:52 2007
@@ -22,11 +22,13 @@
#define _DeliveryAdapter_
#include "BrokerMessageBase.h"
+#include "DeliveryToken.h"
#include "qpid/framing/amqp_types.h"
namespace qpid {
namespace broker {
+ typedef framing::RequestId DeliveryId;
/**
* The intention behind this interface is to separate the generic
* handling of some form of message delivery to clients that is
@@ -40,8 +42,8 @@
class DeliveryAdapter
{
public:
- virtual framing::RequestId getNextDeliveryTag() = 0;
- virtual void deliver(Message::shared_ptr& msg, framing::RequestId tag) = 0;
+ virtual DeliveryId deliver(Message::shared_ptr& msg, DeliveryToken::shared_ptr token) = 0;
+ virtual void redeliver(Message::shared_ptr& msg, DeliveryToken::shared_ptr token, DeliveryId tag) = 0;
virtual ~DeliveryAdapter(){}
};
Added: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliveryToken.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliveryToken.h?view=auto&rev=560285
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliveryToken.h (added)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliveryToken.h Fri Jul 27 08:44:52 2007
@@ -0,0 +1,45 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#ifndef _DeliveryToken_
+#define _DeliveryToken_
+
+#include <boost/shared_ptr.hpp>
+
+namespace qpid {
+namespace broker {
+
+ /**
+ * A DeliveryToken allows the delivery of a message to be
+ * associated with whatever mechanism caused it to be
+ * delivered. (i.e. its a form of Memento).
+ */
+ class DeliveryToken
+ {
+ public:
+ typedef boost::shared_ptr<DeliveryToken> shared_ptr;
+
+ virtual ~DeliveryToken(){}
+ };
+
+}}
+
+
+#endif
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliveryToken.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliveryToken.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageHandlerImpl.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageHandlerImpl.cpp?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageHandlerImpl.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageHandlerImpl.cpp Fri Jul 27 08:44:52 2007
@@ -21,8 +21,6 @@
#include "BrokerChannel.h"
#include "qpid/framing/FramingContent.h"
#include "Connection.h"
-#include "ConsumeAdapter.h"
-#include "GetAdapter.h"
#include "Broker.h"
#include "BrokerMessageMessage.h"
#include "qpid/framing/MessageAppendBody.h"
@@ -45,66 +43,44 @@
MessageHandlerImpl::cancel(const string& destination )
{
channel.cancel(destination);
- //client.ok();
}
void
-MessageHandlerImpl::open(const string& reference)
+MessageHandlerImpl::open(const string& /*reference*/)
{
- references.open(reference);
- //client.ok();
+ throw ConnectionException(540, "References no longer supported");
}
void
-MessageHandlerImpl::append(const framing::MethodContext& context)
+MessageHandlerImpl::append(const framing::MethodContext& /*context*/)
{
- MessageAppendBody::shared_ptr body(boost::shared_polymorphic_downcast<MessageAppendBody>(context.methodBody));
- references.get(body->getReference())->append(body);
- //client.ok();
+ throw ConnectionException(540, "References no longer supported");
}
void
-MessageHandlerImpl::close(const string& reference)
+MessageHandlerImpl::close(const string& /*reference*/)
{
- Reference::shared_ptr ref = references.get(reference);
- //client.ok();
-
- // Send any transfer messages to their correct exchanges and okay them
- const Reference::Messages& msgs = ref->getMessages();
- for (Reference::Messages::const_iterator m = msgs.begin(); m != msgs.end(); ++m) {
- channel.handleInlineTransfer(*m);
- client.setResponseTo((*m)->getRequestId());
- client.ok();
- }
- ref->close();
+ throw ConnectionException(540, "References no longer supported");
}
void
MessageHandlerImpl::checkpoint(const string& /*reference*/,
const string& /*identifier*/ )
{
- // Initial implementation (which is conforming) is to do nothing here
- // and return offset zero for the resume
- //client.ok();
+ throw ConnectionException(540, "References no longer supported");
}
void
-MessageHandlerImpl::resume(const string& reference,
+MessageHandlerImpl::resume(const string& /*reference*/,
const string& /*identifier*/ )
{
- // Initial (null) implementation
- // open reference and return 0 offset
- references.open(reference);
- client.offset(0);
+ throw ConnectionException(540, "References no longer supported");
}
void
MessageHandlerImpl::offset(uint64_t /*value*/ )
{
- // Shouldn't ever receive this as it is reponse to resume
- // which is never sent
- // TODO astitcher 2007-02-16 What is the correct exception to throw here?
- THROW_QPID_ERROR(INTERNAL_ERROR, "impossible");
+ throw ConnectionException(540, "References no longer supported");
}
void
@@ -120,14 +96,12 @@
if(!destination.empty() && channel.exists(destination))
throw ConnectionException(530, "Consumer tags must be unique");
string tag = destination;
- channel.consume(std::auto_ptr<DeliveryAdapter>(new ConsumeAdapter(adapter, destination, connection.getFrameMax())),
- tag, queue, !noAck, exclusive,
- noLocal ? &connection : 0, &filter);
- //client.ok();
+ channel.consume(MessageMessage::getToken(destination), tag, queue, !noAck, exclusive, noLocal ? &connection : 0, &filter);
// Dispatch messages as there is now a consumer.
queue->requestDispatch();
}
+
void
MessageHandlerImpl::get(uint16_t /*ticket*/,
const string& queueName,
@@ -136,11 +110,11 @@
{
Queue::shared_ptr queue = getQueue(queueName);
- GetAdapter out(adapter, queue, destination, connection.getFrameMax());
- if(channel.get(out, queue, !noAck)) {
- client.ok();
+ if (channel.get(MessageMessage::getToken(destination), queue, !noAck)){
+ //don't send any response... rely on execution completion
} else {
- client.empty();
+ //temporarily disabled:
+ //client.empty();
}
}
@@ -167,14 +141,12 @@
//TODO: handle global
channel.setPrefetchSize(prefetchSize);
channel.setPrefetchCount(prefetchCount);
- //client.ok();
}
void
MessageHandlerImpl::recover(bool requeue)
{
channel.recover(requeue);
- //client.ok();
}
void
@@ -193,11 +165,8 @@
if (transfer->getBody().isInline()) {
MessageMessage::shared_ptr message(new MessageMessage(&connection, requestId, transfer));
channel.handleInlineTransfer(message);
- client.ok();
} else {
- Reference::shared_ptr ref(references.get(transfer->getBody().getValue()));
- MessageMessage::shared_ptr message(new MessageMessage(&connection, requestId, transfer, ref));
- ref->addMessage(message);
+ throw ConnectionException(540, "References no longer supported");
}
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.cpp?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.cpp Fri Jul 27 08:44:52 2007
@@ -25,10 +25,11 @@
using namespace qpid::broker;
using namespace qpid::framing;
+using namespace qpid::sys;
SemanticHandler::SemanticHandler(ChannelId id, Connection& c) :
connection(c),
- channel(c, id, &c.broker.getStore())
+ channel(c, *this, id, &c.broker.getStore())
{
init(id, connection.getOutput(), connection.getVersion());
adapter = std::auto_ptr<BrokerAdapter>(new BrokerAdapter(channel, connection, connection.broker, *this));
@@ -75,10 +76,24 @@
}
}
-void SemanticHandler::complete(uint32_t mark, uint16_t /*range- not decoded correctly yet*/)
+void SemanticHandler::complete(uint32_t cumulative, SequenceNumberSet range)
{
- //just record it for now (will eventually need to use it to ack messages):
- outgoing.lwm = SequenceNumber(mark);
+ //record:
+ SequenceNumber mark(cumulative);
+ if (outgoing.lwm < mark) {
+ outgoing.lwm = mark;
+ //ack messages:
+ channel.ack(mark.getValue(), true);
+ //std::cout << "[" << this << "] acknowledged: " << mark << std::endl;
+ }
+ if (range.size() % 2) { //must be even number
+ throw ConnectionException(530, "Received odd number of elements in ranged mark");
+ } else {
+ //TODO: need to keep a record of the full range previously acked
+ for (SequenceNumberSet::iterator i = range.begin(); i != range.end(); i++) {
+ channel.ack((uint64_t) i->getValue(), (uint64_t) (++i)->getValue());
+ }
+ }
}
void SemanticHandler::flush()
@@ -86,8 +101,8 @@
//flush doubles as a sync to begin with - send an execution.complete
incoming.lwm = incoming.hwm;
if (isOpen()) {
- /*use dummy value for range which is not yet encoded correctly*/
- send(make_shared_ptr(new ExecutionCompleteBody(getVersion(), incoming.hwm.getValue(), 0)));
+ Mutex::ScopedLock l(outLock);
+ ChannelAdapter::send(make_shared_ptr(new ExecutionCompleteBody(getVersion(), incoming.hwm.getValue(), SequenceNumberSet())));
}
}
@@ -140,3 +155,28 @@
channel.handleHeartbeat(body);
}
+DeliveryId SemanticHandler::deliver(Message::shared_ptr& msg, DeliveryToken::shared_ptr token)
+{
+ Mutex::ScopedLock l(outLock);
+ SequenceNumber copy(outgoing.hwm);
+ ++copy;
+ msg->deliver(*this, copy.getValue(), token, connection.getFrameMax());
+ //std::cout << "[" << this << "] delivered: " << outgoing.hwm.getValue() << std::endl;
+ return outgoing.hwm.getValue();
+}
+
+void SemanticHandler::redeliver(Message::shared_ptr& msg, DeliveryToken::shared_ptr token, DeliveryId tag)
+{
+ msg->deliver(*this, tag, token, connection.getFrameMax());
+}
+
+RequestId SemanticHandler::send(shared_ptr<AMQBody> body, Correlator::Action action)
+{
+ Mutex::ScopedLock l(outLock);
+ uint8_t type(body->type());
+ if (type == REQUEST_BODY || type == RESPONSE_BODY || type == METHOD_BODY) {
+ ++outgoing.hwm;
+ //std::cout << "[" << this << "] allocated: " << outgoing.hwm.getValue() << " to " << *body << std::endl;
+ }
+ return ChannelAdapter::send(body, action);
+}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.h?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.h Fri Jul 27 08:44:52 2007
@@ -24,6 +24,7 @@
#include <memory>
#include "BrokerChannel.h"
#include "Connection.h"
+#include "DeliveryAdapter.h"
#include "qpid/framing/amqp_types.h"
#include "qpid/framing/AMQP_ServerOperations.h"
#include "qpid/framing/FrameHandler.h"
@@ -36,6 +37,7 @@
class framing::ChannelAdapter;
class SemanticHandler : private framing::ChannelAdapter,
+ private DeliveryAdapter,
public framing::FrameHandler,
public framing::AMQP_ServerOperations::ExecutionHandler
{
@@ -44,6 +46,7 @@
std::auto_ptr<BrokerAdapter> adapter;
framing::Window incoming;
framing::Window outgoing;
+ sys::Mutex outLock;
void handleL4(boost::shared_ptr<qpid::framing::AMQMethodBody> method,
const qpid::framing::MethodContext& context);
@@ -55,12 +58,22 @@
void handleHeader(boost::shared_ptr<qpid::framing::AMQHeaderBody>);
void handleContent(boost::shared_ptr<qpid::framing::AMQContentBody>);
void handleHeartbeat(boost::shared_ptr<qpid::framing::AMQHeartbeatBody>);
+
+ framing::RequestId send(shared_ptr<framing::AMQBody> body, framing::Correlator::Action action=framing::Correlator::Action());
+
+
+ //delivery adapter methods:
+ DeliveryId deliver(Message::shared_ptr& msg, DeliveryToken::shared_ptr token);
+ void redeliver(Message::shared_ptr& msg, DeliveryToken::shared_ptr token, DeliveryId tag);
+
public:
SemanticHandler(framing::ChannelId id, Connection& c);
+
+ //frame handler:
void handle(framing::AMQFrame& frame);
//execution class method handlers:
- void complete(uint32_t cumulativeExecutionMark, uint16_t);
+ void complete(uint32_t cumulativeExecutionMark, framing::SequenceNumberSet range);
void flush();
};
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/ClientChannel.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/ClientChannel.cpp?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/ClientChannel.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/ClientChannel.cpp Fri Jul 27 08:44:52 2007
@@ -77,7 +77,7 @@
connection->connector->init(); // Send ProtocolInit block.
ConnectionStartBody::shared_ptr connectionStart =
responses.receive<ConnectionStartBody>();
-
+
FieldTable props;
string mechanism("PLAIN");
string response = ((char)0) + uid + ((char)0) + pwd;
@@ -85,7 +85,7 @@
ConnectionTuneBody::shared_ptr proposal =
sendAndReceive<ConnectionTuneBody>(
make_shared_ptr(new ConnectionStartOkBody(
- version, connectionStart->getRequestId(),
+ version, //connectionStart->getRequestId(),
props, mechanism,
response, locale)));
@@ -98,7 +98,7 @@
**/
sendCommand(make_shared_ptr(new ConnectionTuneOkBody(
- version, proposal->getRequestId(),
+ version, //proposal->getRequestId(),
proposal->getChannelMax(), connection->getMaxFrameSize(),
proposal->getHeartbeat())));
@@ -222,10 +222,10 @@
}
}
-void Channel::handleChannel(AMQMethodBody::shared_ptr method, const MethodContext& ctxt) {
+void Channel::handleChannel(AMQMethodBody::shared_ptr method, const MethodContext& /*ctxt*/) {
switch (method->amqpMethodId()) {
case ChannelCloseBody::METHOD_ID:
- sendCommand(make_shared_ptr(new ChannelCloseOkBody(version, ctxt.getRequestId())));
+ sendCommand(make_shared_ptr(new ChannelCloseOkBody(version/*, ctxt.getRequestId()*/)));
peerClose(shared_polymorphic_downcast<ChannelCloseBody>(method));
return;
case ChannelFlowBody::METHOD_ID:
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/SessionManager.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/SessionManager.cpp?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/SessionManager.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/SessionManager.cpp Fri Jul 27 08:44:52 2007
@@ -36,7 +36,7 @@
using namespace broker;
/** Handler to send frames direct to local broker (bypass correlation etc.) */
-struct BrokerHandler : public FrameHandler, private ChannelAdapter {
+ struct BrokerHandler : public FrameHandler, private ChannelAdapter, private DeliveryAdapter {
Connection connection;
Channel channel;
BrokerAdapter adapter;
@@ -51,7 +51,7 @@
//
BrokerHandler(Broker& broker) :
connection(0, broker),
- channel(connection, 1, 0),
+ channel(connection, *this, 1, 0),
adapter(channel, connection, broker, *this) {}
void handle(AMQFrame& frame) {
@@ -68,6 +68,10 @@
virtual void handleMethodInContext(shared_ptr<AMQMethodBody>, const MethodContext&){}
// No-op send.
virtual RequestId send(shared_ptr<AMQBody>, Correlator::Action) { return 0; }
+
+ //delivery adapter methods, also no-ops:
+ virtual DeliveryId deliver(Message::shared_ptr&, DeliveryToken::shared_ptr) { return 0; }
+ virtual void redeliver(Message::shared_ptr&, DeliveryToken::shared_ptr, DeliveryId) {}
};
/** Wrap plain AMQFrames in SessionFrames */
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQMethodBody.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQMethodBody.cpp?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQMethodBody.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQMethodBody.cpp Fri Jul 27 08:44:52 2007
@@ -60,4 +60,9 @@
decodeContent(buffer);
}
+void AMQMethodBody::encode(Buffer& buffer) const {
+ encodeId(buffer);
+ encodeContent(buffer);
+}
+
}} // namespace qpid::framing
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQMethodBody.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQMethodBody.h?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQMethodBody.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQMethodBody.h Fri Jul 27 08:44:52 2007
@@ -47,6 +47,7 @@
AMQMethodBody(ProtocolVersion ver) : version(ver) {}
virtual ~AMQMethodBody() {}
void decode(Buffer&, uint32_t);
+ virtual void encode(Buffer& buffer) const;
virtual MethodId amqpMethodId() const = 0;
virtual ClassId amqpClassId() const = 0;
@@ -64,8 +65,8 @@
virtual bool isRequest() const { return false; }
virtual bool isResponse() const { return false; }
- protected:
static uint32_t baseSize() { return 4; }
+ protected:
struct ClassMethodId {
uint16_t classId;
@@ -76,6 +77,9 @@
void encodeId(Buffer& buffer) const;
virtual void encodeContent(Buffer& buffer) const = 0;
virtual void decodeContent(Buffer& buffer) = 0;
+
+ virtual void printPrefix(std::ostream&) const {}
+
};
Added: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumberSet.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumberSet.cpp?view=auto&rev=560285
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumberSet.cpp (added)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumberSet.cpp Fri Jul 27 08:44:52 2007
@@ -0,0 +1,61 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 "SequenceNumberSet.h"
+
+using namespace qpid::framing;
+
+void SequenceNumberSet::encode(Buffer& buffer) const
+{
+ buffer.putShort(size());
+ for (const_iterator i = begin(); i != end(); i++) {
+ buffer.putLong(i->getValue());
+ }
+}
+
+void SequenceNumberSet::decode(Buffer& buffer)
+{
+ uint16_t count = buffer.getShort();
+ for (uint16_t i = 0; i < count; i++) {
+ push_back(SequenceNumber(buffer.getLong()));
+ }
+}
+
+uint32_t SequenceNumberSet::encodedSize() const
+{
+ return 2 /*count*/ + (size() * 4);
+}
+
+namespace qpid{
+namespace framing{
+
+std::ostream& operator<<(std::ostream& out, const SequenceNumberSet& set) {
+ out << "{";
+ for (SequenceNumberSet::const_iterator i = set.begin(); i != set.end(); i++) {
+ if (i != set.begin()) out << ", ";
+ out << (i->getValue());
+ }
+ out << "}";
+ return out;
+}
+
+}
+}
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumberSet.cpp
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumberSet.cpp
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumberSet.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumberSet.h?view=auto&rev=560285
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumberSet.h (added)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumberSet.h Fri Jul 27 08:44:52 2007
@@ -0,0 +1,49 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#ifndef _framing_SequenceNumberSet_h
+#define _framing_SequenceNumberSet_h
+
+#include <ostream>
+#include <vector>
+#include "amqp_types.h"
+#include "Buffer.h"
+#include "SequenceNumber.h"
+
+namespace qpid {
+namespace framing {
+
+class SequenceNumberSet : public std::vector<SequenceNumber>
+{
+public:
+ typedef std::vector<SequenceNumber>::const_iterator const_iterator;
+ typedef std::vector<SequenceNumber>::iterator iterator;
+
+ void encode(Buffer& buffer) const;
+ void decode(Buffer& buffer);
+ uint32_t encodedSize() const;
+
+ friend std::ostream& operator<<(std::ostream&, const SequenceNumberSet&);
+};
+
+}} // namespace qpid::framing
+
+
+#endif
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumberSet.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumberSet.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/amqp_types.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/amqp_types.h?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/amqp_types.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/amqp_types.h Fri Jul 27 08:44:52 2007
@@ -53,6 +53,7 @@
// Types represented by classes.
class Content;
class FieldTable;
+class SequenceNumberSet;
// Useful constants
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/BrokerChannelTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/BrokerChannelTest.cpp?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/BrokerChannelTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/BrokerChannelTest.cpp Fri Jul 27 08:44:52 2007
@@ -48,30 +48,21 @@
void close() {};
};
-struct DeliveryRecorder
+struct DeliveryRecorder : DeliveryAdapter
{
- typedef std::pair<Message::shared_ptr, RequestId> Delivery;
+ DeliveryId id;
+ typedef std::pair<Message::shared_ptr, DeliveryToken::shared_ptr> Delivery;
std::vector<Delivery> delivered;
- struct Adapter : DeliveryAdapter
+ DeliveryId deliver(Message::shared_ptr& msg, DeliveryToken::shared_ptr token)
{
- RequestId id;
- DeliveryRecorder& recorder;
-
- Adapter(DeliveryRecorder& r) : recorder(r) {}
-
- RequestId getNextDeliveryTag() { return id + 1; }
- void deliver(Message::shared_ptr& msg, RequestId tag)
- {
- recorder.delivered.push_back(Delivery(msg, tag));
- id++;
- }
-
- };
+ delivered.push_back(Delivery(msg, token));
+ return ++id;
+ }
- std::auto_ptr<DeliveryAdapter> createAdapter()
+ void redeliver(Message::shared_ptr& msg, DeliveryToken::shared_ptr token, DeliveryId /*tag*/)
{
- return std::auto_ptr<DeliveryAdapter>(new Adapter(*this));
+ delivered.push_back(Delivery(msg, token));
}
};
@@ -166,6 +157,7 @@
}
};
+ DeliveryRecorder recorder;
public:
@@ -179,13 +171,13 @@
void testConsumerMgmt(){
Queue::shared_ptr queue(new Queue("my_queue"));
- Channel channel(connection, 0, 0);
+ Channel channel(connection, recorder, 0, 0);
channel.open();
CPPUNIT_ASSERT(!channel.exists("my_consumer"));
ConnectionToken* owner = 0;
string tag("my_consumer");
- std::auto_ptr<DeliveryAdapter> unused;
+ DeliveryToken::shared_ptr unused;
channel.consume(unused, tag, queue, false, false, owner);
string tagA;
string tagB;
@@ -205,24 +197,25 @@
}
void testDeliveryNoAck(){
- Channel channel(connection, 7);
+ Channel channel(connection, recorder, 7);
Message::shared_ptr msg(createMessage("test", "my_routing_key", "my_message_id", 14));
Queue::shared_ptr queue(new Queue("my_queue"));
- DeliveryRecorder recorder;
string tag("test");
- channel.consume(recorder.createAdapter(), tag, queue, false, false, 0);
+ DeliveryToken::shared_ptr token(BasicMessage::createConsumeToken("my-token"));
+ channel.consume(token, tag, queue, false, false, 0);
queue->deliver(msg);
sleep(2);
CPPUNIT_ASSERT_EQUAL((size_t) 1, recorder.delivered.size());
CPPUNIT_ASSERT_EQUAL(msg, recorder.delivered.front().first);
+ CPPUNIT_ASSERT_EQUAL(token, recorder.delivered.front().second);
}
void testStaging(){
MockMessageStore store;
connection.setFrameMax(1000);
connection.setStagingThreshold(10);
- Channel channel(connection, 1, &store);
+ Channel channel(connection, recorder, 1, &store);
const string data[] = {"abcde", "fghij", "klmno"};
Message* msg = new BasicMessage(0, "my_exchange", "my_routing_key", false, false);
@@ -314,7 +307,7 @@
}
void testFlow(){
- Channel channel(connection, 7);
+ Channel channel(connection, recorder, 7);
channel.open();
//there will always be a connection-start frame
CPPUNIT_ASSERT_EQUAL((size_t) 1, handler.frames.size());
@@ -327,9 +320,9 @@
Message::shared_ptr msg(createMessage("test", "my_routing_key", "my_message_id", 14));
addContent(msg, data);
Queue::shared_ptr queue(new Queue("my_queue"));
- DeliveryRecorder recorder;
string tag("test");
- channel.consume(recorder.createAdapter(), tag, queue, false, false, 0);
+ DeliveryToken::shared_ptr token(BasicMessage::createConsumeToken("my-token"));
+ channel.consume(token, tag, queue, false, false, 0);
channel.flow(false);
queue->deliver(msg);
//ensure no messages have been delivered
@@ -340,6 +333,7 @@
//ensure no messages have been delivered
CPPUNIT_ASSERT_EQUAL((size_t) 1, recorder.delivered.size());
CPPUNIT_ASSERT_EQUAL(msg, recorder.delivered.front().first);
+ CPPUNIT_ASSERT_EQUAL(token, recorder.delivered.front().second);
}
Message* createMessage(const string& exchange, const string& routingKey, const string& messageId, uint64_t contentSize)
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/FramingTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/FramingTest.cpp?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/FramingTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/FramingTest.cpp Fri Jul 27 08:44:52 2007
@@ -108,7 +108,7 @@
{
std::string a = "hostA";
std::string b = "hostB";
- ConnectionRedirectBody in(version, 0, a, b);
+ ConnectionRedirectBody in(version, a, b);
in.encodeContent(buffer);
buffer.flip();
ConnectionRedirectBody out(version);
@@ -146,7 +146,7 @@
std::string a = "hostA";
std::string b = "hostB";
AMQFrame in(version, 999,
- new ConnectionRedirectBody(version, 0, a, b));
+ new ConnectionRedirectBody(version, a, b));
in.encode(buffer);
buffer.flip();
AMQFrame out;
@@ -157,7 +157,7 @@
void testBasicConsumeOkBodyFrame()
{
std::string s = "hostA";
- AMQFrame in(version, 999, new BasicConsumeOkBody(version, 0, s));
+ AMQFrame in(version, 999, new BasicConsumeOkBody(version, s));
in.encode(buffer);
buffer.flip();
AMQFrame out;
@@ -400,22 +400,22 @@
c.declareQueue(queue);
c.bind(exchange, queue, "MyTopic", framing::FieldTable());
broker::InProcessBroker::Conversation::const_iterator i = ibroker.conversation.begin();
- ASSERT_FRAME("BROKER: Frame[channel=0; request(id=1,mark=0): ConnectionStart: versionMajor=0; versionMinor=10; serverProperties={}; mechanisms=PLAIN; locales=en_US]", *i++);
- ASSERT_FRAME("CLIENT: Frame[channel=0; response(id=1,request=1,batch=0): ConnectionStartOk: clientProperties={}; mechanism=PLAIN; response=\000guest\000guest; locale=en_US]", *i++);
- ASSERT_FRAME("BROKER: Frame[channel=0; request(id=2,mark=1): ConnectionTune: channelMax=32767; frameMax=65536; heartbeat=0]", *i++);
- ASSERT_FRAME("CLIENT: Frame[channel=0; response(id=2,request=2,batch=0): ConnectionTuneOk: channelMax=32767; frameMax=65536; heartbeat=0]", *i++);
- ASSERT_FRAME("CLIENT: Frame[channel=0; request(id=1,mark=0): ConnectionOpen: virtualHost=/; capabilities=; insist=1]", *i++);
- ASSERT_FRAME("BROKER: Frame[channel=0; response(id=1,request=1,batch=0): ConnectionOpenOk: knownHosts=]", *i++);
- ASSERT_FRAME("CLIENT: Frame[channel=1; request(id=1,mark=0): ChannelOpen: outOfBand=]", *i++);
- ASSERT_FRAME("BROKER: Frame[channel=1; response(id=1,request=1,batch=0): ChannelOpenOk: ]", *i++);
- ASSERT_FRAME("CLIENT: Frame[channel=1; request(id=2,mark=1): ExchangeDeclare: ticket=0; exchange=MyExchange; type=topic; alternateExchange=; passive=0; durable=0; autoDelete=0; arguments={}]", *i++);
- ASSERT_FRAME("CLIENT: Frame[channel=1; request(id=3,mark=1): ExecutionFlush: ]", *i++);
- ASSERT_FRAME("BROKER: Frame[channel=1; request(id=1,mark=0): ExecutionComplete: cumulativeExecutionMark=2; rangedExecutionSet=0]", *i++);
- ASSERT_FRAME("CLIENT: Frame[channel=1; request(id=4,mark=1): QueueDeclare: ticket=0; queue=MyQueue; alternateExchange=; passive=0; durable=0; exclusive=1; autoDelete=1; nowait=0; arguments={}]", *i++);
- ASSERT_FRAME("BROKER: Frame[channel=1; response(id=2,request=4,batch=0): QueueDeclareOk: queue=MyQueue; messageCount=0; consumerCount=0]", *i++);
- ASSERT_FRAME("CLIENT: Frame[channel=1; request(id=5,mark=2): QueueBind: ticket=0; queue=MyQueue; exchange=MyExchange; routingKey=MyTopic; arguments={}]", *i++);
- ASSERT_FRAME("CLIENT: Frame[channel=1; request(id=6,mark=2): ExecutionFlush: ]", *i++);
- ASSERT_FRAME("BROKER: Frame[channel=1; request(id=2,mark=0): ExecutionComplete: cumulativeExecutionMark=4; rangedExecutionSet=0]", *i++);
+ ASSERT_FRAME("BROKER: Frame[channel=0; ConnectionStart: versionMajor=0; versionMinor=10; serverProperties={}; mechanisms=PLAIN; locales=en_US]", *i++);
+ ASSERT_FRAME("CLIENT: Frame[channel=0; ConnectionStartOk: clientProperties={}; mechanism=PLAIN; response=\000guest\000guest; locale=en_US]", *i++);
+ ASSERT_FRAME("BROKER: Frame[channel=0; ConnectionTune: channelMax=32767; frameMax=65536; heartbeat=0]", *i++);
+ ASSERT_FRAME("CLIENT: Frame[channel=0; ConnectionTuneOk: channelMax=32767; frameMax=65536; heartbeat=0]", *i++);
+ ASSERT_FRAME("CLIENT: Frame[channel=0; ConnectionOpen: virtualHost=/; capabilities=; insist=1]", *i++);
+ ASSERT_FRAME("BROKER: Frame[channel=0; ConnectionOpenOk: knownHosts=]", *i++);
+ ASSERT_FRAME("CLIENT: Frame[channel=1; ChannelOpen: outOfBand=]", *i++);
+ ASSERT_FRAME("BROKER: Frame[channel=1; ChannelOpenOk: ]", *i++);
+ ASSERT_FRAME("CLIENT: Frame[channel=1; ExchangeDeclare: ticket=0; exchange=MyExchange; type=topic; alternateExchange=; passive=0; durable=0; autoDelete=0; arguments={}]", *i++);
+ ASSERT_FRAME("CLIENT: Frame[channel=1; ExecutionFlush: ]", *i++);
+ ASSERT_FRAME("BROKER: Frame[channel=1; ExecutionComplete: cumulativeExecutionMark=2; rangedExecutionSet={}]", *i++);
+ ASSERT_FRAME("CLIENT: Frame[channel=1; QueueDeclare: ticket=0; queue=MyQueue; alternateExchange=; passive=0; durable=0; exclusive=1; autoDelete=1; nowait=0; arguments={}]", *i++);
+ ASSERT_FRAME("BROKER: Frame[channel=1; QueueDeclareOk: queue=MyQueue; messageCount=0; consumerCount=0]", *i++);
+ ASSERT_FRAME("CLIENT: Frame[channel=1; QueueBind: ticket=0; queue=MyQueue; exchange=MyExchange; routingKey=MyTopic; arguments={}]", *i++);
+ ASSERT_FRAME("CLIENT: Frame[channel=1; ExecutionFlush: ]", *i++);
+ ASSERT_FRAME("BROKER: Frame[channel=1; ExecutionComplete: cumulativeExecutionMark=4; rangedExecutionSet={}]", *i++);
}
};
Modified: incubator/qpid/trunk/qpid/python/cpp_failing_0-10.txt
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/cpp_failing_0-10.txt?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/python/cpp_failing_0-10.txt (original)
+++ incubator/qpid/trunk/qpid/python/cpp_failing_0-10.txt Fri Jul 27 08:44:52 2007
@@ -1,4 +1,15 @@
-tests_0-10.message.MessageTests.test_checkpoint
tests_0-10.message.MessageTests.test_reject
tests_0-10.basic.BasicTests.test_get
+
+tests_0-10.message.MessageTests.test_get
+tests_0-10.message.MessageTests.test_checkpoint
+tests_0-10.message.MessageTests.test_empty_reference
+tests_0-10.message.MessageTests.test_reference_already_opened_error
+tests_0-10.message.MessageTests.test_reference_completion
+tests_0-10.message.MessageTests.test_reference_large
+tests_0-10.message.MessageTests.test_reference_multi_transfer
+tests_0-10.message.MessageTests.test_reference_simple
+tests_0-10.message.MessageTests.test_reference_unopened_on_append_error
+tests_0-10.message.MessageTests.test_reference_unopened_on_close_error
+tests_0-10.message.MessageTests.test_reference_unopened_on_transfer_error
Modified: incubator/qpid/trunk/qpid/python/qpid/codec.py
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/qpid/codec.py?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/python/qpid/codec.py (original)
+++ incubator/qpid/trunk/qpid/python/qpid/codec.py Fri Jul 27 08:44:52 2007
@@ -329,12 +329,6 @@
return ReferenceId(self.decode_longstr())
# new domains for 0-10:
-
- def encode_uuid(self, s):
- self.encode_longstr(s)
-
- def decode_uuid(self):
- return self.decode_longstr()
def encode_rfc1982_long(self, s):
self.encode_long(s)
@@ -342,10 +336,21 @@
def decode_rfc1982_long(self):
return self.decode_long()
- #Not done yet
def encode_rfc1982_long_set(self, s):
- self.encode_short(0)
+ self.encode_short(len(s))
+ for i in s:
+ self.encode_long(i)
def decode_rfc1982_long_set(self):
- self.decode_short()
- return 0;
+ count = self.decode_short()
+ set = []
+ for i in range(0, count):
+ set.append(self.decode_long())
+ return set;
+
+ #not correct for 0-10 yet
+ def encode_uuid(self, s):
+ self.encode_longstr(s)
+
+ def decode_uuid(self):
+ return self.decode_longstr()
Modified: incubator/qpid/trunk/qpid/python/qpid/message.py
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/qpid/message.py?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/python/qpid/message.py (original)
+++ incubator/qpid/trunk/qpid/python/qpid/message.py Fri Jul 27 08:44:52 2007
@@ -26,7 +26,10 @@
self.frame = frame
self.method = frame.method_type
self.content = content
-
+ if self.method.klass.name != "execution":
+ self.command_id = self.channel.incoming_completion.sequence.next()
+ #print "allocated: ", self.command_id, "to ", self.method.klass.name, "_", self.method.name
+
def __len__(self):
return len(self.frame.args)
@@ -66,3 +69,6 @@
def __repr__(self):
return Message.REPR % (self.method, self.frame.args, self.content)
+
+ def complete(self, cumulative=True):
+ self.channel.incoming_completion.complete(mark=self.command_id, cumulative=cumulative)
Modified: incubator/qpid/trunk/qpid/python/qpid/peer.py
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/qpid/peer.py?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/python/qpid/peer.py (original)
+++ incubator/qpid/trunk/qpid/python/qpid/peer.py Fri Jul 27 08:44:52 2007
@@ -30,6 +30,7 @@
from queue import Queue, Closed as QueueClosed
from content import Content
from cStringIO import StringIO
+from time import time
class Sequence:
@@ -186,11 +187,11 @@
self.requester = Requester(self.write)
self.responder = Responder(self.write)
- self.completion = ExecutionCompletion()
+ self.completion = OutgoingCompletion()
+ self.incoming_completion = IncomingCompletion(self)
# Use reliable framing if version == 0-9.
- # (also for 0-10 while transitioning...)
- self.reliable = (spec.major == 0 and (spec.minor == 9 or spec.minor == 10))
+ self.reliable = (spec.major == 0 and spec.minor == 9)
self.use_execution_layer = (spec.major == 0 and spec.minor == 10)
self.synchronous = True
@@ -202,6 +203,7 @@
self.incoming.close()
self.responses.close()
self.completion.close()
+ self.incoming_completion.reset()
def write(self, frame, content = None):
if self.closed:
@@ -252,6 +254,9 @@
self.responder.respond(method, batch, request)
def invoke(self, type, args, kwargs):
+ if type.klass.name == "channel" and (type.name == "close" or type.name == "open"):
+ self.completion.reset()
+ self.incoming_completion.reset()
self.completion.next_command(type)
content = kwargs.pop("content", None)
frame = Method(type, type.arguments(*args, **kwargs))
@@ -306,6 +311,13 @@
return Message(self, resp, content)
else:
raise ValueError(resp)
+ elif self.synchronous and not frame.method.response \
+ and self.use_execution_layer and frame.method.klass.name != "execution":
+ self.execution_flush()
+ self.completion.wait()
+ if self.closed:
+ raise Closed(self.reason)
+
except QueueClosed, e:
if self.closed:
raise Closed(self.reason)
@@ -349,21 +361,32 @@
def is_complete(self):
return self.completed.isSet()
-class ExecutionCompletion:
+class OutgoingCompletion:
+ """
+ Manages completion of outgoing commands i.e. command sent by this peer
+ """
+
def __init__(self):
self.condition = threading.Condition()
- self.sequence = Sequence(1)
- self.command_id = 0
- self.mark = 0
+
+ self.sequence = Sequence(1) #issues ids for outgoing commands
+ self.command_id = 0 #last issued id
+ self.mark = 0 #commands up to this mark are known to be complete
+ self.closed = False
def next_command(self, method):
#the following test is a hack until the track/sub-channel is available
if method.klass.name != "execution":
self.command_id = self.sequence.next()
+ def reset(self):
+ self.sequence = Sequence(1) #reset counter
+
def close(self):
+ self.reset()
self.condition.acquire()
try:
+ self.closed = True
self.condition.notifyAll()
finally:
self.condition.release()
@@ -378,11 +401,50 @@
def wait(self, point_of_interest=-1, timeout=None):
if point_of_interest == -1: point_of_interest = self.command_id
+ start_time = time()
+ remaining = timeout
self.condition.acquire()
try:
- if point_of_interest > self.mark:
- self.condition.wait(timeout)
+ while not self.closed and point_of_interest > self.mark:
+ #print "waiting for ", point_of_interest, " mark is currently at ", self.mark
+ self.condition.wait(remaining)
+ if timeout:
+ if start_time + timeout > time(): break
+ else: remaining = timeout - (time() - start_time)
finally:
self.condition.release()
- #todo: retry until timed out or closed
return point_of_interest <= self.mark
+
+class IncomingCompletion:
+ """
+ Manages completion of incoming commands i.e. command received by this peer
+ """
+
+ def __init__(self, channel):
+ self.sequence = Sequence(1) #issues ids for incoming commands
+ self.mark = 0 #id of last command of whose completion notification was sent to the other peer
+ self.channel = channel
+
+ def next_id(self, method):
+ #the following test is a hack until the track/sub-channel is available
+ if method.klass.name != "execution":
+ return self.sequence.next()
+ else:
+ return 0
+
+ def reset(self):
+ self.sequence = Sequence(1) #reset counter
+
+ def complete(self, mark, cumulative=True):
+ if cumulative:
+ if mark > self.mark:
+ self.mark = mark
+ self.channel.execution_complete(cumulative_execution_mark=self.mark)
+ else:
+ #TODO: record and manage the ranges properly
+ range = [mark, mark]
+ self.channel.execution_complete(cumulative_execution_mark=self.mark, ranged_execution_set=range)
+
+
+
+
Modified: incubator/qpid/trunk/qpid/python/qpid/spec.py
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/qpid/spec.py?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/python/qpid/spec.py (original)
+++ incubator/qpid/trunk/qpid/python/qpid/spec.py Fri Jul 27 08:44:52 2007
@@ -240,7 +240,7 @@
"content": None,
"uuid": "",
"rfc1982_long": 0,
- "rfc1982_long_set": 0
+ "rfc1982_long_set": []
}
def define_method(self, name):
Modified: incubator/qpid/trunk/qpid/python/tests_0-10/broker.py
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/tests_0-10/broker.py?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/python/tests_0-10/broker.py (original)
+++ incubator/qpid/trunk/qpid/python/tests_0-10/broker.py Fri Jul 27 08:44:52 2007
@@ -48,7 +48,7 @@
body = "test ack"
ch.message_transfer(routing_key = "otherqueue", body = body)
msg = self.client.queue(ctag).get(timeout = 5)
- msg.ok()
+ msg.complete()
self.assert_(msg.body == body)
def test_simple_delivery_immediate(self):
Modified: incubator/qpid/trunk/qpid/python/tests_0-10/dtx.py
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/tests_0-10/dtx.py?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/python/tests_0-10/dtx.py (original)
+++ incubator/qpid/trunk/qpid/python/tests_0-10/dtx.py Fri Jul 27 08:44:52 2007
@@ -40,6 +40,11 @@
XA_RBROLLBACK = 1
XA_RBTIMEOUT = 2
XA_OK = 8
+ tx_counter = 0
+
+ def reset_channel(self):
+ self.channel.channel_close()
+ self.channel.channel_open()
def test_simple_commit(self):
"""
@@ -56,6 +61,9 @@
#commit
self.assertEqual(self.XA_OK, channel.dtx_coordination_commit(xid=tx, one_phase=True).status)
+ #should close and reopen channel to ensure no unacked messages are held
+ self.reset_channel()
+
#check result
self.assertMessageCount(0, "queue-a")
self.assertMessageCount(1, "queue-b")
@@ -79,6 +87,8 @@
#commit
self.assertEqual(self.XA_OK, channel.dtx_coordination_commit(xid=tx, one_phase=False).status)
+ self.reset_channel()
+
#check result
self.assertMessageCount(0, "queue-a")
self.assertMessageCount(1, "queue-b")
@@ -100,6 +110,8 @@
#rollback
self.assertEqual(self.XA_OK, channel.dtx_coordination_rollback(xid=tx).status)
+ self.reset_channel()
+
#check result
self.assertMessageCount(1, "queue-a")
self.assertMessageCount(0, "queue-b")
@@ -123,6 +135,8 @@
#rollback
self.assertEqual(self.XA_OK, channel.dtx_coordination_rollback(xid=tx).status)
+ self.reset_channel()
+
#check result
self.assertMessageCount(1, "queue-a")
self.assertMessageCount(0, "queue-b")
@@ -191,6 +205,8 @@
channel = self.channel
#do some transactional work & complete the transaction
self.test_simple_commit()
+ # channel has been reset, so reselect for use with dtx
+ channel.dtx_demarcation_select()
#start association for the same xid as the previously completed txn
tx = self.xid("my-xid")
@@ -355,7 +371,7 @@
self.assertEqual("two", msg.message_id)
channel.message_cancel(destination="results")
#ack the message then close the channel
- msg.ok()
+ msg.complete()
channel.channel_close()
channel = self.channel
@@ -446,7 +462,7 @@
channel2.dtx_demarcation_select()
channel2.dtx_demarcation_start(xid=tx)
channel2.message_get(queue="dummy", destination="dummy")
- self.client.queue("dummy").get(timeout=1).ok()
+ self.client.queue("dummy").get(timeout=1).complete()
channel2.message_transfer(routing_key="dummy", body="whatever")
channel2.channel_close()
@@ -548,7 +564,9 @@
channel.dtx_coordination_rollback(xid=x)
self.fail("Recovered xids not as expected. missing: %s; extra: %s" % (missing, extra))
- def xid(self, txid, branchqual = ''):
+ def xid(self, txid):
+ DtxTests.tx_counter += 1
+ branchqual = "v%s" % DtxTests.tx_counter
return pack('LBB', 0, len(txid), len(branchqual)) + txid + branchqual
def txswap(self, tx, id):
@@ -573,7 +591,7 @@
#consume from src:
channel.message_get(destination="temp-swap", queue=src)
msg = self.client.queue("temp-swap").get(timeout=1)
- msg.ok();
+ msg.complete();
#re-publish to dest
channel.message_transfer(routing_key=dest, message_id=msg.message_id, body=msg.body)
Modified: incubator/qpid/trunk/qpid/python/tests_0-10/example.py
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/tests_0-10/example.py?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/python/tests_0-10/example.py (original)
+++ incubator/qpid/trunk/qpid/python/tests_0-10/example.py Fri Jul 27 08:44:52 2007
@@ -90,5 +90,5 @@
self.assertEqual(body, msg.body)
# Now acknowledge the message.
- msg.ok()
+ msg.complete()
Modified: incubator/qpid/trunk/qpid/python/tests_0-10/message.py
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/tests_0-10/message.py?view=diff&rev=560285&r1=560284&r2=560285
==============================================================================
--- incubator/qpid/trunk/qpid/python/tests_0-10/message.py (original)
+++ incubator/qpid/trunk/qpid/python/tests_0-10/message.py Fri Jul 27 08:44:52 2007
@@ -171,8 +171,8 @@
self.assertEqual("Four", msg4.body)
self.assertEqual("Five", msg5.body)
- msg1.ok(batchoffset=1)#One and Two
- msg4.ok()
+ msg2.complete(cumulative=True)#One and Two
+ msg4.complete(cumulative=False)
channel.message_recover(requeue=False)
@@ -215,8 +215,8 @@
self.assertEqual("Four", msg4.body)
self.assertEqual("Five", msg5.body)
- msg1.ok(batchoffset=1) #One and Two
- msg4.ok() #Four
+ msg2.complete(cumulative=True) #One and Two
+ msg4.complete(cumulative=False) #Four
channel.message_cancel(destination="consumer_tag")
@@ -276,14 +276,13 @@
except Empty: None
#ack messages and check that the next set arrive ok:
- #todo: once batching is implmented, send a single response for all messages
- msg.ok(batchoffset=-4)#1-5
+ msg.complete()
for i in range(6, 11):
msg = queue.get(timeout=1)
self.assertEqual("Message %d" % i, msg.body)
- msg.ok(batchoffset=-4)#6-10
+ msg.complete()
try:
extra = queue.get(timeout=1)
@@ -320,13 +319,13 @@
except Empty: None
#ack messages and check that the next set arrive ok:
- msg.ok(batchoffset=-4)#1-5
+ msg.complete()
for i in range(6, 11):
msg = queue.get(timeout=1)
self.assertEqual("Message %d" % i, msg.body)
- msg.ok(batchoffset=-4)#6-10
+ msg.complete()
try:
extra = queue.get(timeout=1)
@@ -376,9 +375,9 @@
self.assertEqual("Message %d" % i, msg.body)
if (i==13):
- msg.ok(batchoffset=-2)#11, 12 & 13
+ msg.complete()#11, 12 & 13
if(i in [15, 17, 19]):
- msg.ok()
+ msg.complete(cumulative=False)
reply = channel.message_get(no_ack=True, queue="test-get")
self.assertEqual(reply.method.klass.name, "message")
@@ -395,8 +394,7 @@
self.assertEqual(reply.method.name, "ok")
msg = self.client.queue(tag).get(timeout=1)
self.assertEqual("Message %d" % i, msg.body)
- msg.ok()
- #channel.message_ack(delivery_tag=reply.delivery_tag)
+ msg.complete()
reply = channel.message_get(no_ack=True, queue="test-get")
self.assertEqual(reply.method.klass.name, "message")