Session resume support in client & broker: Client can resume a session

after voluntary suspend() or network failure. Frames lost in network
failure are automatically re-transmitted for transparent re-connection.

client::Session improvements:
 - Locking to avoid races between network & user threads.
 - Replaced client::StateManager with sys::StateMonitor - avoid heap allocation.

qpid::Exception clean up:
 - use QPID_MSG consistently to format exception messages.
 - throw typed exceptions (in reply_exceptions.h) for AMQP exceptions.
 - re-throw correct typed exception on client for exceptions from broker.
 - Removed QpidError.h

rubygen/templates/constants.rb: 
 - constants.h: Added FOO_CLASS_ID and FOO_BAR_METHOD_ID constants.
 - reply_constants.h: Added throwReplyException(code, text)
 
log::Logger:
 - Fixed shutdown race in Statement::~Initializer()


git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@588761 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 142 insertions, 0 deletions

diff --git a/cpp/src/tests/SessionState.cpp b/cpp/src/tests/SessionState.cpp
new file mode 100644
index 0000000000..c8d912801e
--- /dev/null
+++ b/cpp/src/tests/SessionState.cpp
@@ -0,0 +1,142 @@
+/*
+ *
+ * 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/SessionState.h"
+
+#define BOOST_AUTO_TEST_MAIN
+#include <boost/test/auto_unit_test.hpp>
+#include <boost/bind.hpp>
+
+using namespace std;
+using namespace qpid::framing;
+using namespace boost;
+
+// Create a frame with a one-char string.
+AMQFrame& frame(char s) {
+    static AMQFrame frame;
+    frame.setBody(AMQContentBody(string(&s, 1)));
+    return frame;
+}
+
+// Extract the one-char string from a frame.
+char charFromFrame(const AMQFrame& f) {
+    const AMQContentBody* b=dynamic_cast<const AMQContentBody*>(f.getBody());
+    BOOST_REQUIRE(b && b->getData().size() > 0);
+    return b->getData()[0];
+}
+
+// Sent chars as frames
+void sent(SessionState& session, const std::string& frames) {
+    for_each(frames.begin(), frames.end(),
+             bind(&SessionState::sent, ref(session), bind(frame, _1)));
+}
+
+// Received chars as frames
+void received(SessionState& session, const std::string& frames) {
+    for_each(frames.begin(), frames.end(),
+             bind(&SessionState::received, session, bind(frame, _1)));
+}
+
+// Make a string from a ReplayRange.
+std::string replayChars(const SessionState::Replay& frames) {
+    string result(frames.size(), ' ');
+    transform(frames.begin(), frames.end(), result.begin(),
+              bind(&charFromFrame, _1));
+    return result;
+}
+
+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(testSent) {
+    // Test that we send solicit-ack at the right interval.
+    AMQContentBody f; 
+    SessionState s1(1);
+    BOOST_CHECK(s1.sent(f));
+    BOOST_CHECK(s1.sent(f));
+    BOOST_CHECK(s1.sent(f));
+    
+    SessionState s3(3);
+    BOOST_CHECK(!s3.sent(f));
+    BOOST_CHECK(!s3.sent(f));
+    BOOST_CHECK(s3.sent(f));
+
+    BOOST_CHECK(!s3.sent(f));
+    BOOST_CHECK(!s3.sent(f));
+    s3.receivedAck(4);
+    BOOST_CHECK(!s3.sent(f));
+    BOOST_CHECK(!s3.sent(f));
+    BOOST_CHECK(s3.sent(f));
+}
+
+BOOST_AUTO_TEST_CASE(testReplay) {
+    // Replay of all frames.
+    SessionState session(100);
+    sent(session, "abc"); 
+    session.suspend(); session.resuming();
+    session.receivedAck(-1);
+    BOOST_CHECK_EQUAL(replayChars(session.replay()), "abc");
+
+    // Replay with acks
+    session.receivedAck(0); // ack a.
+    session.suspend();
+    session.resuming();
+    session.receivedAck(1); // ack b.
+    BOOST_CHECK_EQUAL(replayChars(session.replay()), "c");
+
+    // Replay after further frames.
+    sent(session, "def");
+    session.suspend();
+    session.resuming();
+    session.receivedAck(3);
+    BOOST_CHECK_EQUAL(replayChars(session.replay()), "ef");
+
+    // Bad ack, too high
+    try {
+        session.receivedAck(6);
+        BOOST_FAIL("expected exception");
+    } catch(const qpid::Exception&) {}
+
+}
+
+BOOST_AUTO_TEST_CASE(testReceived) {
+    // Check that we request acks at the right interval.
+    AMQContentBody f;
+    SessionState s1(1);
+    BOOST_CHECK_EQUAL(0u, *s1.received(f));
+    BOOST_CHECK_EQUAL(1u, *s1.received(f));
+    BOOST_CHECK_EQUAL(2u, *s1.received(f));
+
+    SessionState s3(3);
+    BOOST_CHECK(!s3.received(f));
+    BOOST_CHECK(!s3.received(f));
+    BOOST_CHECK_EQUAL(2u, *s3.received(f));
+
+    BOOST_CHECK(!s3.received(f));
+    BOOST_CHECK(!s3.received(f));
+    BOOST_CHECK_EQUAL(5u, *s3.received(f));
+}