Additions to the client API:

 - SubscriptionManager::get(queue) to get a single message from a queue.
 - Set FlowControl per-subscription.


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

4 files changed, 157 insertions, 15 deletions

diff --git a/cpp/src/qpid/client/FlowControl.h b/cpp/src/qpid/client/FlowControl.h
new file mode 100644
index 0000000000..a4ed9879f4
--- /dev/null
+++ b/cpp/src/qpid/client/FlowControl.h
@@ -0,0 +1,73 @@
+#ifndef QPID_CLIENT_FLOWCONTROL_H
+#define QPID_CLIENT_FLOWCONTROL_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.
+ *
+ */
+
+namespace qpid {
+namespace client {
+
+/**
+ * Flow control works by associating a finite amount of "credit"
+ * associated with a subscription.
+ *
+ * Credit includes a message count and a byte count. Each message
+ * received decreases the message count by one, and the byte count by
+ * the size of the message. Either count can have the special value
+ * UNLIMITED which is never decreased.
+ *
+ * A subscription's credit is exhausted when the message count is 0 or
+ * the byte count is too small for the next available message. The
+ * subscription will not receive any further messages until is credit
+ * is renewed.
+ *
+ * In "window mode" credit is automatically renewed when a message is
+ * acknowledged (@see AckPolicy) In non-window mode credit is not
+ * automatically renewed, it must be explicitly re-set (@see
+ * SubscriptionManager)
+ */
+struct FlowControl {
+    static const uint32_t UNLIMITED=0xFFFFFFFF;
+    FlowControl(uint32_t messages_=0, uint32_t bytes_=0, bool window_=false)
+        : messages(messages_), bytes(bytes_), window(window_) {}
+    
+    static FlowControl messageCredit(uint32_t messages_) { return FlowControl(messages_,UNLIMITED,false); }
+    static FlowControl messageWindow(uint32_t messages_) { return FlowControl(messages_,UNLIMITED,true); }
+    static FlowControl byteCredit(uint32_t bytes_) { return FlowControl(UNLIMITED,bytes_,false); }
+    static FlowControl byteWindow(uint32_t bytes_) { return FlowControl(UNLIMITED,bytes_,true); }
+    static FlowControl unlimited() { return FlowControl(UNLIMITED, UNLIMITED, false); }
+    static FlowControl zero() { return FlowControl(0, 0, false); }
+
+    /** Message credit: subscription can accept up to this many messages. */
+    uint32_t messages;
+    /** Byte credit: subscription can accept up to this many bytes of message content. */
+    uint32_t bytes;
+    /** Window mode. If true credit is automatically renewed as messages are acknowledged. */
+    bool window;
+
+    bool operator==(const FlowControl& x) {
+        return messages == x.messages && bytes == x.bytes && window == x.window;
+    };
+};
+
+}} // namespace qpid::client
+
+#endif  /*!QPID_CLIENT_FLOWCONTROL_H*/
diff --git a/cpp/src/qpid/client/LocalQueue.h b/cpp/src/qpid/client/LocalQueue.h
index c76d0756eb..273814f179 100644
--- a/cpp/src/qpid/client/LocalQueue.h
+++ b/cpp/src/qpid/client/LocalQueue.h
@@ -56,7 +56,7 @@ class LocalQueue
      */
     Message pop();
 
-    /** Synonym for get(). */
+    /** Synonym for pop(). */
     Message get() { return pop(); }
     
     /** Return true if local queue is empty. */
diff --git a/cpp/src/qpid/client/SubscriptionManager.cpp b/cpp/src/qpid/client/SubscriptionManager.cpp
index 3fa75a54ac..324b11e1df 100644
--- a/cpp/src/qpid/client/SubscriptionManager.cpp
+++ b/cpp/src/qpid/client/SubscriptionManager.cpp
@@ -25,6 +25,7 @@
 #include <qpid/client/Dispatcher.h>
 #include <qpid/client/Session.h>
 #include <qpid/client/MessageListener.h>
+#include <qpid/framing/Uuid.h>
 #include <set>
 #include <sstream>
 
