summaryrefslogtreecommitdiff
path: root/qpid/cpp/src
diff options
context:
space:
mode:
Diffstat (limited to 'qpid/cpp/src')
-rw-r--r--qpid/cpp/src/qpid/broker/SessionState.cpp3
-rw-r--r--qpid/cpp/src/qpid/broker/amqp_0_10/MessageTransfer.cpp22
-rw-r--r--qpid/cpp/src/qpid/framing/FrameSet.h3
3 files changed, 21 insertions, 7 deletions
diff --git a/qpid/cpp/src/qpid/broker/SessionState.cpp b/qpid/cpp/src/qpid/broker/SessionState.cpp
index 6dfc34a379..c4f2d3f3d3 100644
--- a/qpid/cpp/src/qpid/broker/SessionState.cpp
+++ b/qpid/cpp/src/qpid/broker/SessionState.cpp
@@ -229,8 +229,9 @@ void SessionState::handleContent(AMQFrame& frame)
IncompleteIngressMsgXfer xfer(this, msg);
msg->getIngressCompletion().begin();
- semanticState.route(deliverable.getMessage(), deliverable);
+ // This call should come before routing, because it calcs required credit.
msgBuilder.end();
+ semanticState.route(deliverable.getMessage(), deliverable);
msg->getIngressCompletion().end(xfer); // allows msg to complete xfer
}
}
diff --git a/qpid/cpp/src/qpid/broker/amqp_0_10/MessageTransfer.cpp b/qpid/cpp/src/qpid/broker/amqp_0_10/MessageTransfer.cpp
index c8a293e19e..a43bf8efa6 100644
--- a/qpid/cpp/src/qpid/broker/amqp_0_10/MessageTransfer.cpp
+++ b/qpid/cpp/src/qpid/broker/amqp_0_10/MessageTransfer.cpp
@@ -153,17 +153,27 @@ uint32_t MessageTransfer::getRequiredCredit() const
if (cachedRequiredCredit) {
return requiredCredit;
} else {
- qpid::framing::SumBodySize sum;
- frames.map_if(sum, qpid::framing::TypeFilter2<qpid::framing::HEADER_BODY, qpid::framing::CONTENT_BODY>());
- return sum.getSize();
+ // TODO -- remove this code and replace it with a QPID_ASSERT(cachedRequiredCredit),
+ // then fix whatever breaks. compute should always be called before get.
+ uint32_t sum = 0;
+ for(FrameSet::Frames::const_iterator i = frames.begin(); i != frames.end(); ++i ) {
+ uint8_t type = (*i).getBody()->type();
+ if ((type == qpid::framing::HEADER_BODY ) || (type == qpid::framing::CONTENT_BODY ))
+ sum += (*i).getBody()->encodedSize();
+ }
+ return sum;
}
}
void MessageTransfer::computeRequiredCredit()
{
//add up payload for all header and content frames in the frameset
- qpid::framing::SumBodySize sum;
- frames.map_if(sum, qpid::framing::TypeFilter2<qpid::framing::HEADER_BODY, qpid::framing::CONTENT_BODY>());
- requiredCredit = sum.getSize();
+ uint32_t sum = 0;
+ for(FrameSet::Frames::const_iterator i = frames.begin(); i != frames.end(); ++i ) {
+ uint8_t type = (*i).getBody()->type();
+ if ((type == qpid::framing::HEADER_BODY ) || (type == qpid::framing::CONTENT_BODY ))
+ sum += (*i).getBody()->encodedSize();
+ }
+ requiredCredit = sum;
cachedRequiredCredit = true;
}
diff --git a/qpid/cpp/src/qpid/framing/FrameSet.h b/qpid/cpp/src/qpid/framing/FrameSet.h
index 9abd3ff096..e234864dfd 100644
--- a/qpid/cpp/src/qpid/framing/FrameSet.h
+++ b/qpid/cpp/src/qpid/framing/FrameSet.h
@@ -37,7 +37,10 @@ namespace framing {
*/
class FrameSet
{
+public:
typedef InlineVector<AMQFrame, 4> Frames;
+
+private:
const SequenceNumber id;
Frames parts;
mutable uint64_t contentSize;