Start execution mark from -1 (0xFFFFFFFF)

Rename ackFrequency as ackBatchSize in Dispatcher



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

7 files changed, 16 insertions, 12 deletions

diff --git a/cpp/rubygen/templates/Session.rb b/cpp/rubygen/templates/Session.rb
index b2932f9a2b..d2b4a3233d 100644
--- a/cpp/rubygen/templates/Session.rb
+++ b/cpp/rubygen/templates/Session.rb
@@ -131,6 +131,7 @@ class SessionGen < CppGen
 #include "qpid/framing/ProtocolVersion.h"
 #include "qpid/framing/MethodContent.h"
 #include "qpid/framing/TransferContent.h"
+#include "qpid/client/Completion.h"
 #include "qpid/client/ConnectionImpl.h"
 #include "qpid/client/Response.h"
 #include "qpid/client/ScopedAssociation.h"
diff --git a/cpp/src/qpid/broker/DeliveryRecord.cpp b/cpp/src/qpid/broker/DeliveryRecord.cpp
index f9218655a4..df700cd15a 100644
--- a/cpp/src/qpid/broker/DeliveryRecord.cpp
+++ b/cpp/src/qpid/broker/DeliveryRecord.cpp
@@ -143,6 +143,8 @@ void DeliveryRecord::acquire(std::vector<DeliveryId>& results) {
     if (queue->acquire(msg)) {
         acquired = true;
         results.push_back(id);
+    } else {
+        QPID_LOG(info, "Message already acquired " << id.getValue());
     }
 }
 
diff --git a/cpp/src/qpid/client/BlockingQueue.h b/cpp/src/qpid/client/BlockingQueue.h
index a9d8ec2857..3ec7dbcf44 100644
--- a/cpp/src/qpid/client/BlockingQueue.h
+++ b/cpp/src/qpid/client/BlockingQueue.h
@@ -36,7 +36,7 @@ class BlockingQueue
     sys::Monitor lock;
     std::queue<T> queue;
     bool closed;
- 
+
 public:
 
     BlockingQueue() : closed(false) {}
diff --git a/cpp/src/qpid/client/Dispatcher.cpp b/cpp/src/qpid/client/Dispatcher.cpp
index fd437725ce..5c3c85343e 100644
--- a/cpp/src/qpid/client/Dispatcher.cpp
+++ b/cpp/src/qpid/client/Dispatcher.cpp
@@ -36,14 +36,14 @@ using qpid::sys::Thread;
 namespace qpid {
 namespace client {
 
-    Subscriber::Subscriber(Session& s, MessageListener* l, bool a, uint f) : session(s), listener(l), autoAck(a), ackFrequency(f), count(0) {}
+    Subscriber::Subscriber(Session& s, MessageListener* l, bool a, uint f) : session(s), listener(l), autoAck(a), ackBatchSize(f), count(0) {}
 
 void Subscriber::received(Message& msg)
 {
     if (listener) {
         listener->received(msg);
         if (autoAck) {
-            bool send = (++count >= ackFrequency);
+            bool send = (++count >= ackBatchSize);
             msg.acknowledge(session, true, send);
             if (send) count = 0;
         }
@@ -129,16 +129,16 @@ Subscriber::shared_ptr Dispatcher::find(const std::string& name)
     return i->second;
 }
 
-void Dispatcher::listen(MessageListener* listener, bool autoAck, uint ackFrequency)
+void Dispatcher::listen(MessageListener* listener, bool autoAck, uint ackBatchSize)
 {
     ScopedLock<Mutex> l(lock);
-    defaultListener = Subscriber::shared_ptr(new Subscriber(session, listener, autoAck, ackFrequency));
+    defaultListener = Subscriber::shared_ptr(new Subscriber(session, listener, autoAck, ackBatchSize));
 }
 
-void Dispatcher::listen(const std::string& destination, MessageListener* listener, bool autoAck, uint ackFrequency)
+void Dispatcher::listen(const std::string& destination, MessageListener* listener, bool autoAck, uint ackBatchSize)
 {
     ScopedLock<Mutex> l(lock);
-    listeners[destination] = Subscriber::shared_ptr(new Subscriber(session, listener, autoAck, ackFrequency));
+    listeners[destination] = Subscriber::shared_ptr(new Subscriber(session, listener, autoAck, ackBatchSize));
 }
 
 void Dispatcher::cancel(const std::string& destination)
diff --git a/cpp/src/qpid/client/Dispatcher.h b/cpp/src/qpid/client/Dispatcher.h
index e4a4cec4a6..1d1d9141e3 100644
--- a/cpp/src/qpid/client/Dispatcher.h
+++ b/cpp/src/qpid/client/Dispatcher.h
@@ -40,12 +40,12 @@ class Subscriber : public MessageListener
     Session& session;
     MessageListener* const listener;
     const bool autoAck;
-    const uint ackFrequency;
+    const uint ackBatchSize;
     uint count;
 
 public:
     typedef boost::shared_ptr<Subscriber> shared_ptr;
-    Subscriber(Session& session, MessageListener* listener, bool autoAck = true, uint ackFrequency = 1);
+    Subscriber(Session& session, MessageListener* listener, bool autoAck = true, uint ackBatchSize = 1);
     void received(Message& msg);
     
 };
@@ -77,8 +77,8 @@ public:
     void run();
     void stop();
 
-    void listen(MessageListener* listener, bool autoAck = true, uint ackFrequency = 1);
-    void listen(const std::string& destination, MessageListener* listener, bool autoAck = true, uint ackFrequency = 1);
+    void listen(MessageListener* listener, bool autoAck = true, uint ackBatchSize = 1);
+    void listen(const std::string& destination, MessageListener* listener, bool autoAck = true, uint ackBatchSize = 1);
     void cancel(const std::string& destination);
 };
 
diff --git a/cpp/src/qpid/client/SessionCore.cpp b/cpp/src/qpid/client/SessionCore.cpp
index 82b3e77b94..8bad6ec374 100644
--- a/cpp/src/qpid/client/SessionCore.cpp
+++ b/cpp/src/qpid/client/SessionCore.cpp
@@ -92,6 +92,7 @@ void SessionCore::closed(uint16_t code, const std::string& text)
 void SessionCore::checkClosed()
 {
     if (isClosed) {
+        //TODO: could actually have been a connection exception
         throw ChannelException(reason.code, reason.text);
     }
 }
diff --git a/cpp/src/qpid/framing/SequenceNumber.cpp b/cpp/src/qpid/framing/SequenceNumber.cpp
index 24867130a2..3172246cc2 100644
--- a/cpp/src/qpid/framing/SequenceNumber.cpp
+++ b/cpp/src/qpid/framing/SequenceNumber.cpp
@@ -23,7 +23,7 @@
 
 using qpid::framing::SequenceNumber;
 
-SequenceNumber::SequenceNumber() : value(0) {}
+SequenceNumber::SequenceNumber() : value(0 - 1) {}
 
 SequenceNumber::SequenceNumber(uint32_t v) : value((int32_t) v) {}