Implementation of basic_get.

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

2 files changed, 36 insertions, 10 deletions

diff --git a/cpp/broker/inc/Channel.h b/cpp/broker/inc/Channel.h
index 4f4d8e2890..a5a54aea1f 100644
--- a/cpp/broker/inc/Channel.h
+++ b/cpp/broker/inc/Channel.h
@@ -60,12 +60,24 @@ namespace qpid {
                 Queue::shared_ptr queue;
                 string consumerTag;
                 u_int64_t deliveryTag;
-
-                AckRecord(Message::shared_ptr _msg, Queue::shared_ptr _queue, 
-                          string _consumerTag, u_int64_t _deliveryTag) : msg(_msg), 
-                                                                        queue(_queue), 
-                                                                        consumerTag(_consumerTag),
-                                                                        deliveryTag(_deliveryTag){}
+                bool pull;
+
+                AckRecord(Message::shared_ptr _msg, 
+                          Queue::shared_ptr _queue, 
+                          const string _consumerTag, 
+                          const u_int64_t _deliveryTag) : msg(_msg), 
+                                                          queue(_queue), 
+                                                          consumerTag(_consumerTag),
+                                                          deliveryTag(_deliveryTag),
+                                                          pull(false){}
+
+                AckRecord(Message::shared_ptr _msg, 
+                          Queue::shared_ptr _queue, 
+                          const u_int64_t _deliveryTag) : msg(_msg), 
+                                                          queue(_queue), 
+                                                          consumerTag(""),
+                                                          deliveryTag(_deliveryTag),
+                                                          pull(true){}
             };
 
             typedef std::vector<AckRecord>::iterator ack_iterator; 
@@ -89,12 +101,14 @@ namespace qpid {
                 void operator()(AckRecord& record) const;
             };
 
-            class AddSize{
+            class CalculatePrefetch{
                 u_int32_t size;
+                u_int16_t count;
             public:
-                AddSize();
+                CalculatePrefetch();
                 void operator()(AckRecord& record);
                 u_int32_t getSize();
+                u_int16_t getCount();
             };
 
             const int id;
@@ -106,6 +120,7 @@ namespace qpid {
             u_int32_t prefetchSize;    
             u_int16_t prefetchCount;    
             u_int32_t outstandingSize;    
+            u_int16_t outstandingCount;    
             u_int32_t framesize;
             Message::shared_ptr message;
             NameGenerator tagGenerator;
@@ -136,6 +151,7 @@ namespace qpid {
             bool exists(const string& consumerTag);
             void consume(string& tag, Queue::shared_ptr queue, bool acks, bool exclusive, ConnectionToken* const connection = 0);
             void cancel(const string& tag);
+            bool get(Queue::shared_ptr queue, bool ackExpected);
             void begin();
             void close();
             void commit();
diff --git a/cpp/broker/inc/Message.h b/cpp/broker/inc/Message.h
index 7b2c2bc848..94b9aa5bdd 100644
--- a/cpp/broker/inc/Message.h
+++ b/cpp/broker/inc/Message.h
@@ -49,6 +49,9 @@ namespace qpid {
             content_list content;
             u_int64_t size;
 
+            void sendContent(qpid::framing::OutputHandler* out, 
+                             int channel, u_int32_t framesize);
+
         public:
             typedef std::tr1::shared_ptr<Message> shared_ptr;
 
@@ -61,9 +64,16 @@ namespace qpid {
             bool isComplete();
             const ConnectionToken* const getPublisher();
 
-            void deliver(qpid::framing::OutputHandler* out, int channel, 
-                         string& consumerTag, u_int64_t deliveryTag, 
+            void deliver(qpid::framing::OutputHandler* out, 
+                         int channel, 
+                         const string& consumerTag, 
+                         u_int64_t deliveryTag, 
                          u_int32_t framesize);
+            void sendGetOk(qpid::framing::OutputHandler* out, 
+                           int channel, 
+                           u_int32_t messageCount,
+                           u_int64_t deliveryTag, 
+                           u_int32_t framesize);
             void redeliver();
 
             qpid::framing::BasicHeaderProperties* getHeaderProperties();