Correction for: start a broker in cluster, send messages that are flow to disk, then join a broker to the cluster. Then consume from the new node. Cotent released messages where loosing content. This patch corrects that.

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

4 files changed, 28 insertions, 1 deletions

diff --git a/cpp/src/qpid/client/Connection.cpp b/cpp/src/qpid/client/Connection.cpp
index f450344aa7..3562df4a74 100644
--- a/cpp/src/qpid/client/Connection.cpp
+++ b/cpp/src/qpid/client/Connection.cpp
@@ -119,6 +119,13 @@ void Connection::open(const ConnectionSettings& settings)
         impl->registerFailureCallback ( failureCallback );
 }
 
+const ConnectionSettings& Connection::getNegotiatedSettings()
+{
+    if (!isOpen())
+        throw Exception(QPID_MSG("Connection is not open."));
+     return impl->getNegotiatedSettings();
+}
+
 Session Connection::newSession(const std::string& name, uint32_t timeout) {
     if (!isOpen())
         throw Exception(QPID_MSG("Connection has not yet been opened"));
diff --git a/cpp/src/qpid/client/Connection.h b/cpp/src/qpid/client/Connection.h
index d03542bb5b..ed984ccb42 100644
--- a/cpp/src/qpid/client/Connection.h
+++ b/cpp/src/qpid/client/Connection.h
@@ -175,6 +175,11 @@ class Connection
     std::vector<Url> getKnownBrokers();
     void registerFailureCallback ( boost::function<void ()> fn );
 
+    /**
+     * Return the set of client negotiated settings
+     */
+    const ConnectionSettings& getNegotiatedSettings();
+
   friend class ConnectionAccess; ///<@internal
   friend class SessionBase_0_10; ///<@internal
 };
diff --git a/cpp/src/qpid/client/SessionImpl.cpp b/cpp/src/qpid/client/SessionImpl.cpp
index 7cf68956ea..c179a31853 100644
--- a/cpp/src/qpid/client/SessionImpl.cpp
+++ b/cpp/src/qpid/client/SessionImpl.cpp
@@ -322,6 +322,11 @@ Future SessionImpl::send(const AMQBody& command, const FrameSet& content) {
     return f;
 }
 
+void SessionImpl::sendRawFrame(AMQFrame& frame) {
+    Acquire a(sendLock);
+    handleOut(frame);
+}
+
 Future SessionImpl::sendCommand(const AMQBody& command, const MethodContent* content)
 {
     Acquire a(sendLock);
@@ -767,4 +772,9 @@ void SessionImpl::setWeakPtr(bool weak) {
         connectionShared = connectionWeak.lock();
 }
 
+shared_ptr<ConnectionImpl> SessionImpl::getConnection()
+{
+    return connectionWeak.lock();
+}
+
 }}
diff --git a/cpp/src/qpid/client/SessionImpl.h b/cpp/src/qpid/client/SessionImpl.h
index 9d0c4ff796..d826b759ae 100644
--- a/cpp/src/qpid/client/SessionImpl.h
+++ b/cpp/src/qpid/client/SessionImpl.h
@@ -87,6 +87,7 @@ public:
     Future send(const framing::AMQBody& command);
     Future send(const framing::AMQBody& command, const framing::MethodContent& content);
     Future send(const framing::AMQBody& command, const framing::FrameSet& content);
+    void sendRawFrame(framing::AMQFrame& frame);
 
     Demux& getDemux();
     void markCompleted(const framing::SequenceNumber& id, bool cumulative, bool notifyPeer);
@@ -114,6 +115,11 @@ public:
      */
     void setWeakPtr(bool weak=true);
 
+    /** 
+     * get the Connection associated with this connection
+     */
+    shared_ptr<ConnectionImpl> getConnection();
+
 private:
     enum State {
         INACTIVE,
@@ -204,7 +210,6 @@ private:
     const uint64_t maxFrameSize;
     const SessionId id;
 
-    shared_ptr<ConnectionImpl> connection();
     shared_ptr<ConnectionImpl> connectionShared;
     boost::weak_ptr<ConnectionImpl> connectionWeak;
     bool weakPtr;