OutputControl and subclasses: added giveReadCredit() for IO level flow control.

Cluster: Set read credit limit for cluster connections.


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

5 files changed, 18 insertions, 4 deletions

diff --git a/cpp/src/qpid/cluster/Connection.cpp b/cpp/src/qpid/cluster/Connection.cpp
index 1276a994ac..27b391a1c7 100644
--- a/cpp/src/qpid/cluster/Connection.cpp
+++ b/cpp/src/qpid/cluster/Connection.cpp
@@ -62,17 +62,21 @@ Connection::Connection(Cluster& c, sys::ConnectionOutputHandler& out,
                        const std::string& wrappedId, ConnectionId myId)
     : cluster(c), self(myId), catchUp(false), output(*this, out),
       connection(&output, cluster.getBroker(), wrappedId)
-{
-    QPID_LOG(debug, cluster << " new connection: " << *this);
-}
+{ init(); }
 
 // Local connections
 Connection::Connection(Cluster& c, sys::ConnectionOutputHandler& out,
                        const std::string& wrappedId, MemberId myId, bool isCatchUp)
     : cluster(c), self(myId, this), catchUp(isCatchUp), output(*this, out),
       connection(&output, cluster.getBroker(), wrappedId)
-{
+{ init(); }
+
+void Connection::init() {
     QPID_LOG(debug, cluster << " new connection: " << *this);
+    if (isLocal() && !isCatchUp()) {
+        // FIXME aconway 2008-12-05: configurable credit limit
+        output.giveReadCredit(3);
+    }
 }
 
 Connection::~Connection() {
@@ -187,6 +191,7 @@ size_t Connection::decode(const char* buffer, size_t size) {
         Buffer buf(const_cast<char*>(buffer), size);
         while (localDecoder.decode(buf))
             received(localDecoder.frame);
+        output.giveReadCredit(1);
     }
     else {                      // Multicast local connections.
         assert(isLocal());
@@ -200,6 +205,8 @@ void Connection::deliverBuffer(Buffer& buf) {
     ++deliverSeq;
     while (mcastDecoder.decode(buf))
         delivered(mcastDecoder.frame);
+    if (isLocal()) 
+        output.giveReadCredit(1);
 }
 
 broker::SessionState& Connection::sessionState() {
diff --git a/cpp/src/qpid/cluster/Connection.h b/cpp/src/qpid/cluster/Connection.h
index 36476baa34..ce80f42414 100644
--- a/cpp/src/qpid/cluster/Connection.h
+++ b/cpp/src/qpid/cluster/Connection.h
@@ -142,6 +142,7 @@ class Connection :
     void exchange(const std::string& encoded);
     
   private:
+    void init();
     bool checkUnsupported(const framing::AMQBody& body);
     void deliverClose();
     void deliverDoOutput(uint32_t requested);
@@ -167,6 +168,8 @@ class Connection :
     framing::ChannelId currentChannel;
     boost::shared_ptr<broker::TxBuffer> txBuffer;
     
+    int FIXMEcredit;            // FIXME aconway 2008-12-05: remove
+    
   friend std::ostream& operator<<(std::ostream&, const Connection&);
 };
 
diff --git a/cpp/src/qpid/cluster/NoOpConnectionOutputHandler.h b/cpp/src/qpid/cluster/NoOpConnectionOutputHandler.h
index 3c24dd71f2..74a376a657 100644
--- a/cpp/src/qpid/cluster/NoOpConnectionOutputHandler.h
+++ b/cpp/src/qpid/cluster/NoOpConnectionOutputHandler.h
@@ -39,6 +39,7 @@ class NoOpConnectionOutputHandler : public sys::ConnectionOutputHandler
     virtual void send(framing::AMQFrame&) {}
     virtual void close() {}
     virtual void activateOutput() {}
+    virtual void giveReadCredit(int32_t) {}
 };
 
 }} // namespace qpid::cluster
diff --git a/cpp/src/qpid/cluster/OutputInterceptor.cpp b/cpp/src/qpid/cluster/OutputInterceptor.cpp
index 0563899a1c..77ed154bd0 100644
--- a/cpp/src/qpid/cluster/OutputInterceptor.cpp
+++ b/cpp/src/qpid/cluster/OutputInterceptor.cpp
@@ -55,6 +55,8 @@ void OutputInterceptor::activateOutput() {
     }
 }
 
+void OutputInterceptor::giveReadCredit(int32_t credit) { next->giveReadCredit(credit); }
+
 // Called in write thread when the IO layer has no more data to write.
 // We do nothing in the write thread, we run doOutput only on delivery
 // of doOutput requests.
diff --git a/cpp/src/qpid/cluster/OutputInterceptor.h b/cpp/src/qpid/cluster/OutputInterceptor.h
index f4226d3a40..783a443228 100644
--- a/cpp/src/qpid/cluster/OutputInterceptor.h
+++ b/cpp/src/qpid/cluster/OutputInterceptor.h
@@ -43,6 +43,7 @@ class OutputInterceptor : public sys::ConnectionOutputHandler {
     // sys::ConnectionOutputHandler functions
     void send(framing::AMQFrame& f);
     void activateOutput();
+    void giveReadCredit(int32_t);
     void close();
     size_t getBuffered() const;