Decoupled routing from the channel and message classes.

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

4 files changed, 42 insertions, 40 deletions

diff --git a/cpp/broker/src/Channel.cpp b/cpp/broker/src/Channel.cpp
index 4fb6a52b99..ae99f4e7fa 100644
--- a/cpp/broker/src/Channel.cpp
+++ b/cpp/broker/src/Channel.cpp
@@ -126,38 +126,17 @@ void Channel::ConsumerImpl::cancel(){
     if(queue) queue->cancel(this);
 }
 
-void Channel::handlePublish(Message* msg){
-    if(message.get()){
-        THROW_QPID_ERROR(PROTOCOL_ERROR + 504, "Invalid message sequence: got publish before previous content was completed.");
-    }
-    message = Message::shared_ptr(msg);
-}
-
-void Channel::handleHeader(AMQHeaderBody::shared_ptr header, ExchangeRegistry* exchanges){
+void Channel::checkMessage(const std::string& text){
     if(!message.get()){
-        THROW_QPID_ERROR(PROTOCOL_ERROR + 504, "Invalid message sequence: got header before publish.");
-    }
-    message->setHeader(header);
-    if(message->isComplete()){
-        publish(exchanges);
+        THROW_QPID_ERROR(PROTOCOL_ERROR + 504, text);
     }
 }
 
-void Channel::handleContent(AMQContentBody::shared_ptr content, ExchangeRegistry* exchanges){
-    if(!message.get()){
-        THROW_QPID_ERROR(PROTOCOL_ERROR + 504, "Invalid message sequence: got content before publish.");
-    }
-    message->addContent(content);
-    if(message->isComplete()){
-        publish(exchanges);
-    }
-}
-
-void Channel::publish(ExchangeRegistry* exchanges){
-    if(!route(message, exchanges)){
-        std::cout << "WARNING: Could not route message." << std::endl;
+void Channel::handlePublish(Message* msg){
+    if(message.get()){
+        THROW_QPID_ERROR(PROTOCOL_ERROR + 504, "Invalid message sequence: got publish before previous content was completed.");
     }
-    message.reset();
+    message = Message::shared_ptr(msg);
 }
 
 void Channel::ack(u_int64_t deliveryTag, bool multiple){
diff --git a/cpp/broker/src/Message.cpp b/cpp/broker/src/Message.cpp
index 8ebe40410a..a4ae85e904 100644
--- a/cpp/broker/src/Message.cpp
+++ b/cpp/broker/src/Message.cpp
@@ -90,13 +90,3 @@ const ConnectionToken* const Message::getPublisher(){
     return publisher;
 }
 
-bool qpid::broker::route(Message::shared_ptr& msg, ExchangeRegistry* registry){
-    Exchange* exchange = registry->get(msg->exchange);
-    if(exchange){
-        exchange->route(msg, msg->routingKey, &(msg->getHeaderProperties()->getHeaders()));
-        return true;
-    }else{
-        return false;
-    }
-}
-
diff --git a/cpp/broker/src/Router.cpp b/cpp/broker/src/Router.cpp
new file mode 100644
index 0000000000..c2dd74bf7d
--- /dev/null
+++ b/cpp/broker/src/Router.cpp
@@ -0,0 +1,32 @@
+/*
+ *
+ * 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 "Router.h"
+
+using namespace qpid::broker;
+
+Router::Router(ExchangeRegistry& _registry) : registry(_registry){}
+
+void Router::operator()(Message::shared_ptr& msg){
+    Exchange* exchange = registry.get(msg->getExchange());
+    if(exchange){
+        exchange->route(msg, msg->getRoutingKey(), &(msg->getHeaderProperties()->getHeaders()));
+    }else{
+        std::cout << "WARNING: Could not route message, unknown exchange: " << msg->getExchange() << std::endl;
+    }
+
+}
diff --git a/cpp/broker/src/SessionHandlerImpl.cpp b/cpp/broker/src/SessionHandlerImpl.cpp
index eb8f37030c..63a42a7fd6 100644
--- a/cpp/broker/src/SessionHandlerImpl.cpp
+++ b/cpp/broker/src/SessionHandlerImpl.cpp
@@ -18,8 +18,9 @@
 #include <iostream>
 #include "SessionHandlerImpl.h"
 #include "FanOutExchange.h"
-#include "TopicExchange.h"
 #include "HeadersExchange.h"
+#include "Router.h"
+#include "TopicExchange.h"
 #include "assert.h"
 
 using namespace std::tr1;
@@ -153,11 +154,11 @@ void SessionHandlerImpl::closed(){
 }
 
 void SessionHandlerImpl::handleHeader(u_int16_t channel, AMQHeaderBody::shared_ptr body){
-    getChannel(channel)->handleHeader(body, exchanges);
+    getChannel(channel)->handleHeader(body, Router(*exchanges));
 }
 
 void SessionHandlerImpl::handleContent(u_int16_t channel, AMQContentBody::shared_ptr body){
-    getChannel(channel)->handleContent(body, exchanges);
+    getChannel(channel)->handleContent(body, Router(*exchanges));
 }
 
 void SessionHandlerImpl::handleHeartbeat(AMQHeartbeatBody::shared_ptr body){