diff options
| author | Kim van der Riet <kpvdr@apache.org> | 2009-09-14 15:20:13 +0000 |
|---|---|---|
| committer | Kim van der Riet <kpvdr@apache.org> | 2009-09-14 15:20:13 +0000 |
| commit | 683a5677a37def4ed3c564fbebfe8951bcf86142 (patch) | |
| tree | 5ebe32fcb4a61f32eb12c64e05eceb402d8b5300 /cpp/src/qpid/broker/TopicExchange.cpp | |
| parent | d84a5a7e693922fbb18d7db020c28be8471f11e3 (diff) | |
| download | qpid-python-683a5677a37def4ed3c564fbebfe8951bcf86142.tar.gz | |
Reversed checkin of r.813825 until its problems can be resolved
git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@814692 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'cpp/src/qpid/broker/TopicExchange.cpp')
| -rw-r--r-- | cpp/src/qpid/broker/TopicExchange.cpp | 62 |
1 files changed, 41 insertions, 21 deletions
diff --git a/cpp/src/qpid/broker/TopicExchange.cpp b/cpp/src/qpid/broker/TopicExchange.cpp index 99d6c1cb8d..6bf0b104ea 100644 --- a/cpp/src/qpid/broker/TopicExchange.cpp +++ b/cpp/src/qpid/broker/TopicExchange.cpp @@ -7,9 +7,9 @@ * to you 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 @@ -36,7 +36,7 @@ namespace _qmf = qmf::org::apache::qpid::broker; // - excessive string copying: should be 0 copy, match from original buffer. // - match/lookup: use descision tree or other more efficient structure. -namespace +namespace { const std::string qpidFedOp("qpid.fed.op"); const std::string qpidFedTags("qpid.fed.tags"); @@ -53,7 +53,7 @@ namespace { // Iterate over a string of '.'-separated tokens. struct TokenIterator { typedef pair<const char*,const char*> Token; - + TokenIterator(const char* b, const char* e) : token(make_pair(b, find(b,e,'.'))), end(e) {} bool finished() const { return !token.first; } @@ -122,7 +122,7 @@ class Matcher { Matcher(const string& p, const string& k) : matched(), pattern(&p[0], &p[0]+p.size()), key(&k[0], &k[0]+k.size()) { matched = match(); } - + operator bool() const { return matched; } private: @@ -158,7 +158,7 @@ class Matcher { } if (!pattern.finished() && pattern.match1('#')) pattern.next(); // Trailing # matches empty. - return pattern.finished() && key.finished(); + return pattern.finished() && key.finished(); } bool matched; @@ -173,7 +173,7 @@ string TopicExchange::normalize(const string& pattern) { return normal; } -bool TopicExchange::match(const string& pattern, const string& key) +bool TopicExchange::match(const string& pattern, const string& key) { return Matcher(pattern, key); } @@ -231,11 +231,11 @@ bool TopicExchange::bind(Queue::shared_ptr queue, const string& routingKey, cons */ std::vector<std::string> keys2prop; { - RWlock::ScopedRlock l(lock); + RWlock::ScopedRlock l(lock); for (BindingMap::iterator iter = bindings.begin(); iter != bindings.end(); iter++) { const BoundKey& bk = iter->second; - + if (bk.fedBinding.hasLocal()) { keys2prop.push_back(iter->first); } @@ -293,24 +293,44 @@ bool TopicExchange::isBound(Queue::shared_ptr queue, const string& pattern) return q != qv.end(); } -void TopicExchange::route(Deliverable& msg, const string& routingKey, const FieldTable* /*args*/) -{ - qpid::sys::CopyOnWriteArray<Binding::shared_ptr>::Ptr b(new std::vector<boost::shared_ptr<qpid::broker::Exchange::Binding> >); +void TopicExchange::route(Deliverable& msg, const string& routingKey, const FieldTable* /*args*/){ + Binding::vector mb; PreRoute pr(msg, this); + uint32_t count(0); { - RWlock::ScopedRlock l(lock); - for (BindingMap::iterator i = bindings.begin(); i != bindings.end(); ++i) { - if (match(i->first, routingKey)) { - Binding::vector& qv(i->second.bindingVector); - for(Binding::vector::iterator j = qv.begin(); j != qv.end(); j++){ - b->push_back(*j); - } + RWlock::ScopedRlock l(lock); + for (BindingMap::iterator i = bindings.begin(); i != bindings.end(); ++i) { + if (match(i->first, routingKey)) { + Binding::vector& qv(i->second.bindingVector); + for(Binding::vector::iterator j = qv.begin(); j != qv.end(); j++, count++){ + mb.push_back(*j); } } } + } + + for (Binding::vector::iterator j = mb.begin(); j != mb.end(); ++j) { + msg.deliverTo((*j)->queue); + if ((*j)->mgmtBinding != 0) + (*j)->mgmtBinding->inc_msgMatched (); + } - doRoute(msg, b); + if (mgmtExchange != 0) + { + mgmtExchange->inc_msgReceives (); + mgmtExchange->inc_byteReceives (msg.contentSize ()); + if (count == 0) + { + mgmtExchange->inc_msgDrops (); + mgmtExchange->inc_byteDrops (msg.contentSize ()); + } + else + { + mgmtExchange->inc_msgRoutes (count); + mgmtExchange->inc_byteRoutes (count * msg.contentSize ()); + } + } } bool TopicExchange::isBound(Queue::shared_ptr queue, const string* const routingKey, const FieldTable* const) @@ -323,7 +343,7 @@ bool TopicExchange::isBound(Queue::shared_ptr queue, const string* const routing return bindings.size() > 0; } else if (routingKey) { for (BindingMap::iterator i = bindings.begin(); i != bindings.end(); ++i) { - if (match(i->first, *routingKey)) + if (match(i->first, *routingKey)) return true; } } else { |
