23 files changed, 442 insertions, 281 deletions
diff --git a/cpp/src/tests/ClientChannelTest.cpp b/cpp/src/tests/ClientChannelTest.cpp
index 9a982508d1..454b9ca56d 100644
--- a/cpp/src/tests/ClientChannelTest.cpp
+++ b/cpp/src/tests/ClientChannelTest.cpp
@@ -56,7 +56,7 @@ class ChannelTestBase : public CppUnit::TestCase
         }
     };
     
-    InProcessBrokerClient connection; // client::connection + local broker
+    qpid::InProcessBrokerClient connection; 
     const std::string qname;
     const std::string data;
     Queue queue;
diff --git a/cpp/src/tests/ClientSessionTest.cpp b/cpp/src/tests/ClientSessionTest.cpp
index 2495a06fa4..db2cd62b0a 100644
--- a/cpp/src/tests/ClientSessionTest.cpp
+++ b/cpp/src/tests/ClientSessionTest.cpp
@@ -18,15 +18,21 @@
  * under the License.
  *
  */
-#include <list>
 #include "qpid_test_plugin.h"
 #include "InProcessBroker.h"
 #include "qpid/client/Dispatcher.h"
 #include "qpid/client/Session.h"
 #include "qpid/framing/TransferContent.h"
+#include "qpid/framing/reply_exceptions.h"
+
+#include <boost/optional.hpp>
+
+#include <list>
 
 using namespace qpid::client;
 using namespace qpid::framing;
