qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From cctriel...@apache.org
Subject svn commit: r739378 - in /qpid/trunk/qpid/cpp/src: qpid/broker/Message.cpp qpid/broker/Message.h qpid/client/Connection.cpp qpid/client/Connection.h qpid/client/SessionImpl.cpp qpid/client/SessionImpl.h qpid/cluster/UpdateClient.cpp tests/consume.cpp
Date Fri, 30 Jan 2009 18:59:24 GMT
Author: cctrieloff
Date: Fri Jan 30 18:59:24 2009
New Revision: 739378

URL: http://svn.apache.org/viewvc?rev=739378&view=rev
Log:
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.

Modified:
    qpid/trunk/qpid/cpp/src/qpid/broker/Message.cpp
    qpid/trunk/qpid/cpp/src/qpid/broker/Message.h
    qpid/trunk/qpid/cpp/src/qpid/client/Connection.cpp
    qpid/trunk/qpid/cpp/src/qpid/client/Connection.h
    qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp
    qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.h
    qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.cpp
    qpid/trunk/qpid/cpp/src/tests/consume.cpp

Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Message.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Message.cpp?rev=739378&r1=739377&r2=739378&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Message.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Message.cpp Fri Jan 30 18:59:24 2009
@@ -197,30 +197,39 @@
     }
 }
 
-void Message::sendContent(Queue& queue, framing::FrameHandler& out, uint16_t maxFrameSize)
const
+bool Message::getContentFrame(const Queue& queue, AMQFrame& frame, uint16_t maxContentSize,
uint64_t offset) const
 {
     if (isContentReleased()) {
-        //load content from store in chunks of maxContentSize
-        uint16_t maxContentSize = maxFrameSize - AMQFrame::frameOverhead();
         intrusive_ptr<const PersistableMessage> pmsg(this);
         
         bool done = false;
-        for (uint64_t offset = 0; !done; offset += maxContentSize)
+        string& data = frame.castBody<AMQContentBody>()->getData();
+        store->loadContent(queue, pmsg, data, offset, maxContentSize);
+        done = data.size() < maxContentSize;
+        frame.setBof(false);
+        frame.setEof(true);
+        QPID_LOG(debug, "loaded frame" << frame);
+        if (offset > 0) {
+            frame.setBos(false);
+        }
+        if (!done) {
+            frame.setEos(false);
+        } else return false;
+        return true;
+    }
+    else return false;
+}
+
+void Message::sendContent(const Queue& queue, framing::FrameHandler& out, uint16_t
maxFrameSize) const
+{
+    if (isContentReleased()) {
+
+        uint16_t maxContentSize = maxFrameSize - AMQFrame::frameOverhead();
+        bool morecontent = true;
+        for (uint64_t offset = 0; morecontent; offset += maxContentSize)
         {            
             AMQFrame frame((AMQContentBody()));
-            string& data = frame.castBody<AMQContentBody>()->getData();
-
-            store->loadContent(queue, pmsg, data, offset, maxContentSize);
-            done = data.size() < maxContentSize;
-            frame.setBof(false);
-            frame.setEof(true);
-            if (offset > 0) {
-                frame.setBos(false);
-            }
-            if (!done) {
-                frame.setEos(false);
-            }
-            QPID_LOG(debug, "loaded frame for delivery: " << frame);
+            morecontent = getContentFrame(queue, frame, maxContentSize, offset);
             out.handle(frame);
         }
     } else {

Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Message.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Message.h?rev=739378&r1=739377&r2=739378&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Message.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Message.h Fri Jan 30 18:59:24 2009
@@ -131,7 +131,8 @@
     void releaseContent(MessageStore* store);
     void destroy();
 
-    void sendContent(Queue& queue, framing::FrameHandler& out, uint16_t maxFrameSize)
const;
+    bool getContentFrame(const Queue& queue, framing::AMQFrame& frame, uint16_t maxContentSize,
uint64_t offset) const;
+    void sendContent(const Queue& queue, framing::FrameHandler& out, uint16_t maxFrameSize)
const;
     void sendHeader(framing::FrameHandler& out, uint16_t maxFrameSize) const;
 
     bool isContentLoaded() const;

Modified: qpid/trunk/qpid/cpp/src/qpid/client/Connection.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/Connection.cpp?rev=739378&r1=739377&r2=739378&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/Connection.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/Connection.cpp Fri Jan 30 18:59:24 2009
@@ -119,6 +119,13 @@
         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"));

Modified: qpid/trunk/qpid/cpp/src/qpid/client/Connection.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/Connection.h?rev=739378&r1=739377&r2=739378&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/Connection.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/Connection.h Fri Jan 30 18:59:24 2009
@@ -175,6 +175,11 @@
     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
 };

Modified: qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp?rev=739378&r1=739377&r2=739378&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp Fri Jan 30 18:59:24 2009
@@ -322,6 +322,11 @@
     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 @@
         connectionShared = connectionWeak.lock();
 }
 
