r1102@fuschia: andrew | 2007-02-09 00:52:04 +0000

 Got ack working for the non batched case
 Small tidy up in broker Channel


git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/qpid.0-9@505108 13f79535-47bb-0310-9956-ffa450edef68

4 files changed, 25 insertions, 23 deletions

diff --git a/cpp/lib/broker/BrokerAdapter.cpp b/cpp/lib/broker/BrokerAdapter.cpp
index 625dda1480..bdf41266ce 100644
--- a/cpp/lib/broker/BrokerAdapter.cpp
+++ b/cpp/lib/broker/BrokerAdapter.cpp
@@ -280,7 +280,7 @@ void BrokerAdapter::BrokerAdapter::BasicHandlerImpl::publish(
         BasicMessage* msg = new BasicMessage(
             &connection, exchangeName, routingKey, mandatory, immediate,
             context.methodBody);
-        channel.handlePublish(msg, exchange);
+        channel.handlePublish(msg);
     }else{
         throw ChannelException(
             404, "Exchange not found '" + exchangeName + "'");
diff --git a/cpp/lib/broker/BrokerChannel.cpp b/cpp/lib/broker/BrokerChannel.cpp
index 84ac747846..e0c5eebbec 100644
--- a/cpp/lib/broker/BrokerChannel.cpp
+++ b/cpp/lib/broker/BrokerChannel.cpp
@@ -140,7 +140,9 @@ void Channel::deliver(
 {
     Mutex::ScopedLock locker(deliveryLock);
 
-    u_int64_t deliveryTag = currentDeliveryTag++;
+	// Key the delivered messages to the id of the request in which they're sent 
+    u_int64_t deliveryTag = getNextSendRequestId();
+    
     if(ackExpected){
         unacked.push_back(DeliveryRecord(msg, queue, consumerTag, deliveryTag));
         outstanding.size += msg->contentSize();
@@ -188,24 +190,24 @@ void Channel::ConsumerImpl::requestDispatch(){
     if(blocked) queue->dispatch();
 }
 
-void Channel::handleInlineTransfer(
-    Message::shared_ptr msg, Exchange::shared_ptr& exch)
+void Channel::handleInlineTransfer(Message::shared_ptr msg)
 {
+    Exchange::shared_ptr exchange =
+        connection.broker.getExchanges().get(msg->getExchange());
     if(transactional){
         TxPublish* deliverable = new TxPublish(msg);
-        exch->route(*deliverable, msg->getRoutingKey(), &(msg->getApplicationHeaders()));
+        exchange->route(*deliverable, msg->getRoutingKey(), &(msg->getApplicationHeaders()));
         txBuffer.enlist(new DeletingTxOp(deliverable));
     }else{
         DeliverableMessage deliverable(msg);
-        exch->route(deliverable, msg->getRoutingKey(), &(msg->getApplicationHeaders()));
+        exchange->route(deliverable, msg->getRoutingKey(), &(msg->getApplicationHeaders()));
     }
 }
 
 // FIXME aconway 2007-02-05: Drop exchange member, calculate from
 // message in ::complete().
-void Channel::handlePublish(Message* _message, Exchange::shared_ptr _exchange){
+void Channel::handlePublish(Message* _message){
     Message::shared_ptr message(_message);
-    exchange = _exchange;
     messageBuilder.initialise(message);
 }
 
@@ -239,6 +241,11 @@ void Channel::complete(Message::shared_ptr msg) {
     }
 }
 
+// TODO astitcher 2007-02-08 This only deals correctly with non batched responses
+void Channel::ack(){
+	ack(getRequestInProgress(), false);
+}
+
 void Channel::ack(u_int64_t deliveryTag, bool multiple){
     if(transactional){
         accumulatedAck.update(deliveryTag, multiple);
diff --git a/cpp/lib/broker/BrokerChannel.h b/cpp/lib/broker/BrokerChannel.h
index 6e906e7615..52d8a0abeb 100644
--- a/cpp/lib/broker/BrokerChannel.h
+++ b/cpp/lib/broker/BrokerChannel.h
@@ -91,7 +91,6 @@ class Channel : public framing::ChannelAdapter,
     AccumulatedAck accumulatedAck;
     MessageStore* const store;
     MessageBuilder messageBuilder;//builder for in-progress message
-    Exchange::shared_ptr exchange;//exchange to which any in-progress message was published to
     bool opened;
 
     boost::scoped_ptr<BrokerAdapter> adapter;
@@ -131,15 +130,16 @@ class Channel : public framing::ChannelAdapter,
     void close();
     void commit();
     void rollback();
+    void ack();
     void ack(u_int64_t deliveryTag, bool multiple);
     void recover(bool requeue);
     void deliver(Message::shared_ptr& msg, const string& consumerTag, u_int64_t deliveryTag);            
-    void handlePublish(Message* msg, Exchange::shared_ptr exchange);
+    void handlePublish(Message* msg);
     void handleHeader(boost::shared_ptr<framing::AMQHeaderBody>);
     void handleContent(boost::shared_ptr<framing::AMQContentBody>);
     void handleHeartbeat(boost::shared_ptr<framing::AMQHeartbeatBody>);
     
-    void handleInlineTransfer(Message::shared_ptr msg, Exchange::shared_ptr& exchange);
+    void handleInlineTransfer(Message::shared_ptr msg);
     
     // For ChannelAdapter
     void handleMethodInContext(
diff --git a/cpp/lib/broker/MessageHandlerImpl.cpp b/cpp/lib/broker/MessageHandlerImpl.cpp
index 0d588aa351..7200027115 100644
--- a/cpp/lib/broker/MessageHandlerImpl.cpp
+++ b/cpp/lib/broker/MessageHandlerImpl.cpp
@@ -101,8 +101,9 @@ MessageHandlerImpl::consume(const MethodContext& context,
 void
 MessageHandlerImpl::empty( const MethodContext& )
 {
-    // FIXME astitcher 2007-01-11: 0-9 feature
-    THROW_QPID_ERROR(INTERNAL_ERROR, "Unimplemented ");
+    // Shouldn't ever receive this as it is a response to get
+    // TODO astitcher 2007-02-09 What is the correct exception to throw here?
+    THROW_QPID_ERROR(INTERNAL_ERROR, "Impossible");
 }
 
 void
@@ -112,12 +113,9 @@ MessageHandlerImpl::get( const MethodContext& context,
                          const string& /*destination*/,
                          bool noAck )
 {
-    //assert(0);                // FIXME astitcher 2007-01-11: 0-9 feature
-
     Queue::shared_ptr queue =
         connection.getQueue(queueName, context.channel->getId());
     
-    // FIXME: get is probably Basic specific
     if(channel.get(queue, !noAck))
         client.ok(context);
     else 
@@ -133,9 +131,9 @@ MessageHandlerImpl::offset(const MethodContext&,
 }
 
 void
-MessageHandlerImpl::ok( const MethodContext& )
+MessageHandlerImpl::ok(const MethodContext& /*context*/)
 {
-    // TODO: Need to ack the transfers acknowledged so far for flow control purp oses
+    channel.ack();
 }
 
 void
@@ -162,7 +160,6 @@ void
 MessageHandlerImpl::recover(const MethodContext& context,
                             bool requeue)
 {
-    THROW_QPID_ERROR(INTERNAL_ERROR, "Unimplemented");
     channel.recover(requeue);
     client.ok(context);
 }
@@ -188,7 +185,7 @@ MessageHandlerImpl::resume(const MethodContext&,
 void
 MessageHandlerImpl::transfer(const MethodContext& context,
                              u_int16_t /*ticket*/,
-                             const string& destination,
+                             const string& /* destination */,
                              bool /*redelivered*/,
                              bool /*immediate*/,
                              u_int64_t /*ttl*/,
@@ -211,8 +208,6 @@ MessageHandlerImpl::transfer(const MethodContext& context,
                              qpid::framing::Content body,
                              bool /*mandatory*/)
 {
-    Exchange::shared_ptr exchange(
-        broker.getExchanges().get(destination)); 
     MessageTransferBody::shared_ptr transfer(
         boost::shared_polymorphic_downcast<MessageTransferBody>(
             context.methodBody));
@@ -220,7 +215,7 @@ MessageHandlerImpl::transfer(const MethodContext& context,
         new MessageMessage(&connection, transfer));
     
     if (body.isInline()) 
-        channel.handleInlineTransfer(message, exchange);
+        channel.handleInlineTransfer(message);
     else 
         references.get(body.getValue()).addMessage(message);
     client.ok(context);