1 files changed, 105 insertions, 0 deletions
diff --git a/qpid/cpp/src/qpid/broker/RateFlowcontrol.h b/qpid/cpp/src/qpid/broker/RateFlowcontrol.h
new file mode 100644
index 0000000000..99f9d2c0c4
--- /dev/null
+++ b/qpid/cpp/src/qpid/broker/RateFlowcontrol.h
@@ -0,0 +1,105 @@
+#ifndef broker_RateFlowcontrol_h
+#define broker_RateFlowcontrol_h
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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/sys/Time.h"
+#include "qpid/sys/IntegerTypes.h"
+
+#include <algorithm>
+
+namespace qpid {
+namespace broker {
+
+// Class to keep track of issuing flow control to make sure that the peer doesn't exceed
+// a given message rate
+//
+// Create the object with the target rate
+// Then call sendCredit() whenever credit is issued to the peer
+// Call receivedMessage() whenever a message is received, it returns the credit to issue.
+// 
+// sentCredit() be sensibly called with a 0 parameter to indicate
+// that we sent credit but treat it as if the value was 0 (we may do this at the start of the connection
+// to allow our peer to send messages)
+//
+// receivedMessage() can be called with 0 to indicate that we've not received a message, but
+// tell me what credit I can send.
+class RateFlowcontrol {
+    uint32_t rate; // messages per second
+    uint32_t maxCredit; // max credit issued to client (issued at start)
+    uint32_t requestedCredit;
+    qpid::sys::AbsTime creditSent;
+   
+public:
+    RateFlowcontrol(uint32_t r) :
+         rate(r),
+         maxCredit(0),
+         requestedCredit(0),
+         creditSent(qpid::sys::FAR_FUTURE)
+    {}
+
+    uint32_t getRate() const {
+        return rate;
+    }
+    void sentCredit(const qpid::sys::AbsTime& t, uint32_t credit);
+    uint32_t receivedMessage(const qpid::sys::AbsTime& t, uint32_t  msgs);
+    uint32_t availableCredit(const qpid::sys::AbsTime& t);
+    bool flowStopped() const;
+};
+
+inline void RateFlowcontrol::sentCredit(const qpid::sys::AbsTime& t, uint32_t credit) {
+    // If the client isn't currently requesting credit (ie it's not sent us anything yet) then
+    // this credit goes to the max credit held by the client (it can't go to reduce credit
+    // less than 0)
+    int32_t nextRequestedCredit = requestedCredit - credit;
+    if ( nextRequestedCredit<0 ) {
+        requestedCredit = 0;
+        maxCredit -= nextRequestedCredit;
+    } else {
+        requestedCredit = nextRequestedCredit;
+    }
+    creditSent = t;
+}
+
+inline uint32_t RateFlowcontrol::availableCredit(const qpid::sys::AbsTime& t) {
+    qpid::sys::Duration d(creditSent, t);
+    // Could be -ve before first sentCredit
+    int64_t toSend = std::min(rate * d / qpid::sys::TIME_SEC, static_cast<int64_t>(requestedCredit));
+    return toSend > 0 ? toSend : 0;
+}
+
+inline uint32_t RateFlowcontrol::receivedMessage(const qpid::sys::AbsTime& t, uint32_t  msgs) {
+    requestedCredit +=msgs;
+    // Don't send credit for every message, only send if more than 0.5s since last credit or
+    // we've got less than .25 of the max left (heuristic)
+    return requestedCredit*4 >= maxCredit*3 || qpid::sys::Duration(creditSent, t) >= 500*qpid::sys::TIME_MSEC
+        ? availableCredit(t)
+        : 0;
+}
+
+inline bool RateFlowcontrol::flowStopped() const {
+    return requestedCredit >= maxCredit;
+}
+
+}}
+
+#endif // broker_RateFlowcontrol_h