Author: aconway
Date: Mon Oct 1 08:01:28 2007
New Revision: 580984
URL: http://svn.apache.org/viewvc?rev=580984&view=rev
Log:
* src/qpid/framing/ResumeHandler.cpp: Handle ack and resume logic
for broker and client.
Added:
incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ResumeHandler.cpp (with
props)
incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ResumeHandler.h (with
props)
incubator/qpid/trunk/qpid/cpp/src/tests/ResumeHandler.cpp (with props)
Modified:
incubator/qpid/trunk/qpid/cpp/src/Makefile.am
incubator/qpid/trunk/qpid/cpp/src/tests/Makefile.am
Modified: incubator/qpid/trunk/qpid/cpp/src/Makefile.am
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/Makefile.am?rev=580984&r1=580983&r2=580984&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/Makefile.am (original)
+++ incubator/qpid/trunk/qpid/cpp/src/Makefile.am Mon Oct 1 08:01:28 2007
@@ -110,6 +110,7 @@
qpid/framing/ProtocolInitiation.cpp \
qpid/framing/ProtocolVersion.cpp \
qpid/framing/ProtocolVersionException.cpp \
+ qpid/framing/ResumeHandler.cpp qpid/framing/ResumeHandler.h \
qpid/framing/SendContent.cpp \
qpid/framing/SequenceNumber.cpp \
qpid/framing/SequenceNumberSet.cpp \
Added: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ResumeHandler.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ResumeHandler.cpp?rev=580984&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ResumeHandler.cpp (added)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ResumeHandler.cpp Mon Oct 1
08:01:28 2007
@@ -0,0 +1,56 @@
+/*
+ *
+ * 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 WARRANTIE4bS OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include "ResumeHandler.h"
+#include "qpid/framing/reply_exceptions.h"
+
+#include <boost/bind.hpp>
+
+#include <algorithm>
+
+namespace qpid {
+namespace framing {
+
+void ResumeHandler::ackReceived(SequenceNumber acked) {
+ if (lastSent < acked)
+ throw InvalidArgumentException("Invalid sequence number in ack");
+ size_t keep = lastSent - acked;
+ if (keep < unacked.size())
+ unacked.erase(unacked.begin(), unacked.end()-keep);
+}
+
+void ResumeHandler::resend() {
+ std::for_each(unacked.begin(), unacked.end(),
+ boost::bind(&FrameHandler::handle,out->next, _1));
+}
+
+void ResumeHandler::handleIn(AMQFrame& f) {
+ ++lastReceived;
+ in.next->handle(f);
+}
+
+void ResumeHandler::handleOut(AMQFrame& f) {
+ ++lastSent;
+ unacked.push_back(f);
+ out.next->handle(f);
+}
+
+
+}} // namespace qpid::framing
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ResumeHandler.cpp
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ResumeHandler.cpp
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ResumeHandler.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ResumeHandler.h?rev=580984&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ResumeHandler.h (added)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ResumeHandler.h Mon Oct 1
08:01:28 2007
@@ -0,0 +1,69 @@
+#ifndef QPID_FRAMING_RESUMEHANDLER_H
+#define QPID_FRAMING_RESUMEHANDLER_H
+
+/*
+ *
+ * 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 "qpid/framing/AMQFrame.h"
+#include "qpid/framing/FrameHandler.h"
+#include "qpid/framing/SequenceNumber.h"
+
+#include <deque>
+
+namespace qpid {
+namespace framing {
+
+/**
+ * In/out handler pair for managing exactly-once session delivery.
+ * The same handler is used by client and broker.
+ * This handler only deals with TCP style SequenceNumber acks,
+ * not with fragmented SequenceNumberSet.
+ *
+ * THREAD UNSAFE. Expected to be used in a serialized context.
+ */
+class ResumeHandler : public FrameHandler::InOutHandler
+{
+ public:
+ /** Received acknowledgement for sent frames up to and including sentOk */
+ void ackReceived(SequenceNumber sentOk);
+
+ /** What was the last sequence number we received. */
+ SequenceNumber getLastReceived() { return lastReceived; }
+
+ /** Resend the unacked frames to the output handler */
+ void resend();
+
+ protected:
+ void handleIn(AMQFrame&);
+ void handleOut(AMQFrame&);
+
+ private:
+ typedef std::deque<AMQFrame> Frames;
+ Frames unacked;
+ SequenceNumber lastReceived;
+ SequenceNumber lastSent;
+};
+
+
+}} // namespace qpid::common
+
+
+#endif /*!QPID_FRAMING_RESUMEHANDLER_H*/
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ResumeHandler.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ResumeHandler.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
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?rev=580984&r1=580983&r2=580984&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/Makefile.am (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/Makefile.am Mon Oct 1 08:01:28 2007
@@ -26,6 +26,11 @@
# Session_SOURCES=Session.cpp
# Session_LDADD=-lboost_unit_test_framework $(lib_broker)
+TESTS+=ResumeHandler
+check_PROGRAMS+=ResumeHandler
+ResumeHandler_SOURCES=ResumeHandler.cpp
+ResumeHandler_LDADD=-lboost_unit_test_framework $(lib_common)
+
TESTS+=Blob
check_PROGRAMS+=Blob
Blob_SOURCES=Blob.cpp
Added: incubator/qpid/trunk/qpid/cpp/src/tests/ResumeHandler.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/ResumeHandler.cpp?rev=580984&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/ResumeHandler.cpp (added)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/ResumeHandler.cpp Mon Oct 1
08:01:28 2007
@@ -0,0 +1,87 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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 "qpid/framing/ResumeHandler.h"
+
+#define BOOST_AUTO_TEST_MAIN
+#include <boost/test/auto_unit_test.hpp>
+
+#include <vector>
+
+using namespace std;
+using namespace qpid::framing;
+
+AMQFrame& frame(const char* s) {
+ static AMQFrame frame;
+ frame.setBody(AMQContentBody(s));
+ return frame;
+}
+
+struct Collector : public FrameHandler, public vector<AMQFrame> {
+ void handle(AMQFrame& f) { push_back(f); }
+};
+
+
+namespace qpid {
+namespace framing {
+
+bool operator==(const AMQFrame& a, const AMQFrame& b) {
+ const AMQContentBody* ab=dynamic_cast<const AMQContentBody*>(a.getBody());
+ const AMQContentBody* bb=dynamic_cast<const AMQContentBody*>(b.getBody());
+ return ab && bb && ab->getData() == bb->getData();
+}
+
+}} // namespace qpid::framing
+
+
+BOOST_AUTO_TEST_CASE(testSend) {
+ AMQFrame f;
+ ResumeHandler sender;
+ Collector collect;
+ sender.out.next = &collect;
+ sender.out(frame("a"));
+ BOOST_CHECK_EQUAL(1u, collect.size());
+ BOOST_CHECK_EQUAL(frame("a"), collect[0]);
+ sender.out(frame("b"));
+ sender.out(frame("c"));
+ sender.ackReceived(1); // ack a,b.
+ sender.out(frame("d"));
+ BOOST_CHECK_EQUAL(4u, collect.size());
+ BOOST_CHECK_EQUAL(frame("d"), collect.back());
+ // Now try a resend.
+ collect.clear();
+ sender.resend();
+ BOOST_REQUIRE_EQUAL(collect.size(), 2u);
+ BOOST_CHECK_EQUAL(frame("c"), collect[0]);
+ BOOST_CHECK_EQUAL(frame("d"), collect[1]);
+}
+
+
+BOOST_AUTO_TEST_CASE(testReceive) {
+ ResumeHandler receiver;
+ Collector collect;
+ receiver.in.next = &collect;
+ receiver.in(frame("a"));
+ receiver.in(frame("b"));
+ BOOST_CHECK_EQUAL(receiver.getLastReceived().getValue(), 1u);
+ receiver.in(frame("c"));
+ BOOST_CHECK_EQUAL(receiver.getLastReceived().getValue(), 2u);
+ BOOST_CHECK_EQUAL(3u, collect.size());
+ BOOST_CHECK_EQUAL(frame("a"), collect[0]);
+ BOOST_CHECK_EQUAL(frame("c"), collect[2]);
+}
Propchange: incubator/qpid/trunk/qpid/cpp/src/tests/ResumeHandler.cpp
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/trunk/qpid/cpp/src/tests/ResumeHandler.cpp
------------------------------------------------------------------------------
svn:keywords = Rev Date
