* Made client::Channel bi-modal: 0-8 or 0-9 modes.

* Added dummy impl of client::MessageMessageChannel.
* Generalised ClientChannelTest to be able to test both modes.


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

4 files changed, 425 insertions, 11 deletions

diff --git a/cpp/src/client/ClientChannel.cpp b/cpp/src/client/ClientChannel.cpp
index eda872fc30..99eece46bc 100644
--- a/cpp/src/client/ClientChannel.cpp
+++ b/cpp/src/client/ClientChannel.cpp
@@ -26,8 +26,7 @@
 #include "MethodBodyInstances.h"
 #include "Connection.h"
 #include "BasicMessageChannel.h"
-// FIXME aconway 2007-03-21: 
-//#include "MessageMessageChannel.h"
+#include "MessageMessageChannel.h"
 
 // FIXME aconway 2007-01-26: Evaluate all throws, ensure consistent
 // handling of errors that should close the connection or the channel.
@@ -39,14 +38,15 @@ using namespace qpid::client;
 using namespace qpid::framing;
 using namespace qpid::sys;
 
-Channel::Channel(bool _transactional, u_int16_t _prefetch,
-                  MessageChannel* impl) :
-    // FIXME aconway 2007-03-21: MessageMessageChannel
-    messaging(impl ? impl : new BasicMessageChannel(*this)),
-    connection(0), 
-    prefetch(_prefetch), 
-    transactional(_transactional)
-{ }
+Channel::Channel(bool _transactional, u_int16_t _prefetch, InteropMode mode) :
+    connection(0), prefetch(_prefetch), transactional(_transactional)
+{
+    switch (mode) {
+      case AMQP_08: messaging.reset(new BasicMessageChannel(*this)); break;
+      case AMQP_09: messaging.reset(new MessageMessageChannel(*this)); break;
+      default: assert(0); QPID_ERROR(INTERNAL_ERROR, "Invalid interop-mode.");
+    }
+}
 
 Channel::~Channel(){
     close();
diff --git a/cpp/src/client/ClientChannel.h b/cpp/src/client/ClientChannel.h
index a7e0d2ec31..cf2ea1dbe5 100644
--- a/cpp/src/client/ClientChannel.h
+++ b/cpp/src/client/ClientChannel.h
@@ -112,6 +112,7 @@ class Channel : public framing::ChannelAdapter
   friend class MessageMessageChannel; // for sendAndReceive.
         
   public:
+    enum InteropMode { AMQP_08, AMQP_09 };
 
     /**
      * Creates a channel object.
@@ -130,7 +131,7 @@ class Channel : public framing::ChannelAdapter
      */
     Channel(
         bool transactional = false, u_int16_t prefetch = 500,
-        MessageChannel* messageImpl = 0);
+        InteropMode=AMQP_08);
      
     ~Channel();
 
