From f61e1ef7589da893b9b54448224dc0961515eb40 Mon Sep 17 00:00:00 2001 From: Alan Conway Date: Fri, 26 Oct 2007 19:48:31 +0000 Subject: 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 --- cpp/src/tests/SessionState.cpp | 142 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 142 insertions(+) create mode 100644 cpp/src/tests/SessionState.cpp (limited to 'cpp/src/tests/SessionState.cpp') 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 +#include + +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(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(a.getBody()); + const AMQContentBody* bb=dynamic_cast(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)); +} -- cgit v1.2.1