+using namespace qpid;
+using namespace boost;
 
 struct DummyListener : public MessageListener
 {
@@ -60,58 +66,77 @@ class ClientSessionTest : public CppUnit::TestCase
     CPPUNIT_TEST(testQueueQuery);
     CPPUNIT_TEST(testTransfer);
     CPPUNIT_TEST(testDispatcher);
+    CPPUNIT_TEST(testResumeExpiredError);
+    CPPUNIT_TEST(testUseSuspendedError);
     CPPUNIT_TEST(testSuspendResume);
-    CPPUNIT_TEST(testSuspendResumeErrors);
+    CPPUNIT_TEST(testDisconnectResume);
+    CPPUNIT_TEST(testAutoDelete);
     CPPUNIT_TEST_SUITE_END();
 
-    boost::shared_ptr<Connector> broker;
-    Connection connection;
+    shared_ptr<broker::Broker> broker;
     Session session;
+    // Defer construction & thread creation to setUp
+    boost::optional<InProcessConnection> c;
+    boost::optional<InProcessConnection> c2;
 
 public:
 
-    ClientSessionTest() : broker(new qpid::broker::InProcessBroker()), connection(broker) 
+    void setUp() {
+        broker = broker::Broker::create();
+        c=boost::in_place<InProcessConnection>(broker);
+        c2=boost::in_place<InProcessConnection>(broker);
+    }
+
+    void tearDown() {
+        c2.reset();
+        c.reset();
+        broker.reset();
+    }
+
+    void declareSubscribe(const std::string& q="my-queue",
+                          const std::string& dest="my-dest")
     {
-        connection.open("");
-        session = connection.newSession();
+        // FIXME aconway 2007-10-18: autoDelete queues are destroyed on channel close, not session.
+        // Fix & make all test queues exclusive, autoDelete
+        session.queueDeclare_(queue=q);  // FIXME aconway 2007-10-01: exclusive=true, autoDelete=true);
+        session.messageSubscribe_(queue=q, destination=dest, acquireMode=1);
+        session.messageFlow_(destination=dest, unit=0, value=0xFFFFFFFF);//messages
+        session.messageFlow_(destination=dest, unit=1, value=0xFFFFFFFF);//bytes
     }
 
+    bool queueExists(const std::string& q) {
+        TypedResult<QueueQueryResult> result = session.queueQuery_(q);
+        return result.get().getQueue() == q;
+    }
+    
     void testQueueQuery() 
     {
-        std::string name("my-queue");
-        std::string alternate("amq.fanout");
-        session.queueDeclare((queue=name, alternateExchange=alternate, exclusive=true, autoDelete=true));
-        TypedResult<QueueQueryResult> result = session.queueQuery(name);
+        session = c->newSession();
+        session.queueDeclare_(queue="my-queue", alternateExchange="amq.fanout", exclusive=true, autoDelete=true);
+        TypedResult<QueueQueryResult> result = session.queueQuery_(std::string("my-queue"));
         CPPUNIT_ASSERT_EQUAL(false, result.get().getDurable());
         CPPUNIT_ASSERT_EQUAL(true, result.get().getExclusive());
-        CPPUNIT_ASSERT_EQUAL(alternate, result.get().getAlternateExchange());
+        CPPUNIT_ASSERT_EQUAL(std::string("amq.fanout"),
+                             result.get().getAlternateExchange());
     }
 
     void testTransfer()
     {
-        std::string queueName("my-queue");
-        std::string dest("my-dest");
-        std::string data("my message");
-        session.queueDeclare_(queue=queueName, exclusive=true, autoDelete=true);
-        //subcribe to the queue with confirm_mode = 1:
-        session.messageSubscribe_(queue=queueName, destination=dest, acquireMode=1);
-        session.messageFlow((destination=dest, unit=0, value=1));//messages
-        session.messageFlow((destination=dest, unit=1, value=0xFFFFFFFF));//bytes
-        //publish a message:
-        TransferContent _content(data);
-        _content.getDeliveryProperties().setRoutingKey("my-queue");
-        session.messageTransfer_(content=_content);
+        session = c->newSession();
+        declareSubscribe();
+        session.messageTransfer_(content=TransferContent("my-message", "my-queue"));
         //get & test the message:
         FrameSet::shared_ptr msg = session.get();
         CPPUNIT_ASSERT(msg->isA<MessageTransferBody>());
-        CPPUNIT_ASSERT_EQUAL(data, msg->getContent());
+        CPPUNIT_ASSERT_EQUAL(std::string("my-message"), msg->getContent());
         //confirm receipt:
         session.execution().completed(msg->getId(), true, true);
     }
 
     void testDispatcher()
     {
-        session.queueDeclare_(queue="my-queue", exclusive=true, autoDelete=true);
+        session = c->newSession();
+        declareSubscribe();
 
         TransferContent msg1("One");
         msg1.getDeliveryProperties().setRoutingKey("my-queue");
@@ -125,9 +150,6 @@ public:
         msg3.getDeliveryProperties().setRoutingKey("my-queue");
         session.messageTransfer_(content=msg3);
                 
-        session.messageSubscribe_(queue="my-queue", destination="my-dest", acquireMode=1);
-        session.messageFlow((destination="my-dest", unit=0, value=1));//messages
-        session.messageFlow((destination="my-dest", unit=1, value=0xFFFFFFFF));//bytes
         DummyListener listener(session, "my-dest", 3);
         listener.listen();
         CPPUNIT_ASSERT_EQUAL((size_t) 3, listener.messages.size());        
@@ -140,29 +162,66 @@ public:
 
     }
 
-    void testSuspendResume() {
-        session = connection.newSession(60);
+    void testResumeExpiredError() {
+        session = c->newSession(0);
+        session.suspend();  // session has 0 timeout.
+        try {
+            c->resume(session);
+            CPPUNIT_FAIL("Expected InvalidArgumentException.");
+        } catch(const InvalidArgumentException&) {}
+    }
+
+    void testUseSuspendedError() {
+        session = c->newSession(60);
         session.suspend();
         try {
             session.exchangeQuery_(name="amq.fanout");
             CPPUNIT_FAIL("Expected session suspended exception");
-        } catch(...) {}
-        connection.resume(session);
-        session.exchangeQuery_(name="amq.fanout");
-        // FIXME aconway 2007-09-25: build up session state and confirm
-        //it survives the resume
+        } catch(const CommandInvalidException&) {}
     }
 
-    void testSuspendResumeErrors() {
-        session.suspend();  // session has 0 timeout.
-        try {
-            session.exchangeQuery_(name="amq.fanout");
-            CPPUNIT_FAIL("Expected suspended session exception");
-        } catch(...) {}
-        try {
-            connection.resume(session);
-            CPPUNIT_FAIL("Expected no such session exception.");
-        } catch(...) {}
+    void testSuspendResume() {
+        session = c->newSession(60);
+        declareSubscribe();
+        session.suspend();
+        // Make sure we are still subscribed after resume.
+        c->resume(session);
+        session.messageTransfer_(content=TransferContent("my-message", "my-queue"));
+        FrameSet::shared_ptr msg = session.get();
+        CPPUNIT_ASSERT_EQUAL(string("my-message"), msg->getContent());
+    }
+
+    void testDisconnectResume() {
+        session = c->newSession(60);
+        session.queueDeclare_(queue="before");
+        CPPUNIT_ASSERT(queueExists("before"));
+        // Simulate lost frames.
+        c->discard();
+        session.queueDeclare_(queue=string("after"));
+        c->disconnect(); // Simulate disconnect, resume on a new connection.
+        c2->resume(session);
+        CPPUNIT_ASSERT(queueExists("after"));
+    }
+
+    void testAutoDelete() {
+        // Verify that autoDelete queues survive suspend/resume.
+        session = c->newSession(60);
+        session.queueDeclare_(queue="my-queue", exclusive=true, autoDelete=true);
+        CPPUNIT_ASSERT(queueExists("my-queue"));
+        session.suspend();
+        c->resume(session);
+        CPPUNIT_ASSERT(queueExists("my-queue"));
+
+        // Verify they survive disconnect/resume on new Connection
+        c->disconnect();
+        c2->resume(session);
+
+        try { 
+            // FIXME aconway 2007-10-23: Negative test, need to
+            // fix auto-delete queues to clean up with session, not channel.
+            CPPUNIT_ASSERT(queueExists("my-queue"));
+            CPPUNIT_FAIL("Negative test passed unexpectedly");
+        } catch(const ChannelException&) {}
     }
 };
 
diff --git a/cpp/src/tests/ConcurrentQueue.cpp b/cpp/src/tests/ConcurrentQueue.cpp
index e1adcce0f9..39155b4ff2 100644
--- a/cpp/src/tests/ConcurrentQueue.cpp
+++ b/cpp/src/tests/ConcurrentQueue.cpp
@@ -61,7 +61,7 @@ template <class T> class DualVectorDualLockQueue {
     /** If the queue is non-empty, pop the front item into data and
      * return true. If the queue is empty, return false
      */
-    bool pop(T& data) {
+    bool tryPop(T& data) {
         Mutex::ScopedLock l(popLock);
         if (popIter == popVec.end()) {
             popVec.clear();
@@ -109,7 +109,7 @@ void nspin(const Duration& delay) {
 struct NullQueue {
     NullQueue(int items=0) : npush(items), npop(items) {}
     void push(int) { --npush; }
-    bool pop(int& n) {
+    bool tryPop(int& n) {
         if (npop == 0)
             return false;
         else {
@@ -144,7 +144,7 @@ struct Popper : public Runnable {
     void run() {
         for (int i=items; i > 0; i--) {
             int n;
-            if (queue.pop(n))
+            if (queue.tryPop(n))
                 BOOST_REQUIRE_EQUAL(i,n);
             npause();
         }
diff --git a/cpp/src/tests/EventChannelTest.cpp b/cpp/src/tests/EventChannelTest.cpp
index 3ba54def86..6d8d64e165 100644
--- a/cpp/src/tests/EventChannelTest.cpp
+++ b/cpp/src/tests/EventChannelTest.cpp
@@ -117,18 +117,18 @@ class EventChannelTest : public CppUnit::TestCase
         CPPUNIT_ASSERT(re.hasError());
         try {
             re.throwIfError();
-            CPPUNIT_FAIL("Expected QpidError.");
+            CPPUNIT_FAIL("Expected Exception.");
         }
-        catch (const qpid::QpidError&) { }
+        catch (const qpid::Exception&) { }
 
         //  Bad file descriptor. Note in this case we fail
         //  in postEvent and throw immediately.
         try {
             ReadEvent bad;
             ec->postEvent(bad);
-            CPPUNIT_FAIL("Expected QpidError.");
+            CPPUNIT_FAIL("Expected Exception.");
         }
-        catch (const qpid::QpidError&) { }
+        catch (const qpid::Exception&) { }
     }
 
     void testWrite() {
diff --git a/cpp/src/tests/FramingTest.cpp b/cpp/src/tests/FramingTest.cpp
index 5ca4e6c216..9e82447ffa 100644
--- a/cpp/src/tests/FramingTest.cpp
+++ b/cpp/src/tests/FramingTest.cpp
@@ -18,8 +18,6 @@
  * under the License.
  *
  */
-#include "InProcessBroker.h"
-#include "qpid/QpidError.h"
 #include "qpid/client/Exchange.h"
 #include "qpid/client/Queue.h"
 #include "qpid/client/Connection.h"
@@ -30,6 +28,7 @@
 #include "qpid/framing/ProtocolVersion.h"
 #include "qpid/framing/all_method_bodies.h"
 #include "qpid/framing/amqp_framing.h"
+#include "qpid/framing/reply_exceptions.h"
 #include "qpid_test_plugin.h"
 
 #include <boost/bind.hpp>
@@ -200,18 +199,12 @@ class FramingTest : public CppUnit::TestCase
         try {
             Content content(REFERENCE, "");
             CPPUNIT_ASSERT(false);//fail, expected exception
-        } catch (QpidError& e) {
-            CPPUNIT_ASSERT_EQUAL(FRAMING_ERROR, e.code);
-            CPPUNIT_ASSERT_EQUAL(string("Reference cannot be empty"), e.msg);
-        }
+        } catch (const InvalidArgumentException& e) {}
         
         try {
             Content content(2, "Blah");
             CPPUNIT_ASSERT(false);//fail, expected exception
-        } catch (QpidError& e) {
-            CPPUNIT_ASSERT_EQUAL(FRAMING_ERROR, e.code);
-            CPPUNIT_ASSERT_EQUAL(string("Invalid discriminator: 2"), e.msg);
-        }
+        } catch (const SyntaxErrorException& e) {}
         
         try {
             Buffer wbuff(buffer, sizeof(buffer));
@@ -221,11 +214,8 @@ class FramingTest : public CppUnit::TestCase
             Buffer rbuff(buffer, sizeof(buffer));
             Content content;
             content.decode(rbuff);
-            CPPUNIT_ASSERT(false);//fail, expected exception
-        } catch (QpidError& e) {
-            CPPUNIT_ASSERT_EQUAL(FRAMING_ERROR, e.code);
-            CPPUNIT_ASSERT_EQUAL(string("Invalid discriminator: 2"), e.msg);
-        }
+            CPPUNIT_FAIL("Expected exception");
+        } catch (Exception& e) {}
         
     }
 
diff --git a/cpp/src/tests/HeadersExchangeTest.cpp b/cpp/src/tests/HeadersExchangeTest.cpp
index c47266caa6..f07f238ee4 100644
--- a/cpp/src/tests/HeadersExchangeTest.cpp
+++ b/cpp/src/tests/HeadersExchangeTest.cpp
@@ -19,7 +19,7 @@
  *
  */
 
-#include "qpid/QpidError.h"
+#include "qpid/Exception.h"
 #include "qpid/broker/HeadersExchange.h"
 #include "qpid/framing/FieldTable.h"
 #include "qpid/framing/FieldValue.h"
@@ -118,7 +118,7 @@ class HeadersExchangeTest : public CppUnit::TestCase
         try {
             //just checking this doesn't cause assertion etc
             exchange.bind(queue, key, &args);
-        } catch(qpid::QpidError&) {
+        } catch(qpid::Exception&) {
             //expected
         }
     }
diff --git a/cpp/src/tests/InProcessBroker.h b/cpp/src/tests/InProcessBroker.h
index 2a9f12771b..c5860568db 100644
--- a/cpp/src/tests/InProcessBroker.h
+++ b/cpp/src/tests/InProcessBroker.h
@@ -25,6 +25,9 @@
 #include "qpid/client/Connector.h"
 #include "qpid/client/Connection.h"
 #include "qpid/log/Statement.h"
+#include "qpid/sys/Thread.h"
+#include "qpid/sys/ConcurrentQueue.h"
+#include "qpid/shared_ptr.h"
 
 #include <vector>
 #include <iostream>
@@ -32,112 +35,176 @@
 
 
 namespace qpid {
-namespace broker {
+
 
 /**
- * A broker that implements client::Connector allowing direct
- * in-process connection of client to broker. Used to write round-trip
- * tests without requiring an external broker process.
- *
+ * A client::Connector that connects directly to an in-process broker.
  * Also allows you to "snoop" on frames exchanged between client & broker.
  * 
  * see FramingTest::testRequestResponseRoundtrip() for example of use.
  */
-class InProcessBroker : public client::Connector {
+class InProcessConnector :
+        public client::Connector
+{
   public:
+    typedef sys::Mutex Mutex;
+    typedef Mutex::ScopedLock Lock;
+    typedef framing::FrameHandler FrameHandler;
+    typedef framing::AMQFrame AMQFrame;
+
     enum Sender {CLIENT,BROKER};
 
-    /** A frame tagged with the sender */
-    struct TaggedFrame {
-        TaggedFrame(Sender e, framing::AMQFrame& f) : frame(f), sender(e) {}
-        bool fromBroker() const { return sender == BROKER; }
-        bool fromClient() const { return sender == CLIENT; }
+    /** Simulate the network thread of a peer with a queue and a thread.
+     * With setInputHandler(0) drops frames simulating network packet loss.
+     */
+    class NetworkQueue : public sys::Runnable
+    {
+      public:
+        NetworkQueue(const char* r) : inputHandler(0), receiver(r) {
+            thread=sys::Thread(this);
+        }
 
-        template <class MethodType>
-        MethodType* asMethod() {
-            return dynamic_cast<MethodType*>(frame.getBody());
+        ~NetworkQueue() { 
+            queue.shutdown();
+            thread.join();
         }
-        framing::AMQFrame frame;
-        Sender sender;
+
+        void push(AMQFrame& f) { queue.push(f); }
+
+        void run() {
+            AMQFrame f;
+            while (queue.waitPop(f)) {
+                Lock l(lock);
+                if (inputHandler) { 
+                    QPID_LOG(debug, QPID_MSG(receiver << " RECV: " << f));
+                    inputHandler->handle(f);
+                }
+                else {
+                    QPID_LOG(debug, QPID_MSG(receiver << " DROP: " << f));
+                }
+            }
+        }
+
+        void setInputHandler(FrameHandler* h) {
+            Lock l(lock);
+            inputHandler = h;
+        }
+        
+      private:
+        sys::Mutex lock;
+        sys::ConcurrentQueue<AMQFrame> queue;
+        sys::Thread thread;
+        FrameHandler* inputHandler;
+        const char* const receiver;
     };
-    
-    typedef std::vector<TaggedFrame> Conversation;
-
-    InProcessBroker(framing::ProtocolVersion ver=
-                    framing::highestProtocolVersion
-    ) :
-        Connector(ver),
-        protocolInit(ver),
-        broker(broker::Broker::create()),
-        brokerOut(BROKER, conversation),
+
+    struct InProcessHandler : public sys::ConnectionOutputHandler {
+        Sender from;
+        NetworkQueue queue;
+        const char* const sender;
+
+        InProcessHandler(Sender s)
+            : from(s),
+              queue(from==CLIENT? "BROKER" : "CLIENT"),
+              sender(from==BROKER? "BROKER" : "CLIENT")
+        {}
+
+        ~InProcessHandler() {  }
+        
+        void send(AMQFrame& f) {
+            QPID_LOG(debug, QPID_MSG(sender << " SENT: " << f));
+            queue.push(f);
+        }
+        
+        void close() {
+            // Do not shut down the queue here, we may be in
+            // the queue's dispatch thread. 
+        }
+    };
+
+    InProcessConnector(shared_ptr<broker::Broker> b,
+                       framing::ProtocolVersion v=framing::ProtocolVersion()) :
+        Connector(v),
+        protocolInit(v),
+        broker(b),
+        brokerOut(BROKER),
         brokerConnection(&brokerOut, *broker),
-        clientOut(CLIENT, conversation, &brokerConnection)
-    {}
+        clientOut(CLIENT),
+        isClosed(false)
+    {
+        clientOut.queue.setInputHandler(&brokerConnection);
+    }
 
-    ~InProcessBroker() { broker->shutdown(); }
+    ~InProcessConnector() {
+        close();
+        
+    }
 
     void connect(const std::string& /*host*/, int /*port*/) {}
+
     void init() { brokerConnection.initiated(protocolInit); }
-    void close() {}
+
+    void close() {
+        if (!isClosed) {
+            isClosed = true;
+            brokerOut.close();
+            clientOut.close();
+            brokerConnection.closed();
+        }
+    }
 
     /** Client's input handler. */
     void setInputHandler(framing::InputHandler* handler) {
-        brokerOut.in = handler;
+        brokerOut.queue.setInputHandler(handler);
     }
 
     /** Called by client to send a frame */
     void send(framing::AMQFrame& frame) {
-        clientOut.send(frame);
+        clientOut.handle(frame);
     }
 
-    /** Entire client-broker conversation is recorded here */
-    Conversation conversation;
+    /** Sliently discard frames sent by either party, lost network traffic. */
+    void discard() {
+        brokerOut.queue.setInputHandler(0);
+        clientOut.queue.setInputHandler(0);
+    }
 
   private:
-    /** OutputHandler that forwards data to an InputHandler */
-    struct OutputToInputHandler : public sys::ConnectionOutputHandler {
-        OutputToInputHandler(
-            Sender sender_, Conversation& conversation_,
-            framing::InputHandler* ih=0
-        ) : sender(sender_), conversation(conversation_), in(ih) {}
-
-        void send(framing::AMQFrame& frame) {
-            QPID_LOG(debug,
-                     (sender==CLIENT ? "CLIENT: " : "BROKER: ") << frame);
-            conversation.push_back(TaggedFrame(sender, frame));
-            in->received(frame);
-        }
-
-        void close() {}
-        
-        Sender sender;
-        Conversation& conversation;
-        framing::InputHandler* in;
-    };
-
+    sys::Mutex lock;
     framing::ProtocolInitiation protocolInit;
-    shared_ptr<Broker>  broker;
-    OutputToInputHandler brokerOut;
+    shared_ptr<broker::Broker>  broker;
+    InProcessHandler brokerOut;
     broker::Connection brokerConnection;
-    OutputToInputHandler clientOut;
+    InProcessHandler  clientOut;
+    bool isClosed;
 };
 
-std::ostream& operator<<(
-    std::ostream& out, const InProcessBroker::TaggedFrame& tf)
-{
-    return out << (tf.fromBroker()? "BROKER: ":"CLIENT: ") << tf.frame;
-}
-
-std::ostream& operator<<(
-    std::ostream& out, const InProcessBroker::Conversation& conv)
-{    
-    copy(conv.begin(), conv.end(),
-         std::ostream_iterator<InProcessBroker::TaggedFrame>(out, "\n"));
-    return out;
-}
-
-} // namespace broker
-} // namespace qpid
+struct InProcessConnection : public client::Connection {
+    InProcessConnection(shared_ptr<broker::Broker> b)
+        : client::Connection(
+            shared_ptr<client::Connector>(
+                new InProcessConnector(b)))
+    {
+        open("");
+    }
+
+    ~InProcessConnection() { }
+    
+    /** Simulate disconnected network connection. */
+    void disconnect() { impl->getConnector()->close(); }
+    
+    /** Sliently discard frames sent by either party, lost network traffic. */
+    void discard() {
+        dynamic_pointer_cast<InProcessConnector>(
+            impl->getConnector())->discard();
+    }
+};   
 
+/** A connector with its own broker */
+struct InProcessBroker : public InProcessConnector {
+    InProcessBroker() : InProcessConnector(broker::Broker::create()) {}
+};
+
+} // namespace qpid
 
 #endif // _tests_InProcessBroker_h
diff --git a/cpp/src/tests/Makefile.am b/cpp/src/tests/Makefile.am
index 233614367d..b954ae88f4 100644
--- a/cpp/src/tests/Makefile.am
+++ b/cpp/src/tests/Makefile.am
@@ -20,16 +20,10 @@ CLEANFILES=
 # Unit test programs.
 # 
 
-# FIXME aconway 2007-08-29: enable when session is reinstated.
-# TESTS+=Session
-# check_PROGRAMS+=Session
-# 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+=SessionState
+check_PROGRAMS+=SessionState
+SessionState_SOURCES=SessionState.cpp 
+SessionState_LDADD=-lboost_unit_test_framework $(lib_common)
 
 TESTS+=Blob
 check_PROGRAMS+=Blob
diff --git a/cpp/src/tests/QueueTest.cpp b/cpp/src/tests/QueueTest.cpp
index 114e0045f5..3235fe2418 100644
--- a/cpp/src/tests/QueueTest.cpp
+++ b/cpp/src/tests/QueueTest.cpp
@@ -54,7 +54,7 @@ class FailOnDeliver : public Deliverable
 public:
     void deliverTo(Queue::shared_ptr& queue)
     {
-        throw Exception(boost::format("Invalid delivery to %1%") % queue->getName());
+        throw Exception(QPID_MSG("Invalid delivery to " << queue->getName()));
     }
 };
 
diff --git a/cpp/src/tests/ResumeHandler.cpp b/cpp/src/tests/ResumeHandler.cpp
deleted file mode 100644
index 1073e42a3c..0000000000
--- a/cpp/src/tests/ResumeHandler.cpp
+++ /dev/null
@@ -1,87 +0,0 @@
-/*
- *
- * 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]);
-}
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));
+}
diff --git a/cpp/src/tests/Shlib.cpp b/cpp/src/tests/Shlib.cpp
index 87136425ab..6420af915e 100644
--- a/cpp/src/tests/Shlib.cpp
+++ b/cpp/src/tests/Shlib.cpp
@@ -20,6 +20,7 @@
 
 #include "test_tools.h"
 #include "qpid/sys/Shlib.h"
+#include "qpid/Exception.h"
 
 #define BOOST_AUTO_TEST_MAIN    // Must come before #include<boost/test/*>
 #include <boost/test/auto_unit_test.hpp>
@@ -40,7 +41,7 @@ BOOST_AUTO_TEST_CASE(testShlib) {
         sh.getSymbol("callMe");
         BOOST_FAIL("Expected exception");
     }
-    catch (...) {}
+    catch (const qpid::Exception&) {}
 }
     
 BOOST_AUTO_TEST_CASE(testAutoShlib) {
diff --git a/cpp/src/tests/TimerTest.cpp b/cpp/src/tests/TimerTest.cpp
index 682699dbd3..3f2a1c57ec 100644
--- a/cpp/src/tests/TimerTest.cpp
+++ b/cpp/src/tests/TimerTest.cpp
@@ -26,6 +26,7 @@
 #include <iostream>
 #include <memory>
 #include <boost/format.hpp>
+#include <boost/lexical_cast.hpp>
 
 using namespace qpid::broker;
 using namespace qpid::sys;
diff --git a/cpp/src/tests/TxMocks.h b/cpp/src/tests/TxMocks.h
index e4e74ee535..127a27c005 100644
--- a/cpp/src/tests/TxMocks.h
+++ b/cpp/src/tests/TxMocks.h
@@ -25,7 +25,6 @@
 #include "qpid/Exception.h"
 #include "qpid/broker/TransactionalStore.h"
 #include "qpid/broker/TxOp.h"
-#include <boost/format.hpp>
 #include <iostream>
 #include <vector>
 
@@ -40,9 +39,9 @@ template <class T> void assertEqualVector(std::vector<T>& expected, std::vector<
         i++;
     }
     if (i < expected.size()) {
-        throw qpid::Exception(boost::format("Missing %1%") % expected[i]);
+        throw qpid::Exception(QPID_MSG("Missing " << expected[i]));
     } else if (i < actual.size()) {
-        throw qpid::Exception(boost::format("Extra %1%") % actual[i]);
+        throw qpid::Exception(QPID_MSG("Extra " << actual[i]));
     }
     CPPUNIT_ASSERT_EQUAL(expected.size(), actual.size());
 }
diff --git a/cpp/src/tests/client_test.cpp b/cpp/src/tests/client_test.cpp
index 8cf43ce069..e4fd57824c 100644
--- a/cpp/src/tests/client_test.cpp
+++ b/cpp/src/tests/client_test.cpp
@@ -29,7 +29,6 @@
 #include <iostream>
 
 #include "TestOptions.h"
-#include "qpid/QpidError.h"
 #include "qpid/client/Channel.h"
 #include "qpid/client/Connection.h"
 #include "qpid/client/Message.h"
diff --git a/cpp/src/tests/echo_service.cpp b/cpp/src/tests/echo_service.cpp
index 7989ec8543..c3569d5fd4 100644
--- a/cpp/src/tests/echo_service.cpp
+++ b/cpp/src/tests/echo_service.cpp
@@ -27,7 +27,6 @@
  * sender-specified private queue.
  */
 
-#include "qpid/QpidError.h"
 #include "qpid/client/Channel.h"
 #include "qpid/client/Connection.h"
 #include "qpid/client/Exchange.h"
diff --git a/cpp/src/tests/exception_test.cpp b/cpp/src/tests/exception_test.cpp
index 3feef7e876..3783ae6901 100644
--- a/cpp/src/tests/exception_test.cpp
+++ b/cpp/src/tests/exception_test.cpp
@@ -22,7 +22,6 @@
 #include <iostream>
 
 #include "TestOptions.h"
-#include "qpid/QpidError.h"
 #include "qpid/client/Channel.h"
 #include "qpid/client/Connection.h"
 #include "qpid/client/Message.h"
diff --git a/cpp/src/tests/interop_runner.cpp b/cpp/src/tests/interop_runner.cpp
index 5bfe88662a..56f9cbf3d2 100644
--- a/cpp/src/tests/interop_runner.cpp
+++ b/cpp/src/tests/interop_runner.cpp
@@ -21,7 +21,6 @@
 
 #include "qpid/Options.h"
 #include "qpid/Exception.h"
-#include "qpid/QpidError.h"
 #include "qpid/client/Channel.h"
 #include "qpid/client/Connection.h"
 #include "qpid/client/Exchange.h"
diff --git a/cpp/src/tests/logging.cpp b/cpp/src/tests/logging.cpp
index f5402aaad7..1042e60077 100644
--- a/cpp/src/tests/logging.cpp
+++ b/cpp/src/tests/logging.cpp
@@ -71,22 +71,18 @@ BOOST_AUTO_TEST_CASE(testSelector_enable) {
     BOOST_CHECK(s.isEnabled(critical, "oops"));
 }
 
-Logger& clearLogger() {
-    Logger::instance().clear();
-    return Logger::instance();
-}
-
 BOOST_AUTO_TEST_CASE(testStatementEnabled) {
-    // Verify that the logger enables and disables log statements.
-    Logger& l=clearLogger();
+    // Verify that the singleton enables and disables static
+    // log statements.
+    Logger& l = Logger::instance();
     l.select(Selector(debug));
-    Statement s=QPID_LOG_STATEMENT_INIT(debug);
+    static Statement s=QPID_LOG_STATEMENT_INIT(debug);
     BOOST_CHECK(!s.enabled);
-    Statement::Initializer init(s);
+    static Statement::Initializer init(s);
     BOOST_CHECK(s.enabled);
 
-    Statement s2=QPID_LOG_STATEMENT_INIT(warning);
-    Statement::Initializer init2(s2);
+    static Statement s2=QPID_LOG_STATEMENT_INIT(warning);
+    static Statement::Initializer init2(s2);
     BOOST_CHECK(!s2.enabled);
 
     l.select(Selector(warning));
@@ -98,9 +94,10 @@ struct TestOutput : public Logger::Output {
     vector<string> msg;
     vector<Statement> stmt;
 
-    TestOutput() {
-        Logger::instance().output(qpid::make_auto_ptr<Logger::Output>(this));
+    TestOutput(Logger& l) {
+        l.output(std::auto_ptr<Logger::Output>(this));
     }
+                 
     void log(const Statement& s, const string& m) {
         msg.push_back(m);
         stmt.push_back(s);
@@ -111,10 +108,12 @@ struct TestOutput : public Logger::Output {
 using boost::assign::list_of;
 
 BOOST_AUTO_TEST_CASE(testLoggerOutput) {
-    Logger& l=clearLogger();
+    Logger l;
+    l.clear();
     l.select(Selector(debug));
     Statement s=QPID_LOG_STATEMENT_INIT(debug);
-    TestOutput* out=new TestOutput();
+
+    TestOutput* out=new TestOutput(l);
 
     // Verify message is output.
     l.log(s, "foo");
@@ -122,7 +121,7 @@ BOOST_AUTO_TEST_CASE(testLoggerOutput) {
     BOOST_CHECK_EQUAL(expect, out->msg);
 
     // Verify multiple outputs
-    TestOutput* out2=new TestOutput();
+    TestOutput* out2=new TestOutput(l);
     l.log(Statement(), "baz");
     expect.push_back("baz\n");
     BOOST_CHECK_EQUAL(expect, out->msg);
@@ -131,9 +130,10 @@ BOOST_AUTO_TEST_CASE(testLoggerOutput) {
 }
 
 BOOST_AUTO_TEST_CASE(testMacro) {
-    Logger& l = clearLogger();
+    Logger& l=Logger::instance();
+    l.clear();
     l.select(Selector(info));
-    TestOutput* out=new TestOutput();
+    TestOutput* out=new TestOutput(l);
     QPID_LOG(info, "foo");
     vector<string> expect=list_of("foo\n");
     BOOST_CHECK_EQUAL(expect, out->msg);
@@ -150,9 +150,9 @@ BOOST_AUTO_TEST_CASE(testMacro) {
 }
 
 BOOST_AUTO_TEST_CASE(testLoggerFormat) {
-    Logger& l=clearLogger();
+    Logger& l = Logger::instance();
     l.select(Selector(critical));
-    TestOutput* out=new TestOutput();
+    TestOutput* out=new TestOutput(l);
 
     // Time format is YYY-Month-dd hh:mm:ss
     l.format(Logger::TIME);
@@ -183,7 +183,8 @@ BOOST_AUTO_TEST_CASE(testLoggerFormat) {
 }
 
 BOOST_AUTO_TEST_CASE(testOstreamOutput) {
-    Logger& l=clearLogger();
+    Logger& l=Logger::instance();
+    l.clear();
     l.select(Selector(error));
     ostringstream os;
     l.output(os);
@@ -191,12 +192,12 @@ BOOST_AUTO_TEST_CASE(testOstreamOutput) {
     QPID_LOG(error, "bar");
     QPID_LOG(error, "baz");
     BOOST_CHECK_EQUAL("foo\nbar\nbaz\n", os.str());
-    l.clear();
 }
 
 #if 0 // This test requires manual intervention. Normally disabled.
 BOOST_AUTO_TEST_CASE(testSyslogOutput) {
-    Logger& l = clearLogger();
+    Logger& l=Logger::instance();
+    l.clear();
     l.select(Selector(info));
     l.syslog("qpid_test");
     QPID_LOG(info, "Testing QPID");
@@ -312,7 +313,7 @@ BOOST_AUTO_TEST_CASE(testSelectorFromOptions) {
 }
 
 BOOST_AUTO_TEST_CASE(testOptionsFormat) {
-    Logger& l = clearLogger();
+    Logger l;
     {
         Options opts;
         BOOST_CHECK_EQUAL(Logger::TIME|Logger::LEVEL, l.format(opts));
@@ -344,7 +345,8 @@ BOOST_AUTO_TEST_CASE(testOptionsFormat) {
 }
 
 BOOST_AUTO_TEST_CASE(testLoggerConfigure) {
-    Logger& l = clearLogger();
+    Logger& l=Logger::instance();
+    l.clear();
     Options opts;
     char* argv[]={
         0,
diff --git a/cpp/src/tests/perftest.cpp b/cpp/src/tests/perftest.cpp
index d16ebd43de..bc816f6597 100644
--- a/cpp/src/tests/perftest.cpp
+++ b/cpp/src/tests/perftest.cpp
@@ -27,7 +27,6 @@
 #include "qpid/client/Connection.h"
 #include "qpid/client/MessageListener.h"
 #include "qpid/client/Message.h"
-#include "qpid/QpidError.h"
 #include "qpid/sys/Monitor.h"
 #include "qpid/sys/Time.h"
 
diff --git a/cpp/src/tests/run-unit-tests b/cpp/src/tests/run-unit-tests
index ce8f488b29..464ce131f5 100755
--- a/cpp/src/tests/run-unit-tests
+++ b/cpp/src/tests/run-unit-tests
@@ -24,5 +24,5 @@ test -z "$TEST_ARGS" && TEST_ARGS=".libs/*Test.so"
 test -z "$srcdir" && srcdir=.
 
 # libdlclose_noop prevents unloading symbols needed for valgrind output.
-LD_PRELOAD=.libs/libdlclose_noop.so exec $srcdir/test_env DllPlugInTester -c -b $TEST_ARGS
-
+export LD_PRELOAD=.libs/libdlclose_noop.so
+source $srcdir/run_test DllPlugInTester -c -b $TEST_ARGS
diff --git a/cpp/src/tests/topic_listener.cpp b/cpp/src/tests/topic_listener.cpp
index 9369b591a6..5aef16354e 100644
--- a/cpp/src/tests/topic_listener.cpp
+++ b/cpp/src/tests/topic_listener.cpp
@@ -32,7 +32,6 @@
  * listening).
  */
 
-#include "qpid/QpidError.h"
 #include "TestOptions.h"
 #include "qpid/client/Channel.h"
 #include "qpid/client/Connection.h"
diff --git a/cpp/src/tests/topic_publisher.cpp b/cpp/src/tests/topic_publisher.cpp
index 74fcf8b057..1c5b51309b 100644
--- a/cpp/src/tests/topic_publisher.cpp
+++ b/cpp/src/tests/topic_publisher.cpp
@@ -35,7 +35,6 @@
  */
 
 #include "TestOptions.h"
-#include "qpid/QpidError.h"
 #include "qpid/client/Channel.h"
 #include "qpid/client/Connection.h"
 #include "qpid/client/Exchange.h"