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/19 10:27:42 UTC
svn commit: r557522 [1/2] - in /incubator/qpid/trunk/qpid/cpp:
gentools/src/org/apache/qpid/gentools/ perftest/ src/ src/qpid/broker/
src/qpid/framing/ src/tests/
Author: gsim
Date: Thu Jul 19 01:27:36 2007
New Revision: 557522
URL: http://svn.apache.org/viewvc?view=rev&rev=557522
Log:
removed the need to pass MethodContext/RequestId through proxy and handler/adapter interfaces
Added:
incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumber.cpp (with props)
incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumber.h (with props)
incubator/qpid/trunk/qpid/cpp/src/tests/SequenceNumberTest.cpp (with props)
Modified:
incubator/qpid/trunk/qpid/cpp/gentools/src/org/apache/qpid/gentools/AmqpClass.java
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/src/org/apache/qpid/gentools/Utils.java
incubator/qpid/trunk/qpid/cpp/perftest/ (props changed)
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/BrokerAdapter.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/ConnectionAdapter.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionAdapter.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DtxHandlerImpl.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DtxHandlerImpl.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/GetAdapter.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageHandlerImpl.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageHandlerImpl.h
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/framing/FramingContent.h
incubator/qpid/trunk/qpid/cpp/src/qpid/framing/MethodContext.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/framing/Proxy.h
incubator/qpid/trunk/qpid/cpp/src/tests/BrokerChannelTest.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/ExchangeTest.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/Makefile.am
incubator/qpid/trunk/qpid/cpp/src/tests/MessageBuilderTest.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/MessageTest.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/TxAckTest.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/TxPublishTest.cpp
Modified: incubator/qpid/trunk/qpid/cpp/gentools/src/org/apache/qpid/gentools/AmqpClass.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/gentools/src/org/apache/qpid/gentools/AmqpClass.java?view=diff&rev=557522&r1=557521&r2=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/gentools/src/org/apache/qpid/gentools/AmqpClass.java (original)
+++ incubator/qpid/trunk/qpid/cpp/gentools/src/org/apache/qpid/gentools/AmqpClass.java Thu Jul 19 01:27:36 2007
@@ -93,6 +93,9 @@
thisMethod = new AmqpMethod(methodName, converter);
methodMap.put(methodName, thisMethod);
}
+ int content = Utils.getNamedIntegerAttribute(child, Utils.ATTRIBUTE_CONTENT, 0);
+ thisMethod.content = (content == 1);
+
if (!thisMethod.addFromNode(child, 0, version))
{
String className = converter.prepareClassName(Utils.getNamedAttribute(classNode,
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=557522&r1=557521&r2=557522
==============================================================================
--- 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 Thu Jul 19 01:27:36 2007
@@ -36,6 +36,7 @@
public AmqpFlagMap clientMethodFlagMap; // Method called on client (<chassis name="server"> in XML)
public AmqpFlagMap serverMethodFlagMap; // Method called on server (<chassis name="client"> in XML)
public AmqpFlagMap isResponseFlagMap;
+ public boolean content;
public AmqpMethod(String name, LanguageConverter converter)
{
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=557522&r1=557521&r2=557522
==============================================================================
--- 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 Thu Jul 19 01:27:36 2007
@@ -763,14 +763,23 @@
AmqpVersionSet versionSet = overloadededParameterMap.get(thisFieldMap);
if (!first) sb.append(cr);
sb.append(indent + "virtual "+returnType+" "+ methodName + "(");
- 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 && isSpecialCase(thisClass.name, method.name)) {
+ sb.append("const MethodContext& context");
+ } else {
+ 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");
@@ -888,6 +897,7 @@
sb.append(indent + "{" + cr);
sb.append(indent + "private:" + cr);
sb.append(indent + tab + "ChannelAdapter& channel;" + cr);
+ sb.append(indent + tab + "RequestId responseTo;" + cr);
sb.append(cr);
sb.append(indent + "public:" + cr);
sb.append(indent + tab + "// Constructors and destructors" + cr);
@@ -897,6 +907,9 @@
sb.append(indent + tab + "virtual ~" + className + "() {}" + cr);
sb.append(cr);
sb.append(indent + tab + "static "+className+"& get(" + proxyOuterClassName(serverFlag)+"& proxy) { return proxy.get"+className+"();}\n\n");
+ sb.append(indent + tab + "// set for response correlation" + cr);
+ sb.append(indent + tab + "void setResponseTo(RequestId r) { responseTo = r; }" + cr);
+ sb.append(cr);
sb.append(indent + tab + "// Protocol methods" + cr);
sb.append(cr);
sb.append(generateInnerClassMethods(thisClass, serverFlag, false, indentSize + tabSize, tabSize));
@@ -1004,8 +1017,10 @@
methodName + "(");
sb.append(generateMethodParameterList(thisFieldMap, indentSize + (5*tabSize), false, true, true));
if (method.isResponse(null)) {
+ /*
if (!thisFieldMap.isEmpty()) sb.append(", ");
- sb.append("RequestId responseTo");
+ sb.append("RequestId responseTo");
+ */
}
sb.append(")");
if (versionSet.size() != globalVersionSet.size())
@@ -1457,6 +1472,7 @@
String indent = Utils.createSpaces(indentSize);
String tab = Utils.createSpaces(tabSize);
StringBuffer sb = new StringBuffer();
+ boolean special = isSpecialCase(thisClass.name, method.name);
if (method.serverMethodFlagMap.size() > 0) // At least one AMQP version defines this method as a server method
{
@@ -1466,14 +1482,18 @@
if (bItr.next()) // This is a server operation
{
boolean fieldMapNotEmptyFlag = method.fieldMap.size() > 0;
- sb.append(indent + "void invoke(AMQP_ServerOperations& target, const MethodContext& context)" + cr);
+ sb.append(indent + "void invoke(AMQP_ServerOperations& target, const MethodContext&"
+ + (special ? " context)" : " )") + cr);
sb.append(indent + "{" + cr);
sb.append(indent + tab + "target.get" + thisClass.name + "Handler()->" +
parseForReservedWords(Utils.firstLower(method.name),
- thisClass.name + Utils.firstUpper(method.name) + "Body.invoke()") + "(context");
- if (fieldMapNotEmptyFlag)
+ thisClass.name + Utils.firstUpper(method.name) + "Body.invoke()") + "(");
+ if (special)
+ {
+ sb.append("context");
+ }
+ else if (fieldMapNotEmptyFlag)
{
- sb.append("," + cr);
sb.append(generateFieldList(method.fieldMap, version, false, false, indentSize + 4*tabSize));
sb.append(indent + tab + tab + tab + tab);
}
@@ -1631,5 +1651,10 @@
ccn.append(b);
}
return ccn.toString();
+ }
+
+ private boolean isSpecialCase(String className, String methodName)
+ {
+ return "message".equalsIgnoreCase(className) && ("transfer".equalsIgnoreCase(methodName) || "append".equalsIgnoreCase(methodName));
}
}
Modified: incubator/qpid/trunk/qpid/cpp/gentools/src/org/apache/qpid/gentools/Utils.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/gentools/src/org/apache/qpid/gentools/Utils.java?view=diff&rev=557522&r1=557521&r2=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/gentools/src/org/apache/qpid/gentools/Utils.java (original)
+++ incubator/qpid/trunk/qpid/cpp/gentools/src/org/apache/qpid/gentools/Utils.java Thu Jul 19 01:27:36 2007
@@ -70,6 +70,13 @@
{
return Integer.parseInt(getNamedAttribute(n, attrName));
}
+
+ public static int getNamedIntegerAttribute(Node node, String name, int defaultValue) throws AmqpParseException
+ {
+ NamedNodeMap attributes = node.getAttributes();
+ Attr a = attributes == null ? null : (Attr) attributes.getNamedItem(name);
+ return a == null ? defaultValue : Integer.parseInt(a.getNodeValue());
+ }
// Element functions
Propchange: incubator/qpid/trunk/qpid/cpp/perftest/
------------------------------------------------------------------------------
--- svn:ignore (original)
+++ svn:ignore Thu Jul 19 01:27:36 2007
@@ -1,2 +1,4 @@
Makefile.in
Makefile
+topic_listener
+topic_publisher
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=557522&r1=557521&r2=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/Makefile.am (original)
+++ incubator/qpid/trunk/qpid/cpp/src/Makefile.am Thu Jul 19 01:27:36 2007
@@ -151,6 +151,7 @@
qpid/framing/ProtocolVersionException.cpp \
qpid/framing/Requester.cpp \
qpid/framing/Responder.cpp \
+ qpid/framing/SequenceNumber.cpp \
qpid/framing/Correlator.cpp \
qpid/framing/Value.cpp \
qpid/framing/Proxy.cpp \
@@ -360,6 +361,7 @@
qpid/framing/Requester.h \
qpid/framing/Responder.h \
qpid/framing/SerializeHandler.h \
+ qpid/framing/SequenceNumber.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=557522&r1=557521&r2=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.cpp Thu Jul 19 01:27:36 2007
@@ -53,36 +53,33 @@
return connection.getVersion();
}
-void BrokerAdapter::ChannelHandlerImpl::open(
- const MethodContext& context, const string& /*outOfBand*/){
+void BrokerAdapter::ChannelHandlerImpl::open(const string& /*outOfBand*/){
channel.open();
// FIXME aconway 2007-01-04: provide valid ID as per ampq 0-9
- client.openOk(
- std::string()/* ID */, context.getRequestId());
+ client.openOk(std::string()/* ID */);//GRS, context.getRequestId());
}
-void BrokerAdapter::ChannelHandlerImpl::flow(const MethodContext& context, bool active){
+void BrokerAdapter::ChannelHandlerImpl::flow(bool active){
channel.flow(active);
- client.flowOk(active, context.getRequestId());
+ client.flowOk(active);//GRS, context.getRequestId());
}
-void BrokerAdapter::ChannelHandlerImpl::flowOk(const MethodContext&, bool /*active*/){}
+void BrokerAdapter::ChannelHandlerImpl::flowOk(bool /*active*/){}
-void BrokerAdapter::ChannelHandlerImpl::close(
- const MethodContext& context, uint16_t /*replyCode*/,
+void BrokerAdapter::ChannelHandlerImpl::close(uint16_t /*replyCode*/,
const string& /*replyText*/,
uint16_t /*classId*/, uint16_t /*methodId*/)
{
- client.closeOk(context.getRequestId());
+ client.closeOk();//GRS context.getRequestId());
// FIXME aconway 2007-01-18: Following line will "delete this". Ugly.
connection.closeChannel(channel.getId());
}
-void BrokerAdapter::ChannelHandlerImpl::closeOk(const MethodContext&){}
+void BrokerAdapter::ChannelHandlerImpl::closeOk(){}
-void BrokerAdapter::ExchangeHandlerImpl::declare(const MethodContext& context, uint16_t /*ticket*/, const string& exchange, const string& type,
+void BrokerAdapter::ExchangeHandlerImpl::declare(uint16_t /*ticket*/, const string& exchange, const string& type,
bool passive, bool durable, bool /*autoDelete*/, bool /*internal*/, bool nowait,
const FieldTable& args){
@@ -107,31 +104,30 @@
}
}
if(!nowait){
- client.declareOk(context.getRequestId());
+ client.declareOk();//GRS context.getRequestId());
}
}
-void BrokerAdapter::ExchangeHandlerImpl::delete_(const MethodContext& context, uint16_t /*ticket*/,
+void BrokerAdapter::ExchangeHandlerImpl::delete_(uint16_t /*ticket*/,
const string& name, bool /*ifUnused*/, bool nowait){
//TODO: implement unused
Exchange::shared_ptr exchange(broker.getExchanges().get(name));
if (exchange->isDurable()) broker.getStore().destroy(*exchange);
broker.getExchanges().destroy(name);
- if(!nowait) client.deleteOk(context.getRequestId());
+ if(!nowait) client.deleteOk();//GRS context.getRequestId());
}
-void BrokerAdapter::ExchangeHandlerImpl::query(const MethodContext& context, u_int16_t /*ticket*/, const string& name)
+void BrokerAdapter::ExchangeHandlerImpl::query(u_int16_t /*ticket*/, const string& name)
{
try {
Exchange::shared_ptr exchange(broker.getExchanges().get(name));
- client.queryOk(exchange->getType(), exchange->isDurable(), false, exchange->getArgs(), context.getRequestId());
+ client.queryOk(exchange->getType(), exchange->isDurable(), false, exchange->getArgs());//GRS, context.getRequestId());
} catch (const ChannelException& e) {
- client.queryOk("", false, true, FieldTable(), context.getRequestId());
+ client.queryOk("", false, true, FieldTable());//GRS, context.getRequestId());
}
}
-void BrokerAdapter::BindingHandlerImpl::query(const framing::MethodContext& context,
- u_int16_t /*ticket*/,
+void BrokerAdapter::BindingHandlerImpl::query(u_int16_t /*ticket*/,
const std::string& exchangeName,
const std::string& queueName,
const std::string& key,
@@ -148,22 +144,22 @@
}
if (!exchange) {
- client.queryOk(true, false, false, false, false, context.getRequestId());
+ client.queryOk(true, false, false, false, false);//GRS, context.getRequestId());
} else if (!queueName.empty() && !queue) {
- client.queryOk(false, true, false, false, false, context.getRequestId());
+ client.queryOk(false, true, false, false, false);//GRS, context.getRequestId());
} else if (exchange->isBound(queue, key.empty() ? 0 : &key, args.count() > 0 ? &args : &args)) {
- client.queryOk(false, false, false, false, false, context.getRequestId());
+ client.queryOk(false, false, false, false, false);//GRS, context.getRequestId());
} else {
//need to test each specified option individually
bool queueMatched = queueName.empty() || exchange->isBound(queue, 0, 0);
bool keyMatched = key.empty() || exchange->isBound(Queue::shared_ptr(), &key, 0);
bool argsMatched = args.count() == 0 || exchange->isBound(Queue::shared_ptr(), 0, &args);
- client.queryOk(false, false, !queueMatched, !keyMatched, !argsMatched, context.getRequestId());
+ client.queryOk(false, false, !queueMatched, !keyMatched, !argsMatched);//GRS, context.getRequestId());
}
}
-void BrokerAdapter::QueueHandlerImpl::declare(const MethodContext& context, uint16_t /*ticket*/, const string& name,
+void BrokerAdapter::QueueHandlerImpl::declare(uint16_t /*ticket*/, const string& name,
bool passive, bool durable, bool exclusive,
bool autoDelete, bool nowait, const qpid::framing::FieldTable& arguments){
Queue::shared_ptr queue;
@@ -200,12 +196,11 @@
if (!nowait) {
string queueName = queue->getName();
client.declareOk(
- queueName, queue->getMessageCount(), queue->getConsumerCount(),
- context.getRequestId());
+ queueName, queue->getMessageCount(), queue->getConsumerCount());//GRS, context.getRequestId());
}
}
-void BrokerAdapter::QueueHandlerImpl::bind(const MethodContext& context, uint16_t /*ticket*/, const string& queueName,
+void BrokerAdapter::QueueHandlerImpl::bind(uint16_t /*ticket*/, const string& queueName,
const string& exchangeName, const string& routingKey, bool nowait,
const FieldTable& arguments){
@@ -219,7 +214,7 @@
broker.getStore().bind(*exchange, *queue, routingKey, arguments);
}
}
- if(!nowait) client.bindOk(context.getRequestId());
+ if(!nowait) client.bindOk();//GRS context.getRequestId());
}else{
throw ChannelException(
404, "Bind failed. No such exchange: " + exchangeName);
@@ -227,9 +222,7 @@
}
void
-BrokerAdapter::QueueHandlerImpl::unbind(
- const MethodContext& context,
- uint16_t /*ticket*/,
+BrokerAdapter::QueueHandlerImpl::unbind(uint16_t /*ticket*/,
const string& queueName,
const string& exchangeName,
const string& routingKey,
@@ -245,17 +238,17 @@
broker.getStore().unbind(*exchange, *queue, routingKey, arguments);
}
- client.unbindOk(context.getRequestId());
+ client.unbindOk();//GRS context.getRequestId());
}
-void BrokerAdapter::QueueHandlerImpl::purge(const MethodContext& context, uint16_t /*ticket*/, const string& queueName, bool nowait){
+void BrokerAdapter::QueueHandlerImpl::purge(uint16_t /*ticket*/, const string& queueName, bool nowait){
Queue::shared_ptr queue = getQueue(queueName);
int count = queue->purge();
- if(!nowait) client.purgeOk( count, context.getRequestId());
+ if(!nowait) client.purgeOk( count);//GRS, context.getRequestId());
}
-void BrokerAdapter::QueueHandlerImpl::delete_(const MethodContext& context, uint16_t /*ticket*/, const string& queue,
+void BrokerAdapter::QueueHandlerImpl::delete_(uint16_t /*ticket*/, const string& queue,
bool ifUnused, bool ifEmpty, bool nowait){
ChannelException error(0, "");
int count(0);
@@ -277,21 +270,20 @@
}
if(!nowait)
- client.deleteOk(count, context.getRequestId());
+ client.deleteOk(count);//GRS, context.getRequestId());
}
-void BrokerAdapter::BasicHandlerImpl::qos(const MethodContext& context, uint32_t prefetchSize, uint16_t prefetchCount, bool /*global*/){
+void BrokerAdapter::BasicHandlerImpl::qos(uint32_t prefetchSize, uint16_t prefetchCount, bool /*global*/){
//TODO: handle global
channel.setPrefetchSize(prefetchSize);
channel.setPrefetchCount(prefetchCount);
- client.qosOk(context.getRequestId());
+ client.qosOk();//GRS context.getRequestId());
}
-void BrokerAdapter::BasicHandlerImpl::consume(
- const MethodContext& context, uint16_t /*ticket*/,
+void BrokerAdapter::BasicHandlerImpl::consume(uint16_t /*ticket*/,
const string& queueName, const string& consumerTag,
bool noLocal, bool noAck, bool exclusive,
bool nowait, const FieldTable& fields)
@@ -308,29 +300,26 @@
channel.consume(std::auto_ptr<DeliveryAdapter>(new ConsumeAdapter(adapter, newTag, connection.getFrameMax())),
newTag, queue, !noAck, exclusive, noLocal ? &connection : 0, &fields);
- if(!nowait) client.consumeOk(newTag, context.getRequestId());
+ if(!nowait) client.consumeOk(newTag);//GRS, context.getRequestId());
//allow messages to be dispatched if required as there is now a consumer:
queue->requestDispatch();
}
-void BrokerAdapter::BasicHandlerImpl::cancel(const MethodContext& context, const string& consumerTag, bool nowait){
+void BrokerAdapter::BasicHandlerImpl::cancel(const string& consumerTag, bool nowait){
channel.cancel(consumerTag);
- if(!nowait) client.cancelOk(consumerTag, context.getRequestId());
+ if(!nowait) client.cancelOk(consumerTag);//GRS, context.getRequestId());
}
-void BrokerAdapter::BasicHandlerImpl::publish(
- const MethodContext& context, uint16_t /*ticket*/,
+void BrokerAdapter::BasicHandlerImpl::publish(uint16_t /*ticket*/,
const string& exchangeName, const string& routingKey,
bool mandatory, bool immediate)
{
Exchange::shared_ptr exchange = exchangeName.empty() ? broker.getExchanges().getDefault() : broker.getExchanges().get(exchangeName);
if(exchange){
- BasicMessage* msg = new BasicMessage(
- &connection, exchangeName, routingKey, mandatory, immediate,
- context.methodBody);
+ BasicMessage* msg = new BasicMessage(&connection, exchangeName, routingKey, mandatory, immediate);
channel.handlePublish(msg);
}else{
throw ChannelException(
@@ -338,45 +327,47 @@
}
}
-void BrokerAdapter::BasicHandlerImpl::get(const MethodContext& context, uint16_t /*ticket*/, const string& queueName, bool noAck){
+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)){
string clusterId;//not used, part of an imatix hack
- client.getEmpty(clusterId, context.getRequestId());
+ client.getEmpty(clusterId);//GRS, context.getRequestId());
}
}
-void BrokerAdapter::BasicHandlerImpl::ack(const MethodContext&, uint64_t deliveryTag, bool multiple){
+void BrokerAdapter::BasicHandlerImpl::ack(uint64_t deliveryTag, bool multiple){
channel.ack(deliveryTag, multiple);
}
-void BrokerAdapter::BasicHandlerImpl::reject(const MethodContext&, uint64_t /*deliveryTag*/, bool /*requeue*/){}
+void BrokerAdapter::BasicHandlerImpl::reject(uint64_t /*deliveryTag*/, bool /*requeue*/){}
-void BrokerAdapter::BasicHandlerImpl::recover(const MethodContext&, bool requeue){
+void BrokerAdapter::BasicHandlerImpl::recover(bool requeue)
+{
channel.recover(requeue);
}
-void BrokerAdapter::TxHandlerImpl::select(const MethodContext& context){
+void BrokerAdapter::TxHandlerImpl::select()
+{
channel.startTx();
- client.selectOk(context.getRequestId());
+ client.selectOk();//GRS context.getRequestId());
}
-void BrokerAdapter::TxHandlerImpl::commit(const MethodContext& context){
+void BrokerAdapter::TxHandlerImpl::commit()
+{
channel.commit();
- client.commitOk(context.getRequestId());
+ client.commitOk();//GRS context.getRequestId());
}
-void BrokerAdapter::TxHandlerImpl::rollback(const MethodContext& context){
-
+void BrokerAdapter::TxHandlerImpl::rollback()
+{
channel.rollback();
- client.rollbackOk(context.getRequestId());
+ client.rollbackOk();//GRS context.getRequestId());
channel.recover(false);
}
-void
-BrokerAdapter::ChannelHandlerImpl::ok( const MethodContext& )
+void BrokerAdapter::ChannelHandlerImpl::ok()
{
//no specific action required, generic response handling should be sufficient
}
@@ -385,27 +376,36 @@
//
// Message class method handlers
//
-void
-BrokerAdapter::ChannelHandlerImpl::ping( const MethodContext& context)
+void BrokerAdapter::ChannelHandlerImpl::ping()
{
- client.ok(context.getRequestId());
+ client.ok();//GRS context.getRequestId());
client.pong();
}
void
-BrokerAdapter::ChannelHandlerImpl::pong( const MethodContext& context)
+BrokerAdapter::ChannelHandlerImpl::pong()
{
- client.ok(context.getRequestId());
+ client.ok();//GRS context.getRequestId());
}
-void
-BrokerAdapter::ChannelHandlerImpl::resume(
- const MethodContext&,
- const string& /*channel*/ )
+void BrokerAdapter::ChannelHandlerImpl::resume(const string& /*channel*/)
{
assert(0); // FIXME aconway 2007-01-04: 0-9 feature
}
+
+void BrokerAdapter::setResponseTo(RequestId r)
+{
+ basicHandler.client.setResponseTo(r);
+ channelHandler.client.setResponseTo(r);
+ exchangeHandler.client.setResponseTo(r);
+ bindingHandler.client.setResponseTo(r);
+ messageHandler.client.setResponseTo(r);
+ queueHandler.client.setResponseTo(r);
+ txHandler.client.setResponseTo(r);
+ dtxHandler.setResponseTo(r);
+}
+
}} // namespace qpid::broker
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.h?view=diff&rev=557522&r1=557521&r2=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.h Thu Jul 19 01:27:36 2007
@@ -85,6 +85,7 @@
}
framing::AMQP_ClientProxy& getProxy() { return proxy; }
+ void setResponseTo(framing::RequestId r);
private:
@@ -95,16 +96,16 @@
public:
ChannelHandlerImpl(BrokerAdapter& parent) : HandlerImplType(parent) {}
- void open(const framing::MethodContext& context, const std::string& outOfBand);
- void flow(const framing::MethodContext& context, bool active);
- void flowOk(const framing::MethodContext& context, bool active);
- void ok( const framing::MethodContext& context );
- void ping( const framing::MethodContext& context );
- void pong( const framing::MethodContext& context );
- void resume( const framing::MethodContext& context, const std::string& channelId );
- void close(const framing::MethodContext& context, uint16_t replyCode, const
+ void open(const std::string& outOfBand);
+ void flow(bool active);
+ void flowOk(bool active);
+ void ok( );
+ void ping( );
+ void pong( );
+ void resume( const std::string& channelId );
+ void close(uint16_t replyCode, const
std::string& replyText, uint16_t classId, uint16_t methodId);
- void closeOk(const framing::MethodContext& context);
+ void closeOk();
};
class ExchangeHandlerImpl :
@@ -114,16 +115,14 @@
public:
ExchangeHandlerImpl(BrokerAdapter& parent) : HandlerImplType(parent) {}
- void declare(const framing::MethodContext& context, uint16_t ticket,
+ void declare(uint16_t ticket,
const std::string& exchange, const std::string& type,
bool passive, bool durable, bool autoDelete,
bool internal, bool nowait,
const qpid::framing::FieldTable& arguments);
- void delete_(const framing::MethodContext& context, uint16_t ticket,
+ void delete_(uint16_t ticket,
const std::string& exchange, bool ifUnused, bool nowait);
- void query(const framing::MethodContext& context,
- u_int16_t ticket,
- const string& name);
+ void query(u_int16_t ticket, const string& name);
};
class BindingHandlerImpl :
@@ -133,8 +132,7 @@
public:
BindingHandlerImpl(BrokerAdapter& parent) : HandlerImplType(parent) {}
- void query(const framing::MethodContext& context,
- u_int16_t ticket,
+ void query(u_int16_t ticket,
const std::string& exchange,
const std::string& queue,
const std::string& routingKey,
@@ -148,22 +146,21 @@
public:
QueueHandlerImpl(BrokerAdapter& parent) : HandlerImplType(parent) {}
- void declare(const framing::MethodContext& context, uint16_t ticket, const std::string& queue,
+ void declare(uint16_t ticket, const std::string& queue,
bool passive, bool durable, bool exclusive,
bool autoDelete, bool nowait,
const qpid::framing::FieldTable& arguments);
- void bind(const framing::MethodContext& context, uint16_t ticket, const std::string& queue,
+ void bind(uint16_t ticket, const std::string& queue,
const std::string& exchange, const std::string& routingKey,
bool nowait, const qpid::framing::FieldTable& arguments);
- void unbind(const framing::MethodContext& context,
- uint16_t ticket,
+ void unbind(uint16_t ticket,
const std::string& queue,
const std::string& exchange,
const std::string& routingKey,
const qpid::framing::FieldTable& arguments );
- void purge(const framing::MethodContext& context, uint16_t ticket, const std::string& queue,
+ void purge(uint16_t ticket, const std::string& queue,
bool nowait);
- void delete_(const framing::MethodContext& context, uint16_t ticket, const std::string& queue,
+ void delete_(uint16_t ticket, const std::string& queue,
bool ifUnused, bool ifEmpty,
bool nowait);
};
@@ -177,23 +174,23 @@
public:
BasicHandlerImpl(BrokerAdapter& parent) : HandlerImplType(parent), tagGenerator("sgen") {}
- void qos(const framing::MethodContext& context, uint32_t prefetchSize,
+ void qos(uint32_t prefetchSize,
uint16_t prefetchCount, bool global);
void consume(
- const framing::MethodContext& context, uint16_t ticket, const std::string& queue,
+ uint16_t ticket, const std::string& queue,
const std::string& consumerTag, bool noLocal, bool noAck,
bool exclusive, bool nowait,
const qpid::framing::FieldTable& fields);
- void cancel(const framing::MethodContext& context, const std::string& consumerTag,
+ void cancel(const std::string& consumerTag,
bool nowait);
- void publish(const framing::MethodContext& context, uint16_t ticket,
+ void publish(uint16_t ticket,
const std::string& exchange, const std::string& routingKey,
bool mandatory, bool immediate);
- void get(const framing::MethodContext& context, uint16_t ticket, const std::string& queue,
+ void get(uint16_t ticket, const std::string& queue,
bool noAck);
- void ack(const framing::MethodContext& context, uint64_t deliveryTag, bool multiple);
- void reject(const framing::MethodContext& context, uint64_t deliveryTag, bool requeue);
- void recover(const framing::MethodContext& context, bool requeue);
+ void ack(uint64_t deliveryTag, bool multiple);
+ void reject(uint64_t deliveryTag, bool requeue);
+ void recover(bool requeue);
};
class TxHandlerImpl :
@@ -203,9 +200,9 @@
public:
TxHandlerImpl(BrokerAdapter& parent) : HandlerImplType(parent) {}
- void select(const framing::MethodContext& context);
- void commit(const framing::MethodContext& context);
- void rollback(const framing::MethodContext& context);
+ void select();
+ void commit();
+ void rollback();
};
Connection& connection;
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=557522&r1=557521&r2=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessage.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessage.cpp Thu Jul 19 01:27:36 2007
@@ -29,6 +29,7 @@
#include "qpid/log/Statement.h"
#include "qpid/framing/BasicDeliverBody.h"
#include "qpid/framing/BasicGetOkBody.h"
+#include "qpid/framing/BasicPublishBody.h"
#include "qpid/framing/AMQContentBody.h"
#include "qpid/framing/AMQHeaderBody.h"
#include "qpid/framing/AMQMethodBody.h"
@@ -44,10 +45,10 @@
BasicMessage::BasicMessage(
const ConnectionToken* const _publisher,
const string& _exchange, const string& _routingKey,
- bool _mandatory, bool _immediate, framing::AMQMethodBody::shared_ptr respondTo
+ bool _mandatory, bool _immediate
) :
Message(_publisher, _exchange, _routingKey, _mandatory,
- _immediate, respondTo),
+ _immediate, framing::AMQMethodBody::shared_ptr(new BasicPublishBody(ProtocolVersion(0,9)))),
size(0)
{}
@@ -84,19 +85,20 @@
sendContent(channel, framesize);
}
-void BasicMessage::sendGetOk(const MethodContext& context,
- const std::string& /*destination*/,
+void BasicMessage::sendGetOk(ChannelAdapter& channel,
+ const std::string& /*destination*/,
uint32_t messageCount,
+ uint64_t responseTo,
uint64_t deliveryTag,
uint32_t framesize)
{
- context.channel->send(
+ channel.send(
new BasicGetOkBody(
- context.channel->getVersion(),
- context.methodBody->getRequestId(),
+ channel.getVersion(),
+ responseTo,
deliveryTag, getRedelivered(), getExchange(),
getRoutingKey(), messageCount));
- sendContent(*context.channel, framesize);
+ sendContent(channel, framesize);
}
void BasicMessage::sendContent(
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=557522&r1=557521&r2=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessage.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessage.h Thu Jul 19 01:27:36 2007
@@ -63,8 +63,7 @@
BasicMessage(const ConnectionToken* const publisher,
const string& exchange, const string& routingKey,
- bool mandatory, bool immediate,
- boost::shared_ptr<framing::AMQMethodBody> respondTo);
+ bool mandatory, bool immediate);
BasicMessage();
~BasicMessage();
void setHeader(boost::shared_ptr<framing::AMQHeaderBody> header);
@@ -76,9 +75,10 @@
uint64_t deliveryTag,
uint32_t framesize);
- void sendGetOk(const framing::MethodContext&,
- const std::string& destination,
+ void sendGetOk(framing::ChannelAdapter& channel,
+ const std::string& destination,
uint32_t messageCount,
+ uint64_t responseTo,
uint64_t deliveryTag,
uint32_t framesize);
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=557522&r1=557521&r2=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessageBase.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessageBase.h Thu Jul 19 01:27:36 2007
@@ -85,7 +85,6 @@
const std::string& getExchange() const { return exchange; }
uint64_t getPersistenceId() const { return persistenceId; }
bool getRedelivered() const { return redelivered; }
- AMQMethodBodyPtr getRespondTo() const { return respondTo; }
void setRouting(const std::string& _exchange, const std::string& _routingKey)
{ exchange = _exchange; routingKey = _routingKey; }
@@ -102,9 +101,10 @@
/**
* Used to return a message in response to a get from a queue
*/
- virtual void sendGetOk(const framing::MethodContext& context,
+ virtual void sendGetOk(framing::ChannelAdapter& channel,
const std::string& destination,
uint32_t messageCount,
+ uint64_t responseTo,
uint64_t deliveryTag,
uint32_t framesize) = 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=557522&r1=557521&r2=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessageMessage.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessageMessage.cpp Thu Jul 19 01:27:36 2007
@@ -185,14 +185,14 @@
}
void MessageMessage::sendGetOk(
- const framing::MethodContext& context,
+ framing::ChannelAdapter& channel,
const std::string& destination,
uint32_t /*messageCount*/,
+ uint64_t /*responseTo*/,
uint64_t /*deliveryTag*/,
uint32_t framesize)
{
- framing::ChannelAdapter* channel = context.channel;
- transferMessage(*channel, destination, framesize);
+ transferMessage(channel, destination, framesize);
}
bool MessageMessage::isComplete()
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=557522&r1=557521&r2=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessageMessage.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerMessageMessage.h Thu Jul 19 01:27:36 2007
@@ -58,9 +58,10 @@
uint64_t deliveryTag,
uint32_t framesize);
- void sendGetOk(const framing::MethodContext& context,
- const std::string& destination,
+ void sendGetOk(framing::ChannelAdapter& channel,
+ const std::string& destination,
uint32_t messageCount,
+ uint64_t responseTo,
uint64_t deliveryTag,
uint32_t framesize);
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionAdapter.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionAdapter.cpp?view=diff&rev=557522&r1=557521&r2=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionAdapter.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionAdapter.cpp Thu Jul 19 01:27:36 2007
@@ -50,10 +50,14 @@
)
{
try{
+ handler->client.setResponseTo(context.getRequestId());
method->invoke(*this, context);
+ handler->client.setResponseTo(0);
}catch(ConnectionException& e){
+ handler->client.setResponseTo(0);
handler->client.close(e.code, e.toString(), method->amqpClassId(), method->amqpMethodId());
}catch(std::exception& e){
+ handler->client.setResponseTo(0);
handler->client.close(541/*internal error*/, e.what(), method->amqpClassId(), method->amqpMethodId());
}
}
@@ -82,43 +86,38 @@
proxy(a), client(proxy.getConnection()), connection(c) {}
-void Handler::startOk(
- const MethodContext&, const FieldTable& /*clientProperties*/,
+void Handler::startOk(const FieldTable& /*clientProperties*/,
const string& /*mechanism*/,
const string& /*response*/, const string& /*locale*/)
{
client.tune(framing::CHANNEL_MAX, connection.getFrameMax(), connection.getHeartbeat());
}
-void Handler::secureOk(
- const MethodContext&, const string& /*response*/){}
+void Handler::secureOk(const string& /*response*/){}
-void Handler::tuneOk(
- const MethodContext&, uint16_t /*channelmax*/,
+void Handler::tuneOk(uint16_t /*channelmax*/,
uint32_t framemax, uint16_t heartbeat)
{
connection.setFrameMax(framemax);
connection.setHeartbeat(heartbeat);
}
-void Handler::open(
- const MethodContext& context, const string& /*virtualHost*/,
+void Handler::open(const string& /*virtualHost*/,
const string& /*capabilities*/, bool /*insist*/)
{
string knownhosts;
client.openOk(
- knownhosts, context.getRequestId());
+ knownhosts);//GRS, context.getRequestId());
}
-void Handler::close(
- const MethodContext& context, uint16_t /*replyCode*/, const string& /*replyText*/,
+void Handler::close(uint16_t /*replyCode*/, const string& /*replyText*/,
uint16_t /*classId*/, uint16_t /*methodId*/)
{
- client.closeOk(context.getRequestId());
+ client.closeOk();//GRS context.getRequestId());
connection.getOutput().close();
}
-void Handler::closeOk(const MethodContext&){
+void Handler::closeOk(){
connection.getOutput().close();
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionAdapter.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionAdapter.h?view=diff&rev=557522&r1=557521&r2=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionAdapter.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionAdapter.h Thu Jul 19 01:27:36 2007
@@ -81,22 +81,16 @@
Connection& connection;
Handler(Connection& connection, ConnectionAdapter& adapter);
- void startOk(const framing::MethodContext& context,
- const qpid::framing::FieldTable& clientProperties,
+ void startOk(const qpid::framing::FieldTable& clientProperties,
const std::string& mechanism, const std::string& response,
const std::string& locale);
- void secureOk(const framing::MethodContext& context,
- const std::string& response);
- void tuneOk(const framing::MethodContext& context,
- uint16_t channelMax,
- uint32_t frameMax, uint16_t heartbeat);
- void open(const framing::MethodContext& context,
- const std::string& virtualHost,
- const std::string& capabilities, bool insist);
- void close(const framing::MethodContext& context, uint16_t replyCode,
- const std::string& replyText,
+ void secureOk(const std::string& response);
+ void tuneOk(uint16_t channelMax, uint32_t frameMax, uint16_t heartbeat);
+ void open(const std::string& virtualHost,
+ const std::string& capabilities, bool insist);
+ void close(uint16_t replyCode, const std::string& replyText,
uint16_t classId, uint16_t methodId);
- void closeOk(const framing::MethodContext& context);
+ void closeOk();
};
}}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DtxHandlerImpl.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DtxHandlerImpl.cpp?view=diff&rev=557522&r1=557521&r2=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DtxHandlerImpl.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DtxHandlerImpl.cpp Thu Jul 19 01:27:36 2007
@@ -49,14 +49,13 @@
// DtxDemarcationHandler:
-void DtxHandlerImpl::select(const MethodContext& context )
+void DtxHandlerImpl::select()
{
channel.selectDtx();
- dClient.selectOk(context.getRequestId());
+ dClient.selectOk();//GRS context.getRequestId());
}
-void DtxHandlerImpl::end(const MethodContext& context,
- u_int16_t /*ticket*/,
+void DtxHandlerImpl::end(u_int16_t /*ticket*/,
const string& xid,
bool fail,
bool suspend)
@@ -67,7 +66,7 @@
if (suspend) {
throw ConnectionException(503, "End and suspend cannot both be set.");
} else {
- dClient.endOk(XA_RBROLLBACK, context.getRequestId());
+ dClient.endOk(XA_RBROLLBACK);//GRS, context.getRequestId());
}
} else {
if (suspend) {
@@ -75,15 +74,14 @@
} else {
channel.endDtx(xid, false);
}
- dClient.endOk(XA_OK, context.getRequestId());
+ dClient.endOk(XA_OK);//GRS, context.getRequestId());
}
} catch (const DtxTimeoutException& e) {
- dClient.endOk(XA_RBTIMEOUT, context.getRequestId());
+ dClient.endOk(XA_RBTIMEOUT);//GRS, context.getRequestId());
}
}
-void DtxHandlerImpl::start(const MethodContext& context,
- u_int16_t /*ticket*/,
+void DtxHandlerImpl::start(u_int16_t /*ticket*/,
const string& xid,
bool join,
bool resume)
@@ -97,54 +95,50 @@
} else {
channel.startDtx(xid, broker.getDtxManager(), join);
}
- dClient.startOk(XA_OK, context.getRequestId());
+ dClient.startOk(XA_OK);//GRS, context.getRequestId());
} catch (const DtxTimeoutException& e) {
- dClient.startOk(XA_RBTIMEOUT, context.getRequestId());
+ dClient.startOk(XA_RBTIMEOUT);//GRS, context.getRequestId());
}
}
// DtxCoordinationHandler:
-void DtxHandlerImpl::prepare(const MethodContext& context,
- u_int16_t /*ticket*/,
+void DtxHandlerImpl::prepare(u_int16_t /*ticket*/,
const string& xid)
{
try {
bool ok = broker.getDtxManager().prepare(xid);
- cClient.prepareOk(ok ? XA_OK : XA_RBROLLBACK, context.getRequestId());
+ cClient.prepareOk(ok ? XA_OK : XA_RBROLLBACK);//GRS, context.getRequestId());
} catch (const DtxTimeoutException& e) {
- cClient.prepareOk(XA_RBTIMEOUT, context.getRequestId());
+ cClient.prepareOk(XA_RBTIMEOUT);//GRS, context.getRequestId());
}
}
-void DtxHandlerImpl::commit(const MethodContext& context,
- u_int16_t /*ticket*/,
+void DtxHandlerImpl::commit(u_int16_t /*ticket*/,
const string& xid,
bool onePhase)
{
try {
bool ok = broker.getDtxManager().commit(xid, onePhase);
- cClient.commitOk(ok ? XA_OK : XA_RBROLLBACK, context.getRequestId());
+ cClient.commitOk(ok ? XA_OK : XA_RBROLLBACK);//GRS, context.getRequestId());
} catch (const DtxTimeoutException& e) {
- cClient.commitOk(XA_RBTIMEOUT, context.getRequestId());
+ cClient.commitOk(XA_RBTIMEOUT);//GRS, context.getRequestId());
}
}
-void DtxHandlerImpl::rollback(const MethodContext& context,
- u_int16_t /*ticket*/,
+void DtxHandlerImpl::rollback(u_int16_t /*ticket*/,
const string& xid )
{
try {
broker.getDtxManager().rollback(xid);
- cClient.rollbackOk(XA_OK, context.getRequestId());
+ cClient.rollbackOk(XA_OK);//GRS, context.getRequestId());
} catch (const DtxTimeoutException& e) {
- cClient.rollbackOk(XA_RBTIMEOUT, context.getRequestId());
+ cClient.rollbackOk(XA_RBTIMEOUT);//GRS, context.getRequestId());
}
}
-void DtxHandlerImpl::recover(const MethodContext& context,
- u_int16_t /*ticket*/,
+void DtxHandlerImpl::recover(u_int16_t /*ticket*/,
bool /*startscan*/,
u_int32_t /*endscan*/ )
{
@@ -177,33 +171,34 @@
FieldTable response;
response.setString("xids", data);
- cClient.recoverOk(response, context.getRequestId());
+ cClient.recoverOk(response);//GRS, context.getRequestId());
}
-void DtxHandlerImpl::forget(const MethodContext& /*context*/,
- u_int16_t /*ticket*/,
+void DtxHandlerImpl::forget(u_int16_t /*ticket*/,
const string& xid)
{
//Currently no heuristic completion is supported, so this should never be used.
throw ConnectionException(503, boost::format("Forget is invalid. Branch with xid %1% not heuristically completed!") % xid);
}
-void DtxHandlerImpl::getTimeout(const MethodContext& context,
- const string& xid)
+void DtxHandlerImpl::getTimeout(const string& xid)
{
uint32_t timeout = broker.getDtxManager().getTimeout(xid);
- cClient.getTimeoutOk(timeout, context.getRequestId());
+ cClient.getTimeoutOk(timeout);//GRS, context.getRequestId());
}
-void DtxHandlerImpl::setTimeout(const MethodContext& context,
- u_int16_t /*ticket*/,
+void DtxHandlerImpl::setTimeout(u_int16_t /*ticket*/,
const string& xid,
u_int32_t timeout)
{
broker.getDtxManager().setTimeout(xid, timeout);
- cClient.setTimeoutOk(context.getRequestId());
+ cClient.setTimeoutOk();//GRS context.getRequestId());
}
-
+void DtxHandlerImpl::setResponseTo(framing::RequestId r)
+{
+ dClient.setResponseTo(r);
+ cClient.setResponseTo(r);
+}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DtxHandlerImpl.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DtxHandlerImpl.h?view=diff&rev=557522&r1=557521&r2=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DtxHandlerImpl.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DtxHandlerImpl.h Thu Jul 19 01:27:36 2007
@@ -36,53 +36,31 @@
public:
DtxHandlerImpl(CoreRefs& parent);
+ void setResponseTo(framing::RequestId r);
+
// DtxCoordinationHandler:
- void commit(const framing::MethodContext& context,
- u_int16_t ticket,
- const std::string& xid,
- bool onePhase );
-
- void forget(const framing::MethodContext& context,
- u_int16_t ticket,
- const std::string& xid );
-
- void getTimeout(const framing::MethodContext& context,
- const std::string& xid );
-
- void prepare(const framing::MethodContext& context,
- u_int16_t ticket,
- const std::string& xid );
-
- void recover(const framing::MethodContext& context,
- u_int16_t ticket,
- bool startscan,
- u_int32_t endscan );
-
- void rollback(const framing::MethodContext& context,
- u_int16_t ticket,
- const std::string& xid );
-
- void setTimeout(const framing::MethodContext& context,
- u_int16_t ticket,
- const std::string& xid,
- u_int32_t timeout );
+ void commit(u_int16_t ticket, const std::string& xid, bool onePhase);
+
+ void forget(u_int16_t ticket, const std::string& xid);
+
+ void getTimeout(const std::string& xid);
+
+ void prepare(u_int16_t ticket, const std::string& xid);
+
+ void recover(u_int16_t ticket, bool startscan, u_int32_t endscan);
+
+ void rollback(u_int16_t ticket, const std::string& xid);
+
+ void setTimeout(u_int16_t ticket, const std::string& xid, u_int32_t timeout);
// DtxDemarcationHandler:
- void end(const framing::MethodContext& context,
- u_int16_t ticket,
- const std::string& xid,
- bool fail,
- bool suspend );
-
- void select(const framing::MethodContext& context );
-
- void start(const framing::MethodContext& context,
- u_int16_t ticket,
- const std::string& xid,
- bool join,
- bool resume );
+ void end(u_int16_t ticket, const std::string& xid, bool fail, bool suspend);
+
+ void select();
+
+ void start(u_int16_t ticket, const std::string& xid, bool join, bool resume);
};
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/GetAdapter.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/GetAdapter.cpp?view=diff&rev=557522&r1=557521&r2=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/GetAdapter.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/GetAdapter.cpp Thu Jul 19 01:27:36 2007
@@ -36,6 +36,5 @@
void GetAdapter::deliver(Message::shared_ptr& msg, framing::RequestId deliveryTag)
{
- msg->sendGetOk(MethodContext(&adapter, msg->getRespondTo()), destination,
- queue->getMessageCount(), deliveryTag, framesize);
+ msg->sendGetOk(adapter, destination, queue->getMessageCount(), 1, deliveryTag, framesize);
}
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=557522&r1=557521&r2=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageHandlerImpl.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageHandlerImpl.cpp Thu Jul 19 01:27:36 2007
@@ -42,72 +42,64 @@
//
void
-MessageHandlerImpl::cancel(const MethodContext& context,
- const string& destination )
+MessageHandlerImpl::cancel(const string& destination )
{
channel.cancel(destination);
- client.ok(context.getRequestId());
+ client.ok();//GRS context.getRequestId());
}
void
-MessageHandlerImpl::open(const MethodContext& context,
- const string& reference)
+MessageHandlerImpl::open(const string& reference)
{
references.open(reference);
- client.ok(context.getRequestId());
+ client.ok();//GRS context.getRequestId());
}
void
-MessageHandlerImpl::append(const MethodContext& context,
- const string& reference,
- const string& /*bytes*/ )
+MessageHandlerImpl::append(const framing::MethodContext& context)
{
- references.get(reference)->append(
- boost::shared_polymorphic_downcast<MessageAppendBody>(
- context.methodBody));
- client.ok(context.getRequestId());
+ MessageAppendBody::shared_ptr body(boost::shared_polymorphic_downcast<MessageAppendBody>(context.methodBody));
+ references.get(body->getReference())->append(body);
+ client.ok();//GRS context.getRequestId());
}
void
-MessageHandlerImpl::close(const MethodContext& context,
- const string& reference)
+MessageHandlerImpl::close(const string& reference)
{
Reference::shared_ptr ref = references.get(reference);
- client.ok(context.getRequestId());
+ client.ok();//GRS context.getRequestId());
// 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.ok((*m)->getRequestId());
+ client.setResponseTo((*m)->getRequestId());
+ client.ok();
}
ref->close();
}
void
-MessageHandlerImpl::checkpoint(const MethodContext& context,
- const string& /*reference*/,
+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(context.getRequestId());
+ client.ok();//GRS context.getRequestId());
}
void
-MessageHandlerImpl::resume(const MethodContext& context,
- 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, context.getRequestId());
+ client.offset(0);//GRS, );//GRS, context.getRequestId());
}
void
-MessageHandlerImpl::offset(const MethodContext&,
- uint64_t /*value*/ )
+MessageHandlerImpl::offset(uint64_t /*value*/ )
{
// Shouldn't ever receive this as it is reponse to resume
// which is never sent
@@ -116,8 +108,7 @@
}
void
-MessageHandlerImpl::consume(const MethodContext& context,
- uint16_t /*ticket*/,
+MessageHandlerImpl::consume(uint16_t /*ticket*/,
const string& queueName,
const string& destination,
bool noLocal,
@@ -132,14 +123,13 @@
channel.consume(std::auto_ptr<DeliveryAdapter>(new ConsumeAdapter(adapter, destination, connection.getFrameMax())),
tag, queue, !noAck, exclusive,
noLocal ? &connection : 0, &filter);
- client.ok(context.getRequestId());
+ client.ok();//GRS context.getRequestId());
// Dispatch messages as there is now a consumer.
queue->requestDispatch();
}
void
-MessageHandlerImpl::get( const MethodContext& context,
- uint16_t /*ticket*/,
+MessageHandlerImpl::get(uint16_t /*ticket*/,
const string& queueName,
const string& destination,
bool noAck )
@@ -148,13 +138,13 @@
GetAdapter out(adapter, queue, destination, connection.getFrameMax());
if(channel.get(out, queue, !noAck))
- client.ok(context.getRequestId());
+ client.ok();//GRS context.getRequestId());
else
- client.empty(context.getRequestId());
+ client.empty();//GRS context.getRequestId());
}
void
-MessageHandlerImpl::empty( const MethodContext& )
+MessageHandlerImpl::empty()
{
// Shouldn't ever receive this as it is a response to get
// which is never sent
@@ -163,34 +153,31 @@
}
void
-MessageHandlerImpl::ok(const MethodContext& /*context*/)
+MessageHandlerImpl::ok()
{
channel.ack(adapter.getFirstAckRequest(), adapter.getLastAckRequest());
}
void
-MessageHandlerImpl::qos(const MethodContext& context,
- uint32_t prefetchSize,
+MessageHandlerImpl::qos(uint32_t prefetchSize,
uint16_t prefetchCount,
bool /*global*/ )
{
//TODO: handle global
channel.setPrefetchSize(prefetchSize);
channel.setPrefetchCount(prefetchCount);
- client.ok(context.getRequestId());
+ client.ok();//GRS context.getRequestId());
}
void
-MessageHandlerImpl::recover(const MethodContext& context,
- bool requeue)
+MessageHandlerImpl::recover(bool requeue)
{
channel.recover(requeue);
- client.ok(context.getRequestId());
+ client.ok();//GRS context.getRequestId());
}
void
-MessageHandlerImpl::reject(const MethodContext& /*context*/,
- uint16_t /*code*/,
+MessageHandlerImpl::reject(uint16_t /*code*/,
const string& /*text*/ )
{
//channel.ack();
@@ -198,45 +185,20 @@
}
void
-MessageHandlerImpl::transfer(const MethodContext& context,
- uint16_t /*ticket*/,
- const string& /* destination */,
- bool /*redelivered*/,
- bool /*immediate*/,
- uint64_t /*ttl*/,
- uint8_t /*priority*/,
- uint64_t /*timestamp*/,
- uint8_t /*deliveryMode*/,
- uint64_t /*expiration*/,
- const string& /*exchangeName*/,
- const string& /*routingKey*/,
- const string& /*messageId*/,
- const string& /*correlationId*/,
- const string& /*replyTo*/,
- const string& /*contentType*/,
- const string& /*contentEncoding*/,
- const string& /*userId*/,
- const string& /*appId*/,
- const string& /*transactionId*/,
- const string& /*securityToken*/,
- const framing::FieldTable& /*applicationHeaders*/,
- const framing::Content& body,
- bool /*mandatory*/)
+MessageHandlerImpl::transfer(const framing::MethodContext& context)
{
MessageTransferBody::shared_ptr transfer(
boost::shared_polymorphic_downcast<MessageTransferBody>(
context.methodBody));
RequestId requestId = context.getRequestId();
- if (body.isInline()) {
- MessageMessage::shared_ptr message(
- new MessageMessage(&connection, requestId, transfer));
+ if (transfer->getBody().isInline()) {
+ MessageMessage::shared_ptr message(new MessageMessage(&connection, requestId, transfer));
channel.handleInlineTransfer(message);
- client.ok(requestId);
+ client.ok();
} else {
- Reference::shared_ptr ref(references.get(body.getValue()));
- MessageMessage::shared_ptr message(
- new MessageMessage(&connection, requestId, transfer, ref));
+ Reference::shared_ptr ref(references.get(transfer->getBody().getValue()));
+ MessageMessage::shared_ptr message(new MessageMessage(&connection, requestId, transfer, ref));
ref->addMessage(message);
}
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageHandlerImpl.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageHandlerImpl.h?view=diff&rev=557522&r1=557521&r2=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageHandlerImpl.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageHandlerImpl.h Thu Jul 19 01:27:36 2007
@@ -40,22 +40,16 @@
public:
MessageHandlerImpl(CoreRefs& parent);
- void append(const framing::MethodContext&,
- const std::string& reference,
- const std::string& bytes );
+ void append(const framing::MethodContext& context);
- void cancel(const framing::MethodContext&,
- const std::string& destination );
+ void cancel(const std::string& destination );
- void checkpoint(const framing::MethodContext&,
- const std::string& reference,
+ void checkpoint(const std::string& reference,
const std::string& identifier );
- void close(const framing::MethodContext&,
- const std::string& reference );
+ void close(const std::string& reference );
- void consume(const framing::MethodContext&,
- uint16_t ticket,
+ void consume(uint16_t ticket,
const std::string& queue,
const std::string& destination,
bool noLocal,
@@ -63,62 +57,32 @@
bool exclusive,
const framing::FieldTable& filter );
- void empty( const framing::MethodContext& );
+ void empty();
- void get(const framing::MethodContext&,
- uint16_t ticket,
+ void get(uint16_t ticket,
const std::string& queue,
const std::string& destination,
bool noAck );
- void offset(const framing::MethodContext&,
- uint64_t value );
+ void offset(uint64_t value);
- void ok( const framing::MethodContext& );
+ void ok();
- void open(const framing::MethodContext&,
- const std::string& reference );
+ void open(const std::string& reference );
- void qos(const framing::MethodContext&,
- uint32_t prefetchSize,
+ void qos(uint32_t prefetchSize,
uint16_t prefetchCount,
bool global );
- void recover(const framing::MethodContext&,
- bool requeue );
+ void recover(bool requeue );
- void reject(const framing::MethodContext&,
- uint16_t code,
+ void reject(uint16_t code,
const std::string& text );
- void resume(const framing::MethodContext&,
- const std::string& reference,
+ void resume(const std::string& reference,
const std::string& identifier );
- void transfer(const framing::MethodContext&,
- uint16_t ticket,
- const std::string& destination,
- bool redelivered,
- bool immediate,
- uint64_t ttl,
- uint8_t priority,
- uint64_t timestamp,
- uint8_t deliveryMode,
- uint64_t expiration,
- const std::string& exchange,
- const std::string& routingKey,
- const std::string& messageId,
- const std::string& correlationId,
- const std::string& replyTo,
- const std::string& contentType,
- const std::string& contentEncoding,
- const std::string& userId,
- const std::string& appId,
- const std::string& transactionId,
- const std::string& securityToken,
- const framing::FieldTable& applicationHeaders,
- const framing::Content& body,
- bool mandatory );
+ void transfer(const framing::MethodContext& context);
private:
ReferenceRegistry references;
};
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=557522&r1=557521&r2=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.cpp Thu Jul 19 01:27:36 2007
@@ -37,7 +37,29 @@
void SemanticHandler::handle(framing::AMQFrame& frame)
{
+ //TODO: assembly etc when move to 0-10 framing
+ //
+ //have potentially three separate tracks at this point:
+ //
+ // (1) execution controls
+ // (2) commands
+ // (3) data i.e. content-bearing commands
+ //
+ //framesets on each can be interleaved. framesets on the latter
+ //two share a command-id sequence.
+ //
+ //need to decide what to do if a frame on the command track
+ //arrives while a frameset on the data track is still
+ //open. execute it (i.e. out-of order execution with respect to
+ //the command id sequence) or queue it up.
+
+ //if ready to execute (i.e. if segment is complete or frame is
+ //message content):
handleBody(frame.getBody());
+ //if the frameset is complete, we can move the execution-mark
+ //forward (not for execution controls)
+ //note: need to be more sophisticated than this if we execute
+ //commands that arrive within an active message frameset
}
//ChannelAdapter virtual methods:
@@ -52,9 +74,12 @@
throw ConnectionException(504, out.str());
}
} else {
+ adapter->setResponseTo(context.getRequestId());
method->invoke(*adapter, context);
+ adapter->setResponseTo(0);
}
}catch(ChannelException& e){
+ adapter->setResponseTo(0);
adapter->getProxy().getChannel().close(
e.code, e.toString(),
method->amqpClassId(), method->amqpMethodId());
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=557522&r1=557521&r2=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.h Thu Jul 19 01:27:36 2007
@@ -26,6 +26,7 @@
#include "Connection.h"
#include "qpid/framing/amqp_types.h"
#include "qpid/framing/FrameHandler.h"
+#include "qpid/framing/SequenceNumber.h"
namespace qpid {
namespace broker {
@@ -37,6 +38,7 @@
Connection& connection;
Channel channel;
std::auto_ptr<BrokerAdapter> adapter;
+ framing::SequenceNumber executionMark;
//ChannelAdapter virtual methods:
void handleMethodInContext(boost::shared_ptr<qpid::framing::AMQMethodBody> method,
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/FramingContent.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/FramingContent.h?view=diff&rev=557522&r1=557521&r2=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/FramingContent.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/FramingContent.h Thu Jul 19 01:27:36 2007
@@ -1,3 +1,23 @@
+/*
+ *
+ * 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_FramingContent_h
#define _framing_FramingContent_h
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/MethodContext.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/MethodContext.cpp?view=diff&rev=557522&r1=557521&r2=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/MethodContext.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/MethodContext.cpp Thu Jul 19 01:27:36 2007
@@ -24,8 +24,11 @@
namespace framing {
RequestId MethodContext::getRequestId() const {
- return boost::shared_polymorphic_downcast<AMQRequestBody>(methodBody)
- ->getRequestId();
+ if (methodBody->type() == REQUEST_BODY) {
+ return boost::shared_polymorphic_cast<AMQRequestBody>(methodBody)->getRequestId();
+ } else {
+ return 0;
+ }
}
}} // namespace qpid::framing
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/Proxy.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/Proxy.h?view=diff&rev=557522&r1=557521&r2=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/Proxy.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/Proxy.h Thu Jul 19 01:27:36 2007
@@ -20,6 +20,7 @@
*/
#include "ProtocolVersion.h"
+#include "amqp_types.h"
namespace qpid {
namespace framing {
Added: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumber.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumber.cpp?view=auto&rev=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumber.cpp (added)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumber.cpp Thu Jul 19 01:27:36 2007
@@ -0,0 +1,62 @@
+/*
+ *
+ * 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 "SequenceNumber.h"
+
+using qpid::framing::SequenceNumber;
+
+SequenceNumber::SequenceNumber() : value(0) {}
+
+SequenceNumber::SequenceNumber(uint32_t v) : value((int32_t) v) {}
+
+bool SequenceNumber::operator==(const SequenceNumber& other) const
+{
+ return value == other.value;
+}
+
+bool SequenceNumber::operator!=(const SequenceNumber& other) const
+{
+ return !(value == other.value);
+}
+
+
+SequenceNumber& SequenceNumber::operator++()
+{
+ value = value + 1;
+ return *this;
+}
+
+const SequenceNumber SequenceNumber::operator++(int)
+{
+ SequenceNumber old(value);
+ value = value + 1;
+ return old;
+}
+
+bool SequenceNumber::operator<(const SequenceNumber& other) const
+{
+ return (value - other.value) < 0;
+}
+
+bool SequenceNumber::operator>(const SequenceNumber& other) const
+{
+ return other < *this;
+}
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumber.cpp
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumber.cpp
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumber.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumber.h?view=auto&rev=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumber.h (added)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumber.h Thu Jul 19 01:27:36 2007
@@ -0,0 +1,51 @@
+/*
+ *
+ * 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_SequenceNumber_h
+#define _framing_SequenceNumber_h
+
+#include "amqp_types.h"
+
+namespace qpid {
+namespace framing {
+
+/**
+ * 4-byte sequence number that 'wraps around'.
+ */
+class SequenceNumber
+{
+ int32_t value;
+ public:
+ SequenceNumber();
+ SequenceNumber(uint32_t v);
+
+ SequenceNumber& operator++();//prefix ++
+ const SequenceNumber operator++(int);//postfix ++
+ bool operator==(const SequenceNumber& other) const;
+ bool operator!=(const SequenceNumber& other) const;
+ bool operator<(const SequenceNumber& other) const;
+ bool operator>(const SequenceNumber& other) const;
+ uint32_t getValue() const { return (uint32_t) value; }
+};
+
+}} // namespace qpid::framing
+
+
+#endif
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumber.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumber.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
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=557522&r1=557521&r2=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/BrokerChannelTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/BrokerChannelTest.cpp Thu Jul 19 01:27:36 2007
@@ -225,9 +225,7 @@
Channel channel(connection, 1, &store);
const string data[] = {"abcde", "fghij", "klmno"};
- Message* msg = new BasicMessage(
- 0, "my_exchange", "my_routing_key", false, false,
- MockChannel::basicGetBody());
+ Message* msg = new BasicMessage(0, "my_exchange", "my_routing_key", false, false);
store.expect();
store.stage(*msg);
@@ -347,8 +345,7 @@
Message* createMessage(const string& exchange, const string& routingKey, const string& messageId, uint64_t contentSize)
{
BasicMessage* msg = new BasicMessage(
- 0, exchange, routingKey, false, false,
- MockChannel::basicGetBody());
+ 0, exchange, routingKey, false, false);
AMQHeaderBody::shared_ptr header(new AMQHeaderBody(BASIC));
header->setContentSize(contentSize);
msg->setHeader(header);
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/ExchangeTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/ExchangeTest.cpp?view=diff&rev=557522&r1=557521&r2=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/ExchangeTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/ExchangeTest.cpp Thu Jul 19 01:27:36 2007
@@ -63,11 +63,7 @@
queue.reset();
queue2.reset();
- Message::shared_ptr msgPtr(
- new BasicMessage(
- 0, "e", "A", true, true,
- AMQMethodBody::shared_ptr(
- new BasicGetBody(ProtocolVersion()))));
+ Message::shared_ptr msgPtr(new BasicMessage(0, "e", "A", true, true));
DeliverableMessage msg(msgPtr);
topic.route(msg, "abc", 0);
direct.route(msg, "abc", 0);
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/Makefile.am
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/Makefile.am?view=diff&rev=557522&r1=557521&r2=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/Makefile.am (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/Makefile.am Thu Jul 19 01:27:36 2007
@@ -84,7 +84,8 @@
framing_unit_tests = \
FieldTableTest \
FramingTest \
- HeaderTest
+ HeaderTest \
+ SequenceNumberTest
misc_unit_tests = \
ProducerConsumerTest
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/MessageBuilderTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/MessageBuilderTest.cpp?view=diff&rev=557522&r1=557521&r2=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/MessageBuilderTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/MessageBuilderTest.cpp Thu Jul 19 01:27:36 2007
@@ -118,8 +118,7 @@
Message::shared_ptr message(
new BasicMessage(
- 0, "test", "my_routing_key", false, false,
- MockChannel::basicGetBody()));
+ 0, "test", "my_routing_key", false, false));
AMQHeaderBody::shared_ptr header(new AMQHeaderBody(BASIC));
header->setContentSize(0);
@@ -137,8 +136,7 @@
string data1("abcdefg");
Message::shared_ptr message(
- new BasicMessage(0, "test", "my_routing_key", false, false,
- MockChannel::basicGetBody()));
+ new BasicMessage(0, "test", "my_routing_key", false, false));
AMQHeaderBody::shared_ptr header(new AMQHeaderBody(BASIC));
header->setContentSize(7);
AMQContentBody::shared_ptr part1(new AMQContentBody(data1));
@@ -160,8 +158,7 @@
string data2("hijklmn");
Message::shared_ptr message(
- new BasicMessage(0, "test", "my_routing_key", false, false,
- MockChannel::basicGetBody()));
+ new BasicMessage(0, "test", "my_routing_key", false, false));
AMQHeaderBody::shared_ptr header(new AMQHeaderBody(BASIC));
header->setContentSize(14);
AMQContentBody::shared_ptr part1(new AMQContentBody(data1));
@@ -191,8 +188,7 @@
string data2("hijklmn");
Message::shared_ptr message(
- new BasicMessage(0, "test", "my_routing_key", false, false,
- MockChannel::basicGetBody()));
+ new BasicMessage(0, "test", "my_routing_key", false, false));
AMQHeaderBody::shared_ptr header(new AMQHeaderBody(BASIC));
header->setContentSize(14);
BasicHeaderProperties* properties = dynamic_cast<BasicHeaderProperties*>(header->getProperties());
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/MessageTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/MessageTest.cpp?view=diff&rev=557522&r1=557521&r2=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/MessageTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/MessageTest.cpp Thu Jul 19 01:27:36 2007
@@ -46,8 +46,7 @@
string data2("hijklmn");
BasicMessage::shared_ptr msg(
- new BasicMessage(0, exchange, routingKey, false, false,
- MockChannel::basicGetBody()));
+ new BasicMessage(0, exchange, routingKey, false, false));
AMQHeaderBody::shared_ptr header(new AMQHeaderBody(BASIC));
header->setContentSize(14);
AMQContentBody::shared_ptr part1(new AMQContentBody(data1));
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp?view=diff&rev=557522&r1=557521&r2=557522
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp Thu Jul 19 01:27:36 2007
@@ -69,8 +69,7 @@
public:
Message::shared_ptr message(std::string exchange, std::string routingKey) {
return Message::shared_ptr(
- new BasicMessage(0, exchange, routingKey, true, true,
- MockChannel::basicGetBody()));
+ new BasicMessage(0, exchange, routingKey, true, true));
}
void testConsumers(){