diff options
| -rw-r--r-- | cpp/src/qpid/broker/BrokerChannel.cpp | 20 | ||||
| -rw-r--r-- | cpp/src/qpid/broker/BrokerChannel.h | 5 |
2 files changed, 20 insertions, 5 deletions
diff --git a/cpp/src/qpid/broker/BrokerChannel.cpp b/cpp/src/qpid/broker/BrokerChannel.cpp index 0dc4bed661..11d8db73f8 100644 --- a/cpp/src/qpid/broker/BrokerChannel.cpp +++ b/cpp/src/qpid/broker/BrokerChannel.cpp @@ -335,15 +335,25 @@ void Channel::complete(Message::shared_ptr msg) { } } + + void Channel::route(Message::shared_ptr msg, Deliverable& strategy) { - Exchange::shared_ptr exchange = connection.broker.getExchanges().get(msg->getExchange()); - assert(exchange.get()); - exchange->route(strategy, msg->getRoutingKey(), &(msg->getApplicationHeaders())); + + std::string routeToExchangeName = msg->getExchange(); + // cache the exchange lookup + if (!cacheExchange.get() || cacheExchangeName != routeToExchangeName){ + cacheExchangeName = routeToExchangeName; + cacheExchange = connection.broker.getExchanges().get(routeToExchangeName); + } + + assert(cacheExchange.get()); + cacheExchange->route(strategy, msg->getRoutingKey(), &(msg->getApplicationHeaders())); + if (!strategy.delivered) { //TODO:if reject-unroutable, then reject //else route to alternate exchange - if (exchange->getAlternate()) { - exchange->getAlternate()->route(strategy, msg->getRoutingKey(), &(msg->getApplicationHeaders())); + if (cacheExchange->getAlternate()) { + cacheExchange->getAlternate()->route(strategy, msg->getRoutingKey(), &(msg->getApplicationHeaders())); } } diff --git a/cpp/src/qpid/broker/BrokerChannel.h b/cpp/src/qpid/broker/BrokerChannel.h index 021110cf8c..fcfcd73679 100644 --- a/cpp/src/qpid/broker/BrokerChannel.h +++ b/cpp/src/qpid/broker/BrokerChannel.h @@ -74,6 +74,7 @@ class Channel : public CompletionHandler bool blocked; bool windowing; uint32_t msgCredit; + uint32_t byteCredit; public: @@ -116,6 +117,10 @@ class Channel : public CompletionHandler MessageBuilder messageBuilder;//builder for in-progress message bool opened; bool flowActive; + + std::string cacheExchangeName; // pair holds last exchange used for routing + Exchange::shared_ptr cacheExchange; + void route(Message::shared_ptr msg, Deliverable& strategy); void complete(Message::shared_ptr msg);// completion handler for MessageBuilder void record(const DeliveryRecord& delivery); |