diff --git a/cpp/src/client/MessageMessageChannel.cpp b/cpp/src/client/MessageMessageChannel.cpp
new file mode 100644
index 0000000000..25fbb95413
--- /dev/null
+++ b/cpp/src/client/MessageMessageChannel.cpp
@@ -0,0 +1,331 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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 <iostream>
+#include <boost/format.hpp>
+#include "MessageMessageChannel.h"
+#include "../framing/AMQMethodBody.h"
+#include "ClientChannel.h"
+#include "ReturnedMessageHandler.h"
+#include "MessageListener.h"
+#include "../framing/FieldTable.h"
+#include "Connection.h"
+#include "../shared_ptr.h"
+
+namespace qpid {
+namespace client {
+
+using namespace std;
+using namespace sys;
+using namespace framing;
+
+MessageMessageChannel::MessageMessageChannel(Channel& ch)
+    : channel(ch), tagCount(0) {}
+
+string MessageMessageChannel::newTag() {
+    Mutex::ScopedLock l(lock);
+    return (boost::format("__tag%d")%++tagCount).str();
+}
+
+void MessageMessageChannel::consume(
+    Queue& queue, std::string& tag, MessageListener* /*listener*/, 
+    AckMode ackMode, bool noLocal, bool /*synch*/, const FieldTable* fields)
+{
+    if (tag.empty())
+        tag = newTag();
+    channel.sendAndReceive<MessageOkBody>(
+        new MessageConsumeBody(
+            channel.getVersion(), 0, queue.getName(), tag, noLocal,
+            ackMode == NO_ACK, false, fields ? *fields : FieldTable()));
+    
+//     // FIXME aconway 2007-02-20: Race condition!
+//     // We could receive the first message for the consumer
+//     // before we create the consumer below.
+//     // Move consumer creation to handler for MessageConsumeOkBody
+//     {
+//         Mutex::ScopedLock l(lock);
+//         ConsumerMap::iterator i = consumers.find(tag);
+//         if (i != consumers.end())
+//             THROW_QPID_ERROR(CLIENT_ERROR,
+//                              "Consumer already exists with tag="+tag);
+//         Consumer& c = consumers[tag];
+//         c.listener = listener;
+//         c.ackMode = ackMode;
+//         c.lastDeliveryTag = 0;
+//     }
+}
+
+
+void MessageMessageChannel::cancel(const std::string& /*tag*/, bool /*synch*/) {
+    // FIXME aconway 2007-02-23: 
+//     Consumer c;
+//     {
+//         Mutex::ScopedLock l(lock);
+//         ConsumerMap::iterator i = consumers.find(tag);
+//         if (i == consumers.end())
+//             return;
+//         c = i->second;
+//         consumers.erase(i);
+//     }
+//     if(c.ackMode == LAZY_ACK && c.lastDeliveryTag > 0) 
+//         channel.send(new MessageAckBody(channel.version, c.lastDeliveryTag, true));
+//     channel.sendAndReceiveSync<MessageCancelOkBody>(
+//         synch, new MessageCancelBody(channel.version, tag, !synch));
+}
+
+void MessageMessageChannel::close(){
+    // FIXME aconway 2007-02-23: 
+//     ConsumerMap consumersCopy;
+//     {
+//         Mutex::ScopedLock l(lock);
+//         consumersCopy = consumers;
+//         consumers.clear();
+//     }
+//     for (ConsumerMap::iterator i=consumersCopy.begin();
+//          i  != consumersCopy.end(); ++i)
+//     {
+//         Consumer& c = i->second;
+//         if ((c.ackMode == LAZY_ACK || c.ackMode == AUTO_ACK)
+//             && c.lastDeliveryTag > 0)
+//         {
+//             channel.send(new MessageAckBody(channel.version, c.lastDeliveryTag, true));
+//         }
+//     }
+//     incoming.shutdown();
+}
+
+
+/** Destination ID for the current get.
+ * Must not clash with a generated consumer ID.
+ * TODO aconway 2007-03-06: support multiple outstanding gets?
+ */
+const string getDestinationId("__get__");
+
+bool MessageMessageChannel::get(
+    Message& , const Queue& , AckMode )
+{
+    Mutex::ScopedLock l(lock);
+//     incoming.addDestination(getDestinationId, getDest);
+//     channel.send(
+//         new MessageGetBody(
+//             channel.version, 0, queue.getName(), getDestinationId, ackMode));
+//    return getDest.wait(msg);
+    return false;
+}
+
+
+/** Convert a message to a transfer command. */
+MessageTransferBody::shared_ptr makeTransfer(
+    ProtocolVersion version,
+    const Message& msg, const string& destination,
+    const std::string& routingKey, bool mandatory, bool immediate)
+{
+    return MessageTransferBody::shared_ptr(
+        new MessageTransferBody(
+            version,
+            0,                  // FIXME aconway 2007-04-03: ticket.
+            destination,
+            msg.isRedelivered(),
+            immediate,
+            0,                  // FIXME aconway 2007-02-23: ttl
+            msg.getPriority(),
+            msg.getTimestamp(),
+            static_cast<uint8_t>(msg.getDeliveryMode()),
+            0,                  // FIXME aconway 2007-04-03: Expiration
+            string(),           // Exchange: for broker use only.
+            routingKey,
+            msg.getMessageId(),
+            msg.getCorrelationId(),
+            msg.getReplyTo(),
+            msg.getContentType(),
+            msg.getContentEncoding(),
+            msg.getUserId(),
+            msg.getAppId(),
+            string(),       // FIXME aconway 2007-04-03: TransactionId
+            string(),        //FIXME aconway 2007-04-03: SecurityToken
+            msg.getHeaders(),
+            Content(INLINE, msg.getData()),
+            mandatory
+        ));
+}
+
+void MessageMessageChannel::publish(
+    const Message& msg, const Exchange& exchange,
+    const std::string& routingKey, bool mandatory, bool immediate)
+{
+    MessageTransferBody::shared_ptr transfer = makeTransfer(
+        channel.getVersion(),
+        msg, exchange.getName(), routingKey, mandatory, immediate);
+    // Frame itself uses 8 bytes.
+    u_int32_t frameMax = channel.connection->getMaxFrameSize() - 8;
+    if (transfer->size()  > frameMax) {
+        // FIXME aconway 2007-02-23: 
+        throw QPID_ERROR(INTERNAL_ERROR, "References not yet implemented");
+    }
+    channel.sendAndReceive<MessageOkBody>(transfer.get());
+}
+        
+
+void MessageMessageChannel::handle(boost::shared_ptr<AMQMethodBody> method) {
+    assert(method->amqpClassId() ==MessageTransferBody::CLASS_ID);
+    switch(method->amqpMethodId()) {
+      case MessageAppendBody::METHOD_ID: {
+          MessageAppendBody::shared_ptr append =
+              shared_polymorphic_downcast<MessageAppendBody>(method);
+          incoming.appendReference(append->getReference(), append->getBytes());
+          break;
+      }
+      case MessageOpenBody::METHOD_ID: {
+          MessageOpenBody::shared_ptr open =
+              shared_polymorphic_downcast<MessageOpenBody>(method);
+          incoming.openReference(open->getReference());
+          break;
+      }
+
+      case MessageCloseBody::METHOD_ID: {
+          MessageCloseBody::shared_ptr close =
+              shared_polymorphic_downcast<MessageCloseBody>(method);
+          incoming.closeReference(close->getReference());
+          break;
+      }
+
+      case MessageEmptyBody::METHOD_ID: {
+          // FIXME aconway 2007-04-04: 
+          // getDest.empty();
+          break;
+      }
+
+      case MessageCancelBody::METHOD_ID:
+      case MessageCheckpointBody::METHOD_ID:
+
+        // FIXME aconway 2007-04-03:  TODO
+      case MessageOkBody::METHOD_ID:
+      case MessageOffsetBody::METHOD_ID:
+      case MessageQosBody::METHOD_ID:
+      case MessageRecoverBody::METHOD_ID:
+      case MessageRejectBody::METHOD_ID:
+      case MessageResumeBody::METHOD_ID:
+      case MessageTransferBody::METHOD_ID:
+      default:
+        throw Channel::UnknownMethod();
+    }
+}
+
+void MessageMessageChannel::handle(AMQHeaderBody::shared_ptr ){
+    throw QPID_ERROR(INTERNAL_ERROR, "Basic protocol not supported");
+}
+    
+void MessageMessageChannel::handle(AMQContentBody::shared_ptr ){
+    throw QPID_ERROR(INTERNAL_ERROR, "Basic protocol not supported");
+}
+
+// FIXME aconway 2007-02-23: 
+// void MessageMessageChannel::deliver(IncomingMessage::Destination& consumer, Message& msg){
+//     //record delivery tag:
+//     consumer.lastDeliveryTag = msg.getDeliveryTag();
+
+//     //allow registered listener to handle the message
+//     consumer.listener->received(msg);
+
+//     if(channel.isOpen()){
+//         bool multiple(false);
+//         switch(consumer.ackMode){
+//           case LAZY_ACK: 
+//             multiple = true;
+//             if(++(consumer.count) < channel.getPrefetch())
+//                 break;
+//             //else drop-through
+//           case AUTO_ACK:
+//             consumer.lastDeliveryTag = 0;
+//             channel.send(
+//                 new MessageAckBody(
+//                     channel.version, msg.getDeliveryTag(), multiple));
+//           case NO_ACK:          // Nothing to do
+//           case CLIENT_ACK:      // User code must ack.
+//             break;
+//             // TODO aconway 2007-02-22: Provide a way for user
+//             // to ack!
+//         }
+//     }
+
+//     //as it stands, transactionality is entirely orthogonal to ack
+//     //mode, though the acks will not be processed by the broker under
+//     //a transaction until it commits.
+// }
+
+
+void MessageMessageChannel::run() {
+    // FIXME aconway 2007-02-23: 
+//     while(channel.isOpen()) {
+//         try {
+//             Message msg = incoming.waitDispatch();
+//             if(msg.getMethod()->isA<MessageReturnBody>()) {
+//                 ReturnedMessageHandler* handler=0;
+//                 {
+//                     Mutex::ScopedLock l(lock);
+//                     handler=returnsHandler;
+//                 }
+//                 if(handler == 0) {
+//                     // TODO aconway 2007-02-20: proper logging.
+//                     cout << "Message returned: " << msg.getData() << endl;
+//                 }
+//                 else 
+//                     handler->returned(msg);
+//             }
+//             else {
+//                 MessageDeliverBody::shared_ptr deliverBody =
+//                     boost::shared_polymorphic_downcast<MessageDeliverBody>(
+//                         msg.getMethod());
+//                 std::string tag = deliverBody->getConsumerTag();
+//                 Consumer consumer;
+//                 {
+//                     Mutex::ScopedLock l(lock);
+//                     ConsumerMap::iterator i = consumers.find(tag);
+//                     if(i == consumers.end()) 
+//                         THROW_QPID_ERROR(PROTOCOL_ERROR+504,
+//                                          "Unknown consumer tag=" + tag);
+//                     consumer = i->second;
+//                 }
+//                 deliver(consumer, msg);
+//             }
+//         }
+//         catch (const ShutdownException&) {
+//             /* Orderly shutdown */
+//         }
+//         catch (const Exception& e) {
+//             // FIXME aconway 2007-02-20: Report exception to user.
+//             cout << "client::Message::run() terminated by: " << e.toString()
+//                  << "(" << typeid(e).name() << ")" << endl;
+//         }
+//     }
+}
+
+void MessageMessageChannel::setReturnedMessageHandler(
+    ReturnedMessageHandler* )
+{
+    throw QPID_ERROR(INTERNAL_ERROR, "Message class does not support returns");
+}
+
+void MessageMessageChannel::setQos(){
+    channel.sendAndReceive<MessageOkBody>(
+        new MessageQosBody(channel.version, 0, channel.getPrefetch(), false));
+    if(channel.isTransactional())
+        channel.sendAndReceive<TxSelectOkBody>(
+            new TxSelectBody(channel.version));
+}
+
+}} // namespace qpid::client
diff --git a/cpp/src/client/MessageMessageChannel.h b/cpp/src/client/MessageMessageChannel.h
new file mode 100644
index 0000000000..4c4721ce90
--- /dev/null
+++ b/cpp/src/client/MessageMessageChannel.h
@@ -0,0 +1,82 @@
+#ifndef _client_MessageMessageChannel_h
+#define _client_MessageMessageChannel_h
+
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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 "MessageChannel.h"
+#include "IncomingMessage.h"
+#include "../sys/Monitor.h"
+#include <boost/ptr_container/ptr_map.hpp>
+
+namespace qpid {
+namespace client {
+/**
+ * Messaging implementation using AMQP 0-9 MessageMessageChannel class
+ * to send and receiving messages.
+ */
+class MessageMessageChannel : public MessageChannel
+{
+  public:
+    MessageMessageChannel(Channel& parent);
+    
+    void consume(
+        Queue& queue, std::string& tag, MessageListener* listener, 
+        AckMode ackMode = NO_ACK, bool noLocal = false, bool synch = true,
+        const framing::FieldTable* fields = 0);
+    
+    void cancel(const std::string& tag, bool synch = true);
+
+    bool get(Message& msg, const Queue& queue, AckMode ackMode = NO_ACK);
+
+    void publish(const Message& msg, const Exchange& exchange,
+                 const std::string& routingKey, 
+                 bool mandatory = false, bool immediate = false);
+
+    void setReturnedMessageHandler(ReturnedMessageHandler* handler);
+
+    void run();
+
+    void handle(boost::shared_ptr<framing::AMQMethodBody>);
+
+    void handle(shared_ptr<framing::AMQHeaderBody>);
+
+    void handle(shared_ptr<framing::AMQContentBody>);
+    
+    void setQos();
+    
+    void close();
+
+  private:
+    typedef boost::ptr_map<std::string, IncomingMessage::WaitableDestination>
+    Destinations;
+
+    std::string newTag();
+
+    sys::Mutex lock;
+    Channel& channel;
+    IncomingMessage incoming;
+    long tagCount;
+};
+
+}} // namespace qpid::client
+
+
+
+#endif  /*!_client_MessageMessageChannel_h*/
+