+shared_ptr<ConnectionImpl> SessionImpl::getConnection()
+{
+    return connectionWeak.lock();
+}
+
 }}

Modified: qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.h?rev=739378&r1=739377&r2=739378&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.h Fri Jan 30 18:59:24 2009
@@ -87,6 +87,7 @@
     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 @@
      */
     void setWeakPtr(bool weak=true);
 
+    /** 
+     * get the Connection associated with this connection
+     */
+    shared_ptr<ConnectionImpl> getConnection();
+
 private:
     enum State {
         INACTIVE,
@@ -204,7 +210,6 @@
     const uint64_t maxFrameSize;
     const SessionId id;
 
-    shared_ptr<ConnectionImpl> connection();
     shared_ptr<ConnectionImpl> connectionShared;
     boost::weak_ptr<ConnectionImpl> connectionWeak;
     bool weakPtr;

Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.cpp?rev=739378&r1=739377&r2=739378&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.cpp Fri Jan 30 18:59:24 2009
@@ -168,6 +168,7 @@
         session.exchangeUnbind(queue, UpdateClient::UPDATE);
     }
 
+
     void updateQueuedMessage(const broker::QueuedMessage& message) {
         if (!haveLastPos || message.position - lastPos != 1)  {
             ClusterConnectionProxy(session).queuePosition(queue, message.position.getValue()-1);
@@ -177,12 +178,27 @@
         SessionBase_0_10Access sb(session);
         framing::MessageTransferBody transfer(
             framing::ProtocolVersion(), UpdateClient::UPDATE, message::ACCEPT_MODE_NONE,
message::ACQUIRE_MODE_PRE_ACQUIRED);
+        
         sb.get()->send(transfer, message.payload->getFrames());
+        if (message.payload->isContentReleased()){
+            uint16_t maxFrameSize = sb.get()->getConnection()->getNegotiatedSettings().maxFrameSize;
+
+            uint16_t maxContentSize = maxFrameSize - AMQFrame::frameOverhead();
+            bool morecontent = true;
+            for (uint64_t offset = 0; morecontent; offset += maxContentSize)
+            {            
+                AMQFrame frame((AMQContentBody()));
+                morecontent = message.payload->getContentFrame(*(message.queue), frame,
maxContentSize, offset);
+                sb.get()->sendRawFrame(frame);
+            }
+        }
     }
 
     void updateMessage(const boost::intrusive_ptr<broker::Message>& message) {
         updateQueuedMessage(broker::QueuedMessage(0, message, haveLastPos? lastPos.getValue()+1
: 1));
     }
+    
+   
 };
 
 

Modified: qpid/trunk/qpid/cpp/src/tests/consume.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/consume.cpp?rev=739378&r1=739377&r2=739378&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/consume.cpp (original)
+++ qpid/trunk/qpid/cpp/src/tests/consume.cpp Fri Jan 30 18:59:24 2009
@@ -44,15 +44,19 @@
     string queue;
     bool declare;
     bool summary;
+    bool print;
+    bool durable;
     
     Args() : count(1000), ack(0), queue("publish-consume"),
-             declare(false), summary(false)
+             declare(false), summary(false), print(false)
     {
         addOptions()
             ("count", optValue(count, "N"), "number of messages to publish")
             ("ack-frequency", optValue(ack, "N"), "ack every N messages (0 means use no-ack
mode)")
             ("queue", optValue(queue, "<queue name>"), "queue to consume from")
             ("declare", optValue(declare), "declare the queue")
+            ("durable", optValue(durable), "declare the queue durable, use with declare")
+            ("print-data", optValue(print), "Print the recieved data at info level")
             ("s,summary", optValue(summary), "Print undecorated rate.");
     }
 };
@@ -73,7 +77,7 @@
     void consume()
     {
         if (opts.declare)
-            session.queueDeclare(opts.queue);
+            session.queueDeclare(arg::queue=opts.queue, arg::durable=opts.durable);
         SubscriptionManager subs(session);
         LocalQueue lq;
         SubscriptionSettings settings;
@@ -85,6 +89,7 @@
         for (size_t i = 0; i < opts.count; ++i) {
             msg=lq.pop();
             QPID_LOG(info, "Received: " << msg.getMessageProperties().getCorrelationId());
+            if (opts.print) QPID_LOG(info, "Data: " << msg.getData());
         }
         if (opts.ack != 0)
             sub.accept(sub.getUnaccepted()); // Cumulative ack for final batch.



---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project:      http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org


Mime
View raw message