@@ -34,35 +35,48 @@ namespace client {
 
 SubscriptionManager::SubscriptionManager(const Session& s)
     : dispatcher(s), session(s),
-      messages(UNLIMITED), bytes(UNLIMITED), window(true),
+      flowControl(UNLIMITED, UNLIMITED, false),
       acceptMode(0), acquireMode(0),
       autoStop(true)
 {}
 
 void SubscriptionManager::subscribeInternal(
-    const std::string& q, const std::string& dest)
+    const std::string& q, const std::string& dest, const FlowControl& fc)
 {
     session.messageSubscribe( 
         arg::queue=q, arg::destination=dest,
         arg::acceptMode=acceptMode, arg::acquireMode=acquireMode);
-    setFlowControl(dest, messages, bytes, window); 
+    if (fc.messages || fc.bytes) // No need to set if all 0.
+        setFlowControl(dest, fc);
 }
 
 void SubscriptionManager::subscribe(
     MessageListener& listener, const std::string& q, const std::string& d)
 {
+    subscribe(listener, q, getFlowControl(), d);
+}
+
+void SubscriptionManager::subscribe(
+    MessageListener& listener, const std::string& q, const FlowControl& fc, const std::string& d)
+{
     std::string dest=d.empty() ? q:d;
     dispatcher.listen(dest, &listener, autoAck);
-    return subscribeInternal(q, dest);
+    return subscribeInternal(q, dest, fc);
 }
 
 void SubscriptionManager::subscribe(
     LocalQueue& lq, const std::string& q, const std::string& d)
 {
+    subscribe(lq, q, getFlowControl(), d);
+}
+
+void SubscriptionManager::subscribe(
+    LocalQueue& lq, const std::string& q, const FlowControl& fc, const std::string& d)
+{
     std::string dest=d.empty() ? q:d;
     lq.session=session;
     lq.queue=session.getExecution().getDemux().add(dest, ByTransferDest(dest));
-    return subscribeInternal(q, dest);
+    return subscribeInternal(q, dest, fc);
 }
 
 void SubscriptionManager::setFlowControl(
@@ -74,14 +88,20 @@ void SubscriptionManager::setFlowControl(
     session.sync();
 }
 
+void SubscriptionManager::setFlowControl(const std::string& dest, const FlowControl& fc) {
+    setFlowControl(dest, fc.messages, fc.bytes, fc.window);
+}
+
+void SubscriptionManager::setFlowControl(const FlowControl& fc) { flowControl=fc; }
+
 void SubscriptionManager::setFlowControl(
     uint32_t messages_,  uint32_t bytes_, bool window_)
 {
-    messages=messages_;
-    bytes=bytes_;
-    window=window_;
+    setFlowControl(FlowControl(messages_, bytes_, window_));
 }
 
+const FlowControl& SubscriptionManager::getFlowControl() const { return flowControl; }
+
 void SubscriptionManager::setAcceptMode(bool c) { acceptMode=c; }
 
 void SubscriptionManager::setAcquireMode(bool a) { acquireMode=a; }
@@ -109,6 +129,12 @@ void SubscriptionManager::stop()
     dispatcher.stop();
 }
 
+Message SubscriptionManager::get(const std::string& queue) {
+    LocalQueue lq;
+    subscribe(lq, queue, FlowControl::messageCredit(1), framing::Uuid(true).str());
+    return lq.get();
+}
+
 }} // namespace qpid::client
 
 #endif
diff --git a/cpp/src/qpid/client/SubscriptionManager.h b/cpp/src/qpid/client/SubscriptionManager.h
index 930175564e..0aa55099f5 100644
--- a/cpp/src/qpid/client/SubscriptionManager.h
+++ b/cpp/src/qpid/client/SubscriptionManager.h
@@ -27,8 +27,8 @@
 #include <qpid/client/Session.h>
 #include <qpid/client/MessageListener.h>
 #include <qpid/client/LocalQueue.h>
+#include <qpid/client/FlowControl.h>
 #include <qpid/sys/Runnable.h>
-
 #include <set>
 #include <sstream>
 
@@ -48,13 +48,11 @@ class SubscriptionManager : public sys::Runnable
     typedef sys::Mutex::ScopedLock Lock;
     typedef sys::Mutex::ScopedUnlock Unlock;
 
-    void subscribeInternal(const std::string& q, const std::string& dest);
+    void subscribeInternal(const std::string& q, const std::string& dest, const FlowControl&);
     
     qpid::client::Dispatcher dispatcher;
     qpid::client::AsyncSession session;
-    uint32_t messages;
-    uint32_t bytes;
-    bool window;
+    FlowControl flowControl;
     AckPolicy autoAck;
     bool acceptMode;
     bool acquireMode;
@@ -72,6 +70,38 @@ class SubscriptionManager : public sys::Runnable
      * 
      *@param listener Listener object to receive messages.
      *@param queue Name of the queue to subscribe to.
+     *@param flow initial FlowControl for the subscription.
+     *@param tag Unique destination tag for the listener.
+     * If not specified, the queue name is used.
+     */
+    void subscribe(MessageListener& listener,
+                   const std::string& queue,
+                   const FlowControl& flow,
+                   const std::string& tag=std::string());
+
+    /**
+     * Subscribe a LocalQueue to receive messages from queue.
+     * 
+     * Incoming messages are stored in the queue for you to retrieve.
+     * 
+     *@param queue Name of the queue to subscribe to.
+     *@param flow initial FlowControl for the subscription.
+     *@param tag Unique destination tag for the listener.
+     * If not specified, the queue name is used.
+     */
+    void subscribe(LocalQueue& localQueue,
+                   const std::string& queue,
+                   const FlowControl& flow,
+                   const std::string& tag=std::string());
+
+    /**
+     * Subscribe a MessagesListener to receive messages from queue.
+     *
+     * Provide your own subclass of MessagesListener to process
+     * incoming messages. It will be called for each message received.
+     * 
+     *@param listener Listener object to receive messages.
+     *@param queue Name of the queue to subscribe to.
      *@param tag Unique destination tag for the listener.
      * If not specified, the queue name is used.
      */
@@ -92,6 +122,11 @@ class SubscriptionManager : public sys::Runnable
                    const std::string& queue,
                    const std::string& tag=std::string());
 
+    /**
+     * Get a single message from a queue.
+     */
+    Message get(const std::string& queue);
+
     /** Cancel a subscription. */
     void cancel(const std::string tag);
 
@@ -107,9 +142,17 @@ class SubscriptionManager : public sys::Runnable
     /** Cause run() to return */
     void stop();
 
-
     static const uint32_t UNLIMITED=0xFFFFFFFF;
 
+    /** Set the flow control for destination. */
+    void setFlowControl(const std::string& destintion, const FlowControl& flow);
+
+    /** Set the default initial flow control for subscriptions that do not specify it. */
+    void setFlowControl(const FlowControl& flow);
+
+    /** Get the default flow control for new subscriptions that do not specify it. */
+    const FlowControl& getFlowControl() const;
+
     /** Set the flow control for destination tag.
      *@param tag: name of the destination.
      *@param messages: message credit.