You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kp...@apache.org on 2012/08/03 14:14:12 UTC
svn commit: r1368910 [6/27] - in /qpid/branches/asyncstore: ./ bin/
cpp/bindings/qmf/ruby/ cpp/bindings/qmf2/ruby/ cpp/bindings/qpid/python/
cpp/bindings/qpid/ruby/ cpp/bindings/qpid/ruby/features/
cpp/bindings/qpid/ruby/features/step_definitions/ cpp/...
Modified: qpid/branches/asyncstore/cpp/src/qpid/broker/TopicExchange.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/broker/TopicExchange.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/broker/TopicExchange.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/broker/TopicExchange.cpp Fri Aug 3 12:13:32 2012
@@ -32,20 +32,8 @@ using namespace qpid::sys;
using namespace std;
namespace _qmf = qmf::org::apache::qpid::broker;
-
-// TODO aconway 2006-09-20: More efficient matching algorithm.
-// Areas for improvement:
-// - excessive string copying: should be 0 copy, match from original buffer.
-// - match/lookup: use descision tree or other more efficient structure.
-
-namespace
-{
-const std::string STAR("*");
-const std::string HASH("#");
-}
-
// iterator for federation ReOrigin bind operation
-class TopicExchange::ReOriginIter : public TopicExchange::BindingNode::TreeIterator {
+class TopicExchange::ReOriginIter : public BindingNode::TreeIterator {
public:
ReOriginIter() {};
~ReOriginIter() {};
@@ -61,7 +49,7 @@ public:
// match iterator used by route(): builds BindingList of all unique queues
// that match the routing key.
-class TopicExchange::BindingsFinderIter : public TopicExchange::BindingNode::TreeIterator {
+class TopicExchange::BindingsFinderIter : public BindingNode::TreeIterator {
public:
BindingsFinderIter(BindingList &bl) : b(bl) {};
~BindingsFinderIter() {};
@@ -85,7 +73,7 @@ public:
// Iterator to visit all bindings until a given queue is found
-class TopicExchange::QueueFinderIter : public TopicExchange::BindingNode::TreeIterator {
+class TopicExchange::QueueFinderIter : public BindingNode::TreeIterator {
public:
QueueFinderIter(Queue::shared_ptr queue) : queue(queue), found(false) {};
~QueueFinderIter() {};
@@ -107,58 +95,7 @@ public:
};
-// Iterate over a string of '.'-separated tokens.
-struct TopicExchange::TokenIterator {
- typedef pair<const char*,const char*> Token;
-
- TokenIterator(const char* b, const char* e) : end(e), token(make_pair(b, find(b,e,'.'))) {}
-
- TokenIterator(const string& key) : end(&key[0]+key.size()), token(make_pair(&key[0], find(&key[0],end,'.'))) {}
-
- bool finished() const { return !token.first; }
-
- void next() {
- if (token.second == end)
- token.first = token.second = 0;
- else {
- token.first=token.second+1;
- token.second=(find(token.first, end, '.'));
- }
- }
-
- void pop(string &top) {
- ptrdiff_t l = len();
- if (l) {
- top.assign(token.first, l);
- } else top.clear();
- next();
- }
-
- bool match1(char c) const {
- return token.second==token.first+1 && *token.first == c;
- }
-
- bool match(const Token& token2) const {
- ptrdiff_t l=len();
- return l == token2.second-token2.first &&
- strncmp(token.first, token2.first, l) == 0;
- }
-
- bool match(const string& str) const {
- ptrdiff_t l=len();
- return l == ptrdiff_t(str.size()) &&
- str.compare(0, l, token.first, l) == 0;
- }
-
- ptrdiff_t len() const { return token.second - token.first; }
-
-
- const char* end;
- Token token;
-};
-
-
-class TopicExchange::Normalizer : public TopicExchange::TokenIterator {
+class TopicExchange::Normalizer : public TokenIterator {
public:
Normalizer(string& p)
: TokenIterator(&p[0], &p[0]+p.size()), pattern(p)
@@ -230,7 +167,7 @@ bool TopicExchange::bind(Queue::shared_p
if (args == 0 || fedOp.empty() || fedOp == fedOpBind) {
RWlock::ScopedWlock l(lock);
- BindingKey *bk = bindingTree.addBindingKey(routingPattern);
+ BindingKey *bk = bindingTree.add(routingPattern);
if (bk) {
Binding::vector& qv(bk->bindingVector);
Binding::vector::iterator q;
@@ -324,7 +261,7 @@ bool TopicExchange::deleteBinding(Queue:
nBindings--;
if(qv.empty()) {
- bindingTree.removeBindingKey(routingKey);
+ bindingTree.remove(routingKey);
}
if (mgmtExchange != 0) {
mgmtExchange->dec_bindingCount();
@@ -340,7 +277,7 @@ bool TopicExchange::deleteBinding(Queue:
TopicExchange::BindingKey *TopicExchange::getQueueBinding(Queue::shared_ptr queue, const string& pattern)
{
// Note well: lock held by caller....
- BindingKey *bk = bindingTree.getBindingKey(pattern); // Exact match against binding pattern
+ BindingKey *bk = bindingTree.get(pattern); // Exact match against binding pattern
if (!bk) return 0;
Binding::vector& qv(bk->bindingVector);
Binding::vector::iterator q;
@@ -385,7 +322,7 @@ bool TopicExchange::isBound(Queue::share
} else if (!routingKey && !queue) {
return nBindings > 0;
} else if (routingKey) {
- if (bindingTree.getBindingKey(*routingKey)) {
+ if (bindingTree.get(*routingKey)) {
return true;
}
} else {
@@ -400,294 +337,4 @@ TopicExchange::~TopicExchange() {}
const std::string TopicExchange::typeName("topic");
-//
-// class BindingNode
-//
-
-TopicExchange::BindingNode::~BindingNode()
-{
- childTokens.clear();
-}
-
-
-// Add a binding pattern to the tree. Return a pointer to the binding key
-// of the node that matches the binding pattern.
-TopicExchange::BindingKey*
-TopicExchange::BindingNode::addBindingKey(const std::string& normalizedRoute)
-{
- TokenIterator bKey(normalizedRoute);
- return addBindingKey(bKey, normalizedRoute);
-}
-
-
-// Return a pointer to the binding key of the leaf node that matches the binding pattern.
-TopicExchange::BindingKey*
-TopicExchange::BindingNode::getBindingKey(const std::string& normalizedRoute)
-{
- TokenIterator bKey(normalizedRoute);
- return getBindingKey(bKey);
-}
-
-
-// Delete the binding associated with the given route.
-void TopicExchange::BindingNode::removeBindingKey(const std::string& normalizedRoute)
-{
- TokenIterator bKey2(normalizedRoute);
- removeBindingKey(bKey2, normalizedRoute);
-}
-
-// visit each node in the tree. Note: all nodes are visited,
-// even non-leaf nodes (i.e. nodes without any bindings)
-bool TopicExchange::BindingNode::iterateAll(TopicExchange::BindingNode::TreeIterator& iter)
-{
- if (!iter.visit(*this)) return false;
-
- if (starChild && !starChild->iterateAll(iter)) return false;
-
- if (hashChild && !hashChild->iterateAll(iter)) return false;
-
- for (ChildMap::iterator ptr = childTokens.begin();
- ptr != childTokens.end(); ptr++) {
-
- if (!ptr->second->iterateAll(iter)) return false;
- }
-
- return true;
-}
-
-// applies iter against only matching nodes until iter returns false
-// Note Well: the iter may match against the same node more than once
-// if # wildcards are present!
-bool TopicExchange::BindingNode::iterateMatch(const std::string& routingKey, TreeIterator& iter)
-{
- TopicExchange::TokenIterator rKey(routingKey);
- return iterateMatch( rKey, iter );
-}
-
-
-// recurse over binding using token iterator.
-// Note well: bkey is modified!
-TopicExchange::BindingKey*
-TopicExchange::BindingNode::addBindingKey(TokenIterator &bKey,
- const string& fullPattern)
-{
- if (bKey.finished()) {
- // this node's binding
- if (routePattern.empty()) {
- routePattern = fullPattern;
- } else assert(routePattern == fullPattern);
-
- return &bindings;
-
- } else {
- // pop the topmost token & recurse...
-
- if (bKey.match(STAR)) {
- if (!starChild) {
- starChild.reset(new StarNode());
- }
- bKey.next();
- return starChild->addBindingKey(bKey, fullPattern);
-
- } else if (bKey.match(HASH)) {
- if (!hashChild) {
- hashChild.reset(new HashNode());
- }
- bKey.next();
- return hashChild->addBindingKey(bKey, fullPattern);
-
- } else {
- ChildMap::iterator ptr;
- std::string next_token;
- bKey.pop(next_token);
- ptr = childTokens.find(next_token);
- if (ptr != childTokens.end()) {
- return ptr->second->addBindingKey(bKey, fullPattern);
- } else {
- BindingNode::shared_ptr child(new BindingNode(next_token));
- childTokens[next_token] = child;
- return child->addBindingKey(bKey, fullPattern);
- }
- }
- }
-}
-
-
-// Remove a binding pattern from the tree. Return true if the current
-// node becomes a leaf without any bindings (therefore can be deleted).
-// Note Well: modifies parameter bKey's value!
-bool
-TopicExchange::BindingNode::removeBindingKey(TokenIterator &bKey,
- const string& fullPattern)
-{
- bool remove;
-
- if (!bKey.finished()) {
-
- if (bKey.match(STAR)) {
- bKey.next();
- if (starChild) {
- remove = starChild->removeBindingKey(bKey, fullPattern);
- if (remove) {
- starChild.reset();
- }
- }
- } else if (bKey.match(HASH)) {
- bKey.next();
- if (hashChild) {
- remove = hashChild->removeBindingKey(bKey, fullPattern);
- if (remove) {
- hashChild.reset();
- }
- }
- } else {
- ChildMap::iterator ptr;
- std::string next_token;
- bKey.pop(next_token);
- ptr = childTokens.find(next_token);
- if (ptr != childTokens.end()) {
- remove = ptr->second->removeBindingKey(bKey, fullPattern);
- if (remove) {
- childTokens.erase(ptr);
- }
- }
- }
- }
-
- // no bindings and no children == parent can delete this node.
- return getChildCount() == 0 && bindings.bindingVector.empty();
-}
-
-
-// find the binding key that matches the given binding pattern.
-// Note Well: modifies key parameter!
-TopicExchange::BindingKey*
-TopicExchange::BindingNode::getBindingKey(TokenIterator &key)
-{
- if (key.finished()) {
- return &bindings;
- }
-
- string next_token;
-
- key.pop(next_token);
-
- if (next_token == STAR) {
- if (starChild)
- return starChild->getBindingKey(key);
- } else if (next_token == HASH) {
- if (hashChild)
- return hashChild->getBindingKey(key);
- } else {
- ChildMap::iterator ptr;
- ptr = childTokens.find(next_token);
- if (ptr != childTokens.end()) {
- return ptr->second->getBindingKey(key);
- }
- }
-
- return 0;
-}
-
-
-
-// iterate over all nodes that match the given key. Note well: the set of nodes
-// that are visited includes matching non-leaf nodes.
-// Note well: parameter key is modified!
-bool TopicExchange::BindingNode::iterateMatch(TokenIterator& key, TreeIterator& iter)
-{
- // invariant: key has matched all previous tokens up to this node.
- if (key.finished()) {
- // exact match this node: visit if bound
- if (!bindings.bindingVector.empty())
- if (!iter.visit(*this)) return false;
- }
-
- // check remaining key against children, even if empty.
- return iterateMatchChildren(key, iter);
-}
-
-
-TopicExchange::StarNode::StarNode()
- : BindingNode(STAR) {}
-
-
-// See iterateMatch() above.
-// Special case: this node must verify a token is available (match exactly one).
-bool TopicExchange::StarNode::iterateMatch(TokenIterator& key, TreeIterator& iter)
-{
- // must match one token:
- if (key.finished())
- return true; // match failed, but continue iteration on siblings
-
- // pop the topmost token
- key.next();
-
- if (key.finished()) {
- // exact match this node: visit if bound
- if (!bindings.bindingVector.empty())
- if (!iter.visit(*this)) return false;
- }
-
- return iterateMatchChildren(key, iter);
-}
-
-
-TopicExchange::HashNode::HashNode()
- : BindingNode(HASH) {}
-
-
-// See iterateMatch() above.
-// Special case: can match zero or more tokens at the head of the key.
-bool TopicExchange::HashNode::iterateMatch(TokenIterator& key, TreeIterator& iter)
-{
- // consume each token and look for a match on the
- // remaining key.
- while (!key.finished()) {
- if (!iterateMatchChildren(key, iter)) return false;
- key.next();
- }
-
- if (!bindings.bindingVector.empty())
- return iter.visit(*this);
-
- return true;
-}
-
-
-// helper: iterate over current node's matching children
-bool
-TopicExchange::BindingNode::iterateMatchChildren(const TopicExchange::TokenIterator& key,
- TopicExchange::BindingNode::TreeIterator& iter)
-{
- // always try glob - it can match empty keys
- if (hashChild) {
- TokenIterator tmp(key);
- if (!hashChild->iterateMatch(tmp, iter))
- return false;
- }
-
- if (!key.finished()) {
-
- if (starChild) {
- TokenIterator tmp(key);
- if (!starChild->iterateMatch(tmp, iter))
- return false;
- }
-
- if (!childTokens.empty()) {
- TokenIterator newKey(key);
- std::string next_token;
- newKey.pop(next_token);
-
- ChildMap::iterator ptr = childTokens.find(next_token);
- if (ptr != childTokens.end()) {
- return ptr->second->iterateMatch(newKey, iter);
- }
- }
- }
-
- return true;
-}
-
}} // namespace qpid::broker
Modified: qpid/branches/asyncstore/cpp/src/qpid/broker/TopicExchange.h
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/broker/TopicExchange.h?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/broker/TopicExchange.h (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/broker/TopicExchange.h Fri Aug 3 12:13:32 2012
@@ -28,6 +28,7 @@
#include "qpid/framing/FieldTable.h"
#include "qpid/sys/Monitor.h"
#include "qpid/broker/Queue.h"
+#include "qpid/broker/TopicKeyNode.h"
namespace qpid {
@@ -35,7 +36,6 @@ namespace broker {
class TopicExchange : public virtual Exchange {
- struct TokenIterator;
class Normalizer;
struct BindingKey { // binding for this node
@@ -43,129 +43,44 @@ class TopicExchange : public virtual Exc
FedBinding fedBinding;
};
- // Binding database:
- // The dotted form of a binding key is broken up and stored in a directed tree graph.
- // Common binding prefix are merged. This allows the route match alogrithm to quickly
- // isolate those sub-trees that match a given routingKey.
- // For example, given the routes:
- // a.b.c.<...>
- // a.b.d.<...>
- // a.x.y.<...>
- // The resulting tree would be:
- // a-->b-->c-->...
- // | +-->d-->...
- // +-->x-->y-->...
- //
- class QPID_BROKER_CLASS_EXTERN BindingNode {
- public:
-
- typedef boost::shared_ptr<BindingNode> shared_ptr;
-
- // for database transversal (visit a node).
- class TreeIterator {
- public:
- TreeIterator() {};
- virtual ~TreeIterator() {};
- virtual bool visit(BindingNode& node) = 0;
- };
-
- BindingNode() {};
- BindingNode(const std::string& token) : token(token) {};
- QPID_BROKER_EXTERN virtual ~BindingNode();
-
- // add normalizedRoute to tree, return associated BindingKey
- QPID_BROKER_EXTERN BindingKey* addBindingKey(const std::string& normalizedRoute);
-
- // return BindingKey associated with normalizedRoute
- QPID_BROKER_EXTERN BindingKey* getBindingKey(const std::string& normalizedRoute);
-
- // remove BindingKey associated with normalizedRoute
- QPID_BROKER_EXTERN void removeBindingKey(const std::string& normalizedRoute);
-
- // applies iter against each node in tree until iter returns false
- QPID_BROKER_EXTERN bool iterateAll(TreeIterator& iter);
-
- // applies iter against only matching nodes until iter returns false
- QPID_BROKER_EXTERN bool iterateMatch(const std::string& routingKey, TreeIterator& iter);
-
- std::string routePattern; // normalized binding that matches this node
- BindingKey bindings; // for matches against this node
-
- protected:
-
- std::string token; // portion of pattern represented by this node
-
- // children
- typedef std::map<const std::string, BindingNode::shared_ptr> ChildMap;
- ChildMap childTokens;
- BindingNode::shared_ptr starChild; // "*" subtree
- BindingNode::shared_ptr hashChild; // "#" subtree
-
- unsigned int getChildCount() { return childTokens.size() +
- (starChild ? 1 : 0) + (hashChild ? 1 : 0); }
- BindingKey* addBindingKey(TokenIterator& bKey,
- const std::string& fullPattern);
- bool removeBindingKey(TokenIterator& bKey,
- const std::string& fullPattern);
- BindingKey* getBindingKey(TokenIterator& bKey);
- QPID_BROKER_EXTERN virtual bool iterateMatch(TokenIterator& rKey, TreeIterator& iter);
- bool iterateMatchChildren(const TokenIterator& key, TreeIterator& iter);
- };
-
- // Special case: ("*" token) Node in the tree for a match exactly one wildcard
- class StarNode : public BindingNode {
- public:
- StarNode();
- ~StarNode() {};
+ typedef TopicKeyNode<BindingKey> BindingNode;
- protected:
- virtual bool iterateMatch(TokenIterator& key, TreeIterator& iter);
- };
-
- // Special case: ("#" token) Node in the tree for a match zero or more
- class HashNode : public BindingNode {
- public:
- HashNode();
- ~HashNode() {};
+ BindingKey *getQueueBinding(Queue::shared_ptr queue, const std::string& pattern);
+ bool deleteBinding(Queue::shared_ptr queue,
+ const std::string& routingKey,
+ BindingKey *bk);
- protected:
- virtual bool iterateMatch(TokenIterator& key, TreeIterator& iter);
- };
+ class ReOriginIter;
+ class BindingsFinderIter;
+ class QueueFinderIter;
BindingNode bindingTree;
unsigned long nBindings;
qpid::sys::RWlock lock; // protects bindingTree and nBindings
qpid::sys::RWlock cacheLock; // protects cache
std::map<std::string, BindingList> bindingCache; // cache of matched routes.
+
class ClearCache {
private:
qpid::sys::RWlock* cacheLock;
std::map<std::string, BindingList>* bindingCache;
- bool cleared;
+ bool cleared;
public:
- ClearCache(qpid::sys::RWlock* l, std::map<std::string, BindingList>* bc): cacheLock(l),
- bindingCache(bc),cleared(false) {};
+ ClearCache(qpid::sys::RWlock* l, std::map<std::string, BindingList>* bc) :
+ cacheLock(l), bindingCache(bc),cleared(false) {};
void clearCache() {
- qpid::sys::RWlock::ScopedWlock l(*cacheLock);
- if (!cleared) {
- bindingCache->clear();
- cleared =true;
- }
+ qpid::sys::RWlock::ScopedWlock l(*cacheLock);
+ if (!cleared) {
+ bindingCache->clear();
+ cleared =true;
+ }
};
~ClearCache(){
- clearCache();
+ clearCache();
};
};
- BindingKey *getQueueBinding(Queue::shared_ptr queue, const std::string& pattern);
- bool deleteBinding(Queue::shared_ptr queue,
- const std::string& routingKey,
- BindingKey *bk);
- class ReOriginIter;
- class BindingsFinderIter;
- class QueueFinderIter;
-
- public:
+public:
static const std::string typeName;
static QPID_BROKER_EXTERN std::string normalize(const std::string& pattern);
@@ -199,7 +114,6 @@ class TopicExchange : public virtual Exc
};
-
}
}
Modified: qpid/branches/asyncstore/cpp/src/qpid/broker/windows/SaslAuthenticator.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/broker/windows/SaslAuthenticator.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/broker/windows/SaslAuthenticator.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/broker/windows/SaslAuthenticator.cpp Fri Aug 3 12:13:32 2012
@@ -32,6 +32,8 @@
using namespace qpid::framing;
using qpid::sys::SecurityLayer;
+using std::string;
+
namespace qpid {
namespace broker {
@@ -79,7 +81,7 @@ void SaslAuthenticator::fini(void)
return;
}
-std::auto_ptr<SaslAuthenticator> SaslAuthenticator::createAuthenticator(Connection& c, bool)
+std::auto_ptr<SaslAuthenticator> SaslAuthenticator::createAuthenticator(Connection& c)
{
if (c.getBroker().getOptions().auth) {
return std::auto_ptr<SaslAuthenticator>(new SspiAuthenticator(c));
Modified: qpid/branches/asyncstore/cpp/src/qpid/broker/windows/SslProtocolFactory.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/broker/windows/SslProtocolFactory.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/broker/windows/SslProtocolFactory.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/broker/windows/SslProtocolFactory.cpp Fri Aug 3 12:13:32 2012
@@ -44,26 +44,34 @@
namespace qpid {
namespace sys {
+
+class Timer;
+
namespace windows {
struct SslServerOptions : qpid::Options
{
std::string certStore;
+ std::string certStoreLocation;
std::string certName;
uint16_t port;
bool clientAuth;
SslServerOptions() : qpid::Options("SSL Options"),
- certStore("My"), port(5671), clientAuth(false)
+ certStore("My"),
+ certStoreLocation("CurrentUser"),
+ certName("localhost"),
+ port(5671),
+ clientAuth(false)
{
qpid::Address me;
if (qpid::sys::SystemInfo::getLocalHostname(me))
certName = me.host;
- else
- certName = "localhost";
addOptions()
("ssl-cert-store", optValue(certStore, "NAME"), "Local store name from which to obtain certificate")
+ ("ssl-cert-store-location", optValue(certStoreLocation, "NAME"),
+ "Local store name location for certificates ( CurrentUser | LocalMachine | CurrentService )")
("ssl-cert-name", optValue(certName, "NAME"), "Name of the certificate to use")
("ssl-port", optValue(port, "PORT"), "Port on which to listen for SSL connections")
("ssl-require-client-authentication", optValue(clientAuth),
@@ -72,10 +80,12 @@ struct SslServerOptions : qpid::Options
};
class SslProtocolFactory : public qpid::sys::ProtocolFactory {
- const bool tcpNoDelay;
boost::ptr_vector<Socket> listeners;
boost::ptr_vector<AsynchAcceptor> acceptors;
+ Timer& brokerTimer;
+ uint32_t maxNegotiateTime;
uint16_t listeningPort;
+ const bool tcpNoDelay;
std::string brokerHost;
const bool clientAuthSelected;
std::auto_ptr<qpid::sys::AsynchAcceptor> acceptor;
@@ -83,7 +93,9 @@ class SslProtocolFactory : public qpid::
CredHandle credHandle;
public:
- SslProtocolFactory(const SslServerOptions&, const std::string& host, const std::string& port, int backlog, bool nodelay);
+ SslProtocolFactory(const SslServerOptions&, const std::string& host, const std::string& port,
+ int backlog, bool nodelay,
+ Timer& timer, uint32_t maxTime);
~SslProtocolFactory();
void accept(sys::Poller::shared_ptr, sys::ConnectionCodec::Factory*);
void connect(sys::Poller::shared_ptr, const std::string& host, const std::string& port,
@@ -120,8 +132,8 @@ static struct SslPlugin : public Plugin
const broker::Broker::Options& opts = broker->getOptions();
ProtocolFactory::shared_ptr protocol(new SslProtocolFactory(options,
"", boost::lexical_cast<std::string>(options.port),
- opts.connectionBacklog,
- opts.tcpNoDelay));
+ opts.connectionBacklog, opts.tcpNoDelay,
+ broker->getTimer(), opts.maxNegotiateTime));
QPID_LOG(notice, "Listening for SSL connections on TCP port " << protocol->getPort());
broker->registerProtocolFactory("ssl", protocol);
} catch (const std::exception& e) {
@@ -132,9 +144,12 @@ static struct SslPlugin : public Plugin
} sslPlugin;
SslProtocolFactory::SslProtocolFactory(const SslServerOptions& options,
- const std::string& host, const std::string& port, int backlog,
- bool nodelay)
- : tcpNoDelay(nodelay),
+ const std::string& host, const std::string& port,
+ int backlog, bool nodelay,
+ Timer& timer, uint32_t maxTime)
+ : brokerTimer(timer),
+ maxNegotiateTime(maxTime),
+ tcpNoDelay(nodelay),
clientAuthSelected(options.clientAuth) {
// Make sure that certificate store is good before listening to sockets
@@ -142,11 +157,25 @@ SslProtocolFactory::SslProtocolFactory(c
SecInvalidateHandle(&credHandle);
// Get the certificate for this server.
+ DWORD flags = 0;
+ std::string certStoreLocation = options.certStoreLocation;
+ std::transform(certStoreLocation.begin(), certStoreLocation.end(), certStoreLocation.begin(), ::tolower);
+ if (certStoreLocation == "currentuser") {
+ flags = CERT_SYSTEM_STORE_CURRENT_USER;
+ } else if (certStoreLocation == "localmachine") {
+ flags = CERT_SYSTEM_STORE_LOCAL_MACHINE;
+ } else if (certStoreLocation == "currentservice") {
+ flags = CERT_SYSTEM_STORE_CURRENT_SERVICE;
+ } else {
+ QPID_LOG(error, "Unrecognised SSL certificate store location: " << options.certStoreLocation
+ << " - Using default location");
+ }
HCERTSTORE certStoreHandle;
certStoreHandle = ::CertOpenStore(CERT_STORE_PROV_SYSTEM_A,
X509_ASN_ENCODING,
0,
- CERT_SYSTEM_STORE_LOCAL_MACHINE,
+ flags |
+ CERT_STORE_READONLY_FLAG,
options.certStore.c_str());
if (!certStoreHandle)
throw qpid::Exception(QPID_MSG("Opening store " << options.certStore << " " << qpid::sys::strError(GetLastError())));
@@ -252,7 +281,7 @@ void SslProtocolFactory::established(sys
boost::bind(&AsynchIOHandler::idle, async, _1));
}
- async->init(aio, 4);
+ async->init(aio, brokerTimer, maxNegotiateTime, 4);
aio->start(poller);
}
Modified: qpid/branches/asyncstore/cpp/src/qpid/client/Connection.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/client/Connection.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/client/Connection.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/client/Connection.cpp Fri Aug 3 12:13:32 2012
@@ -136,7 +136,7 @@ const ConnectionSettings& Connection::ge
Session Connection::newSession(const std::string& name, uint32_t timeout) {
if (!isOpen())
- throw Exception(QPID_MSG("Connection has not yet been opened"));
+ throw TransportFailure("Can't create session, connection is not open");
Session s;
SessionBase_0_10Access(s).set(impl->newSession(name, timeout));
return s;
Modified: qpid/branches/asyncstore/cpp/src/qpid/client/ConnectionHandler.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/client/ConnectionHandler.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/client/ConnectionHandler.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/client/ConnectionHandler.cpp Fri Aug 3 12:13:32 2012
@@ -258,7 +258,7 @@ void ConnectionHandler::start(const Fiel
}
if (sasl.get()) {
- string response;
+ std::string response;
if (sasl->start(join(mechlist), response, getSecuritySettings ? getSecuritySettings() : 0)) {
proxy.startOk(properties, sasl->getMechanism(), response, locale);
} else {
@@ -272,7 +272,7 @@ void ConnectionHandler::start(const Fiel
}
} else {
//TODO: verify that desired mechanism and locale are supported
- string response = ((char)0) + username + ((char)0) + password;
+ std::string response = ((char)0) + username + ((char)0) + password;
proxy.startOk(properties, mechanism, response, locale);
}
}
@@ -280,7 +280,7 @@ void ConnectionHandler::start(const Fiel
void ConnectionHandler::secure(const std::string& challenge)
{
if (sasl.get()) {
- string response = sasl->step(challenge);
+ std::string response = sasl->step(challenge);
proxy.secureOk(response);
} else {
throw NotImplementedException("Challenge-response cycle not yet implemented in client");
Modified: qpid/branches/asyncstore/cpp/src/qpid/client/PrivateImplRef.h
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/client/PrivateImplRef.h?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/client/PrivateImplRef.h (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/client/PrivateImplRef.h Fri Aug 3 12:13:32 2012
@@ -77,15 +77,15 @@ template <class T> class PrivateImplRef
static void set(T& t, const intrusive_ptr& p) {
if (t.impl == p) return;
- if (t.impl) boost::intrusive_ptr_release(t.impl);
+ if (t.impl) intrusive_ptr_release(t.impl);
t.impl = p.get();
- if (t.impl) boost::intrusive_ptr_add_ref(t.impl);
+ if (t.impl) intrusive_ptr_add_ref(t.impl);
}
// Helper functions to implement the ctor, dtor, copy, assign
- static void ctor(T& t, Impl* p) { t.impl = p; if (p) boost::intrusive_ptr_add_ref(p); }
+ static void ctor(T& t, Impl* p) { t.impl = p; if (p) intrusive_ptr_add_ref(p); }
static void copy(T& t, const T& x) { if (&t == &x) return; t.impl = 0; assign(t, x); }
- static void dtor(T& t) { if(t.impl) boost::intrusive_ptr_release(t.impl); }
+ static void dtor(T& t) { if(t.impl) intrusive_ptr_release(t.impl); }
static T& assign(T& t, const T& x) { set(t, get(x)); return t;}
};
Modified: qpid/branches/asyncstore/cpp/src/qpid/client/SessionImpl.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/client/SessionImpl.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/client/SessionImpl.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/client/SessionImpl.cpp Fri Aug 3 12:13:32 2012
@@ -426,7 +426,7 @@ void SessionImpl::sendContent(const Meth
uint32_t remaining = data_length - offset;
while (remaining > 0) {
uint32_t length = remaining > frag_size ? frag_size : remaining;
- string frag(content.getData().substr(offset, length));
+ std::string frag(content.getData().substr(offset, length));
AMQFrame frame((AMQContentBody(frag)));
frame.setFirstSegment(false);
frame.setLastSegment(true);
Modified: qpid/branches/asyncstore/cpp/src/qpid/client/SslConnector.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/client/SslConnector.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/client/SslConnector.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/client/SslConnector.cpp Fri Aug 3 12:13:32 2012
@@ -94,8 +94,6 @@ class SslConnector : public Connector
sys::ShutdownHandler* shutdownHandler;
framing::InputHandler* input;
- framing::InitiationHandler* initialiser;
- framing::OutputHandler* output;
Writer writer;
@@ -176,6 +174,7 @@ SslConnector::SslConnector(Poller::share
initiated(false),
closed(true),
shutdownHandler(0),
+ input(0),
writer(maxFrameSize, cimpl),
aio(0),
poller(p)
Modified: qpid/branches/asyncstore/cpp/src/qpid/client/SubscriptionManagerImpl.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/client/SubscriptionManagerImpl.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/client/SubscriptionManagerImpl.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/client/SubscriptionManagerImpl.cpp Fri Aug 3 12:13:32 2012
@@ -39,6 +39,16 @@ SubscriptionManagerImpl::SubscriptionMan
: dispatcher(s), session(s), autoStop(true)
{}
+SubscriptionManagerImpl::~SubscriptionManagerImpl()
+{
+ sys::Mutex::ScopedLock l(lock);
+ for (std::map<std::string, Subscription>::iterator i = subscriptions.begin(); i != subscriptions.end(); ++i) {
+ boost::intrusive_ptr<SubscriptionImpl> s = PrivateImplRef<Subscription>::get(i->second);
+ if (s) s->cancelDiversion();
+ }
+ subscriptions.clear();
+}
+
Subscription SubscriptionManagerImpl::subscribe(
MessageListener& listener, const std::string& q, const SubscriptionSettings& ss, const std::string& n)
{
Modified: qpid/branches/asyncstore/cpp/src/qpid/client/SubscriptionManagerImpl.h
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/client/SubscriptionManagerImpl.h?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/client/SubscriptionManagerImpl.h (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/client/SubscriptionManagerImpl.h Fri Aug 3 12:13:32 2012
@@ -99,7 +99,8 @@ class SubscriptionManagerImpl : public s
public:
/** Create a new SubscriptionManagerImpl associated with a session */
SubscriptionManagerImpl(const Session& session);
-
+ ~SubscriptionManagerImpl();
+
/**
* Subscribe a MessagesListener to receive messages from queue.
*
Modified: qpid/branches/asyncstore/cpp/src/qpid/client/TCPConnector.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/client/TCPConnector.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/client/TCPConnector.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/client/TCPConnector.cpp Fri Aug 3 12:13:32 2012
@@ -76,6 +76,7 @@ TCPConnector::TCPConnector(Poller::share
initiated(false),
closed(true),
shutdownHandler(0),
+ input(0),
connector(0),
aio(0),
poller(p)
@@ -265,7 +266,7 @@ size_t TCPConnector::encode(const char*
return bytesWritten;
}
-bool TCPConnector::readbuff(AsynchIO& aio, AsynchIO::BufferBase* buff)
+void TCPConnector::readbuff(AsynchIO& aio, AsynchIO::BufferBase* buff)
{
Codec* codec = securityLayer.get() ? (Codec*) securityLayer.get() : (Codec*) this;
int32_t decoded = codec->decode(buff->bytes+buff->dataStart, buff->dataCount);
@@ -280,10 +281,9 @@ bool TCPConnector::readbuff(AsynchIO& ai
// Give whole buffer back to aio subsystem
aio.queueReadBuffer(buff);
}
- return true;
}
-size_t TCPConnector::decode(const char* buffer, size_t size)
+size_t TCPConnector::decode(const char* buffer, size_t size)
{
framing::Buffer in(const_cast<char*>(buffer), size);
if (!initiated) {
Modified: qpid/branches/asyncstore/cpp/src/qpid/client/TCPConnector.h
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/client/TCPConnector.h?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/client/TCPConnector.h (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/client/TCPConnector.h Fri Aug 3 12:13:32 2012
@@ -66,8 +66,6 @@ class TCPConnector : public Connector, p
sys::ShutdownHandler* shutdownHandler;
framing::InputHandler* input;
- framing::InitiationHandler* initialiser;
- framing::OutputHandler* output;
sys::Socket socket;
@@ -102,7 +100,7 @@ protected:
void start(sys::AsynchIO* aio_);
void initAmqp();
virtual void connectFailed(const std::string& msg);
- bool readbuff(qpid::sys::AsynchIO&, qpid::sys::AsynchIOBufferBase*);
+ void readbuff(qpid::sys::AsynchIO&, qpid::sys::AsynchIOBufferBase*);
void writebuff(qpid::sys::AsynchIO&);
void eof(qpid::sys::AsynchIO&);
void disconnected(qpid::sys::AsynchIO&);
Modified: qpid/branches/asyncstore/cpp/src/qpid/client/amqp0_10/AddressResolution.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/client/amqp0_10/AddressResolution.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/client/amqp0_10/AddressResolution.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/client/amqp0_10/AddressResolution.cpp Fri Aug 3 12:13:32 2012
@@ -241,6 +241,7 @@ class Subscription : public Exchange, pu
const std::string actualType;
const bool exclusiveQueue;
const bool exclusiveSubscription;
+ const std::string alternateExchange;
FieldTable queueOptions;
FieldTable subscriptionOptions;
Bindings bindings;
@@ -507,7 +508,8 @@ Subscription::Subscription(const Address
durable(Opt(address)/LINK/DURABLE),
actualType(type.empty() ? (specifiedType.empty() ? TOPIC_EXCHANGE : specifiedType) : type),
exclusiveQueue((Opt(address)/LINK/X_DECLARE/EXCLUSIVE).asBool(true)),
- exclusiveSubscription((Opt(address)/LINK/X_SUBSCRIBE/EXCLUSIVE).asBool(exclusiveQueue))
+ exclusiveSubscription((Opt(address)/LINK/X_SUBSCRIBE/EXCLUSIVE).asBool(exclusiveQueue)),
+ alternateExchange((Opt(address)/LINK/X_DECLARE/ALTERNATE_EXCHANGE).str())
{
(Opt(address)/LINK/X_DECLARE/ARGUMENTS).collect(queueOptions);
(Opt(address)/LINK/X_SUBSCRIBE/ARGUMENTS).collect(subscriptionOptions);
@@ -568,7 +570,9 @@ void Subscription::subscribe(qpid::clien
//create subscription queue:
session.queueDeclare(arg::queue=queue, arg::exclusive=exclusiveQueue,
- arg::autoDelete=!reliable, arg::durable=durable, arg::arguments=queueOptions);
+ arg::autoDelete=!reliable, arg::durable=durable,
+ arg::alternateExchange=alternateExchange,
+ arg::arguments=queueOptions);
//'default' binding:
bindings.bind(session);
//any explicit bindings:
Modified: qpid/branches/asyncstore/cpp/src/qpid/client/amqp0_10/ConnectionImpl.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/client/amqp0_10/ConnectionImpl.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/client/amqp0_10/ConnectionImpl.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/client/amqp0_10/ConnectionImpl.cpp Fri Aug 3 12:13:32 2012
@@ -78,7 +78,7 @@ bool expired(const sys::AbsTime& start,
if (timeout == 0) return true;
if (timeout == FOREVER) return false;
sys::Duration used(start, sys::now());
- sys::Duration allowed(int64_t(timeout*sys::TIME_SEC));
+ sys::Duration allowed((int64_t)(timeout*sys::TIME_SEC));
return allowed < used;
}
Modified: qpid/branches/asyncstore/cpp/src/qpid/client/amqp0_10/OutgoingMessage.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/client/amqp0_10/OutgoingMessage.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/client/amqp0_10/OutgoingMessage.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/client/amqp0_10/OutgoingMessage.cpp Fri Aug 3 12:13:32 2012
@@ -58,7 +58,12 @@ void OutgoingMessage::convert(const qpid
if (address) {
message.getMessageProperties().setReplyTo(AddressResolution::convert(address));
}
- translate(from.getProperties(), message.getMessageProperties().getApplicationHeaders());
+ if (!subject.empty()) {
+ Variant v(subject); v.setEncoding("utf8");
+ translate(from.getProperties(), SUBJECT, v, message.getMessageProperties().getApplicationHeaders());
+ } else {
+ translate(from.getProperties(), message.getMessageProperties().getApplicationHeaders());
+ }
if (from.getTtl().getMilliseconds()) {
message.getDeliveryProperties().setTtl(from.getTtl().getMilliseconds());
}
@@ -89,16 +94,14 @@ void OutgoingMessage::convert(const qpid
}
}
-void OutgoingMessage::setSubject(const std::string& subject)
+void OutgoingMessage::setSubject(const std::string& s)
{
- if (!subject.empty()) {
- message.getMessageProperties().getApplicationHeaders().setString(SUBJECT, subject);
- }
+ subject = s;
}
std::string OutgoingMessage::getSubject() const
{
- return message.getMessageProperties().getApplicationHeaders().getAsString(SUBJECT);
+ return subject;
}
}}} // namespace qpid::client::amqp0_10
Modified: qpid/branches/asyncstore/cpp/src/qpid/client/amqp0_10/OutgoingMessage.h
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/client/amqp0_10/OutgoingMessage.h?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/client/amqp0_10/OutgoingMessage.h (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/client/amqp0_10/OutgoingMessage.h Fri Aug 3 12:13:32 2012
@@ -35,6 +35,7 @@ struct OutgoingMessage
{
qpid::client::Message message;
qpid::client::Completion status;
+ std::string subject;
void convert(const qpid::messaging::Message&);
void setSubject(const std::string& subject);
Modified: qpid/branches/asyncstore/cpp/src/qpid/client/amqp0_10/ReceiverImpl.h
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/client/amqp0_10/ReceiverImpl.h?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/client/amqp0_10/ReceiverImpl.h (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/client/amqp0_10/ReceiverImpl.h Fri Aug 3 12:13:32 2012
@@ -78,7 +78,6 @@ class ReceiverImpl : public qpid::messag
std::auto_ptr<MessageSource> source;
uint32_t capacity;
qpid::client::AsyncSession session;
- qpid::messaging::MessageListener* listener;
uint32_t window;
void startFlow(const sys::Mutex::ScopedLock&); // Dummy param, call with lock held
Modified: qpid/branches/asyncstore/cpp/src/qpid/client/amqp0_10/SenderImpl.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/client/amqp0_10/SenderImpl.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/client/amqp0_10/SenderImpl.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/client/amqp0_10/SenderImpl.cpp Fri Aug 3 12:13:32 2012
@@ -37,10 +37,10 @@ SenderImpl::SenderImpl(SessionImpl& _par
void SenderImpl::send(const qpid::messaging::Message& message, bool sync)
{
if (unreliable) { // immutable, don't need lock
- UnreliableSend f(*this, &message);
+ UnreliableSend f(*this, message);
parent->execute(f);
} else {
- Send f(*this, &message);
+ Send f(*this, message);
while (f.repeat) parent->execute(f);
}
if (sync) parent->sync(true);
@@ -117,8 +117,8 @@ void SenderImpl::sendImpl(const qpid::me
{
sys::Mutex::ScopedLock l(lock);
std::auto_ptr<OutgoingMessage> msg(new OutgoingMessage());
- msg->convert(m);
msg->setSubject(m.getSubject().empty() ? address.getSubject() : m.getSubject());
+ msg->convert(m);
outgoing.push_back(msg.release());
sink->send(session, name, outgoing.back());
}
@@ -127,8 +127,8 @@ void SenderImpl::sendUnreliable(const qp
{
sys::Mutex::ScopedLock l(lock);
OutgoingMessage msg;
- msg.convert(m);
msg.setSubject(m.getSubject().empty() ? address.getSubject() : m.getSubject());
+ msg.convert(m);
sink->send(session, name, msg);
}
Modified: qpid/branches/asyncstore/cpp/src/qpid/client/amqp0_10/SenderImpl.h
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/client/amqp0_10/SenderImpl.h?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/client/amqp0_10/SenderImpl.h (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/client/amqp0_10/SenderImpl.h Fri Aug 3 12:13:32 2012
@@ -99,32 +99,32 @@ class SenderImpl : public qpid::messagin
struct Send : Command
{
- const qpid::messaging::Message* message;
+ const qpid::messaging::Message& message;
bool repeat;
- Send(SenderImpl& i, const qpid::messaging::Message* m) : Command(i), message(m), repeat(true) {}
+ Send(SenderImpl& i, const qpid::messaging::Message& m) : Command(i), message(m), repeat(true) {}
void operator()()
{
impl.waitForCapacity();
//from this point message will be recorded if there is any
//failure (and replayed) so need not repeat the call
repeat = false;
- impl.sendImpl(*message);
+ impl.sendImpl(message);
}
};
struct UnreliableSend : Command
{
- const qpid::messaging::Message* message;
+ const qpid::messaging::Message& message;
- UnreliableSend(SenderImpl& i, const qpid::messaging::Message* m) : Command(i), message(m) {}
+ UnreliableSend(SenderImpl& i, const qpid::messaging::Message& m) : Command(i), message(m) {}
void operator()()
{
//TODO: ideally want to put messages on the outbound
//queue and pull them off in io thread, but the old
//0-10 client doesn't support that option so for now
//we simply don't queue unreliable messages
- impl.sendUnreliable(*message);
+ impl.sendUnreliable(message);
}
};
Modified: qpid/branches/asyncstore/cpp/src/qpid/client/windows/SslConnector.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/client/windows/SslConnector.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/client/windows/SslConnector.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/client/windows/SslConnector.cpp Fri Aug 3 12:13:32 2012
@@ -68,7 +68,7 @@ class SslConnector : public qpid::client
// A number of AsynchIO callbacks go right through to TCPConnector, but
// we can't boost::bind to a protected ancestor, so these methods redirect
// to those TCPConnector methods.
- bool redirectReadbuff(qpid::sys::AsynchIO&, qpid::sys::AsynchIOBufferBase*);
+ void redirectReadbuff(qpid::sys::AsynchIO&, qpid::sys::AsynchIOBufferBase*);
void redirectWritebuff(qpid::sys::AsynchIO&);
void redirectEof(qpid::sys::AsynchIO&);
@@ -111,9 +111,9 @@ void SslConnector::negotiationDone(SECUR
connectFailed(QPID_MSG(qpid::sys::strError(status)));
}
-bool SslConnector::redirectReadbuff(qpid::sys::AsynchIO& a,
+void SslConnector::redirectReadbuff(qpid::sys::AsynchIO& a,
qpid::sys::AsynchIOBufferBase* b) {
- return readbuff(a, b);
+ readbuff(a, b);
}
void SslConnector::redirectWritebuff(qpid::sys::AsynchIO& a) {
Modified: qpid/branches/asyncstore/cpp/src/qpid/cluster/Connection.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/cluster/Connection.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/cluster/Connection.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/cluster/Connection.cpp Fri Aug 3 12:13:32 2012
@@ -58,6 +58,8 @@
namespace qpid {
namespace cluster {
+using std::string;
+
using namespace framing;
using namespace framing::cluster;
using amqp_0_10::ListCodec;
@@ -83,7 +85,9 @@ Connection::Connection(Cluster& c, sys::
const std::string& mgmtId,
const ConnectionId& id, const qpid::sys::SecuritySettings& external)
: cluster(c), self(id), catchUp(false), announced(false), output(*this, out),
- connectionCtor(&output, cluster.getBroker(), mgmtId, external, false, 0, true),
+ connectionCtor(&output, cluster.getBroker(), mgmtId, external,
+ false/*isLink*/, 0/*objectId*/, true/*shadow*/, false/*delayManagement*/,
+ false/*authenticated*/),
expectProtocolHeader(false),
mcastFrameHandler(cluster.getMulticast(), self),
updateIn(c.getUpdateReceiver()),
@@ -100,9 +104,10 @@ Connection::Connection(Cluster& c, sys::
external,
isLink,
isCatchUp ? ++catchUpId : 0,
- // The first catch-up connection is not considered a shadow
- // as it needs to be authenticated.
- isCatchUp && self.second > 1),
+ // The first catch-up connection is not a shadow
+ isCatchUp && self.second > 1,
+ false, // delayManagement
+ true), // catch up connecytions are authenticated
expectProtocolHeader(isLink),
mcastFrameHandler(cluster.getMulticast(), self),
updateIn(c.getUpdateReceiver()),
@@ -272,6 +277,8 @@ void Connection::closed() {
if (announced)
cluster.getMulticast().mcastControl(
ClusterConnectionDeliverCloseBody(), self);
+ else
+ close();
}
}
catch (const std::exception& e) {
@@ -404,11 +411,12 @@ void Connection::shadowSetUser(const std
}
void Connection::consumerState(const string& name, bool blocked, bool notifyEnabled, const SequenceNumber& position,
- uint32_t usedMsgCredit, uint32_t usedByteCredit)
+ uint32_t usedMsgCredit, uint32_t usedByteCredit, const uint32_t deliveryCount)
{
broker::SemanticState::ConsumerImpl::shared_ptr c = semanticState().find(name);
c->setPosition(position);
c->setBlocked(blocked);
+ c->setDeliveryCount(deliveryCount);
if (c->getCredit().isWindowMode()) c->getCredit().consume(usedMsgCredit, usedByteCredit);
if (notifyEnabled) c->enableNotify(); else c->disableNotify();
updateIn.consumerNumbering.add(c);
@@ -522,6 +530,7 @@ broker::QueuedMessage Connection::getUpd
boost::shared_ptr<broker::Queue> updateq = findQueue(UpdateClient::UPDATE);
assert(!updateq->isDurable());
broker::QueuedMessage m = updateq->get();
+ updateq->dequeue(0, m);
if (!m.payload) throw Exception(QPID_MSG(cluster << " empty update queue"));
return m;
}
@@ -782,16 +791,18 @@ void Connection::managementSetupState(
void Connection::config(const std::string& encoded) {
Buffer buf(const_cast<char*>(encoded.data()), encoded.size());
string kind;
+ uint32_t p = buf.getPosition();
buf.getShortString (kind);
- if (kind == "link") {
+ buf.setPosition(p);
+ if (broker::Link::isEncodedLink(kind)) {
broker::Link::shared_ptr link =
- broker::Link::decode(cluster.getBroker().getLinks(), buf);
+ broker::Link::decode(cluster.getBroker().getLinks(), buf);
QPID_LOG(debug, cluster << " updated link "
<< link->getHost() << ":" << link->getPort());
}
- else if (kind == "bridge") {
+ else if (broker::Bridge::isEncodedBridge(kind)) {
broker::Bridge::shared_ptr bridge =
- broker::Bridge::decode(cluster.getBroker().getLinks(), buf);
+ broker::Bridge::decode(cluster.getBroker().getLinks(), buf);
QPID_LOG(debug, cluster << " updated bridge " << bridge->getName());
}
else throw Exception(QPID_MSG("Update failed, invalid kind of config: " << kind));
Modified: qpid/branches/asyncstore/cpp/src/qpid/cluster/Connection.h
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/cluster/Connection.h?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/cluster/Connection.h (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/cluster/Connection.h Fri Aug 3 12:13:32 2012
@@ -110,7 +110,7 @@ class Connection :
void deliveredFrame(const EventFrame&);
void consumerState(const std::string& name, bool blocked, bool notifyEnabled, const qpid::framing::SequenceNumber& position,
- uint32_t usedMsgCredit, uint32_t usedByteCredit);
+ uint32_t usedMsgCredit, uint32_t usedByteCredit, const uint32_t deliveryCount);
// ==== Used in catch-up mode to build initial state.
//
@@ -228,6 +228,7 @@ class Connection :
uint64_t objectId;
bool shadow;
bool delayManagement;
+ bool authenticated;
ConnectionCtor(
sys::ConnectionOutputHandler* out_,
@@ -237,17 +238,18 @@ class Connection :
bool isLink_=false,
uint64_t objectId_=0,
bool shadow_=false,
- bool delayManagement_=false
+ bool delayManagement_=false,
+ bool authenticated_=true
) : out(out_), broker(broker_), mgmtId(mgmtId_), external(external_),
isLink(isLink_), objectId(objectId_), shadow(shadow_),
- delayManagement(delayManagement_)
+ delayManagement(delayManagement_), authenticated(authenticated_)
{}
std::auto_ptr<broker::Connection> construct() {
return std::auto_ptr<broker::Connection>(
new broker::Connection(
out, broker, mgmtId, external, isLink, objectId,
- shadow, delayManagement)
+ shadow, delayManagement, authenticated)
);
}
};
Modified: qpid/branches/asyncstore/cpp/src/qpid/cluster/ConnectionCodec.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/cluster/ConnectionCodec.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/cluster/ConnectionCodec.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/cluster/ConnectionCodec.cpp Fri Aug 3 12:13:32 2012
@@ -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
@@ -22,6 +22,7 @@
#include "qpid/cluster/Connection.h"
#include "qpid/cluster/Cluster.h"
#include "qpid/cluster/ProxyInputHandler.h"
+#include "qpid/broker/AclModule.h"
#include "qpid/broker/Connection.h"
#include "qpid/framing/ConnectionCloseBody.h"
#include "qpid/framing/ConnectionCloseOkBody.h"
@@ -40,17 +41,10 @@ ConnectionCodec::Factory::create(Protoco
const std::string& id,
const qpid::sys::SecuritySettings& external)
{
- broker::Broker& broker = cluster.getBroker();
- if (broker.getConnectionCounter().allowConnection())
- {
- QPID_LOG(error, "Client max connection count limit exceeded: "
- << broker.getOptions().maxConnections << " connection refused");
- return 0;
- }
if (v == ProtocolVersion(0, 10))
return new ConnectionCodec(v, out, id, cluster, false, false, external);
else if (v == ProtocolVersion(0x80 + 0, 0x80 + 10)) // Catch-up connection
- return new ConnectionCodec(v, out, id, cluster, true, false, external);
+ return new ConnectionCodec(v, out, id, cluster, true, false, external);
return 0;
}
Modified: qpid/branches/asyncstore/cpp/src/qpid/cluster/Cpg.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/cluster/Cpg.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/cluster/Cpg.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/cluster/Cpg.cpp Fri Aug 3 12:13:32 2012
@@ -32,7 +32,7 @@
// This is a macro instead of a function because we don't want to
// evaluate the MSG argument unless there is an error.
#define CPG_CHECK(RESULT, MSG) \
- if ((RESULT) != CPG_OK) throw Exception(errorStr((RESULT), (MSG)))
+ if ((RESULT) != CS_OK) throw Exception(errorStr((RESULT), (MSG)))
namespace qpid {
namespace cluster {
@@ -50,13 +50,13 @@ Cpg* Cpg::cpgFromHandle(cpg_handle_t han
// Applies the same retry-logic to all cpg calls that need it.
void Cpg::callCpg ( CpgOp & c ) {
- cpg_error_t result;
+ cs_error_t result;
unsigned int snooze = 10;
for ( unsigned int nth_try = 0; nth_try < cpgRetries; ++ nth_try ) {
- if ( CPG_OK == (result = c.op(handle, & group))) {
+ if ( CS_OK == (result = c.op(handle, & group))) {
break;
}
- else if ( result == CPG_ERR_TRY_AGAIN ) {
+ else if ( result == CS_ERR_TRY_AGAIN ) {
QPID_LOG(info, "Retrying " << c.opName );
sys::usleep ( snooze );
snooze *= 10;
@@ -65,7 +65,7 @@ void Cpg::callCpg ( CpgOp & c ) {
else break; // Don't retry unless CPG tells us to.
}
- if ( result != CPG_OK )
+ if ( result != CS_OK )
CPG_CHECK(result, c.msg(group));
}
@@ -127,9 +127,9 @@ Cpg::Cpg(Handler& h) : IOHandle(new sys:
callbacks.cpg_confchg_fn = &globalConfigChange;
QPID_LOG(notice, "Initializing CPG");
- cpg_error_t err = cpg_initialize(&handle, &callbacks);
+ cs_error_t err = cpg_initialize(&handle, &callbacks);
int retries = 6; // FIXME aconway 2009-08-06: make this configurable.
- while (err == CPG_ERR_TRY_AGAIN && --retries) {
+ while (err == CS_ERR_TRY_AGAIN && --retries) {
QPID_LOG(notice, "Re-trying CPG initialization.");
sys::sleep(5);
err = cpg_initialize(&handle, &callbacks);
@@ -169,11 +169,11 @@ bool Cpg::mcast(const iovec* iov, int io
if (flowState == CPG_FLOW_CONTROL_ENABLED)
return false;
- cpg_error_t result;
+ cs_error_t result;
do {
result = cpg_mcast_joined(handle, CPG_TYPE_AGREED, const_cast<iovec*>(iov), iovLen);
- if (result != CPG_ERR_TRY_AGAIN) CPG_CHECK(result, cantMcastMsg(group));
- } while(result == CPG_ERR_TRY_AGAIN);
+ if (result != CS_ERR_TRY_AGAIN) CPG_CHECK(result, cantMcastMsg(group));
+ } while(result == CS_ERR_TRY_AGAIN);
return true;
}
@@ -187,34 +187,34 @@ void Cpg::shutdown() {
}
void Cpg::dispatchOne() {
- CPG_CHECK(cpg_dispatch(handle,CPG_DISPATCH_ONE), "Error in CPG dispatch");
+ CPG_CHECK(cpg_dispatch(handle,CS_DISPATCH_ONE), "Error in CPG dispatch");
}
void Cpg::dispatchAll() {
- CPG_CHECK(cpg_dispatch(handle,CPG_DISPATCH_ALL), "Error in CPG dispatch");
+ CPG_CHECK(cpg_dispatch(handle,CS_DISPATCH_ALL), "Error in CPG dispatch");
}
void Cpg::dispatchBlocking() {
- CPG_CHECK(cpg_dispatch(handle,CPG_DISPATCH_BLOCKING), "Error in CPG dispatch");
+ CPG_CHECK(cpg_dispatch(handle,CS_DISPATCH_BLOCKING), "Error in CPG dispatch");
}
-string Cpg::errorStr(cpg_error_t err, const std::string& msg) {
+string Cpg::errorStr(cs_error_t err, const std::string& msg) {
std::ostringstream os;
os << msg << ": ";
switch (err) {
- case CPG_OK: os << "ok"; break;
- case CPG_ERR_LIBRARY: os << "library"; break;
- case CPG_ERR_TIMEOUT: os << "timeout"; break;
- case CPG_ERR_TRY_AGAIN: os << "try again"; break;
- case CPG_ERR_INVALID_PARAM: os << "invalid param"; break;
- case CPG_ERR_NO_MEMORY: os << "no memory"; break;
- case CPG_ERR_BAD_HANDLE: os << "bad handle"; break;
- case CPG_ERR_ACCESS: os << "access denied. You may need to set your group ID to 'ais'"; break;
- case CPG_ERR_NOT_EXIST: os << "not exist"; break;
- case CPG_ERR_EXIST: os << "exist"; break;
- case CPG_ERR_NOT_SUPPORTED: os << "not supported"; break;
- case CPG_ERR_SECURITY: os << "security"; break;
- case CPG_ERR_TOO_MANY_GROUPS: os << "too many groups"; break;
+ case CS_OK: os << "ok"; break;
+ case CS_ERR_LIBRARY: os << "library"; break;
+ case CS_ERR_TIMEOUT: os << "timeout"; break;
+ case CS_ERR_TRY_AGAIN: os << "try again"; break;
+ case CS_ERR_INVALID_PARAM: os << "invalid param"; break;
+ case CS_ERR_NO_MEMORY: os << "no memory"; break;
+ case CS_ERR_BAD_HANDLE: os << "bad handle"; break;
+ case CS_ERR_ACCESS: os << "access denied. You may need to set your group ID to 'ais'"; break;
+ case CS_ERR_NOT_EXIST: os << "not exist"; break;
+ case CS_ERR_EXIST: os << "exist"; break;
+ case CS_ERR_NOT_SUPPORTED: os << "not supported"; break;
+ case CS_ERR_SECURITY: os << "security"; break;
+ case CS_ERR_TOO_MANY_GROUPS: os << "too many groups"; break;
default: os << ": unknown cpg error " << err;
};
os << " (" << err << ")";
Modified: qpid/branches/asyncstore/cpp/src/qpid/cluster/Cpg.h
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/cluster/Cpg.h?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/cluster/Cpg.h (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/cluster/Cpg.h Fri Aug 3 12:13:32 2012
@@ -131,7 +131,7 @@ class Cpg : public sys::IOHandle {
CpgOp ( std::string opName )
: opName(opName) { }
- virtual cpg_error_t op ( cpg_handle_t handle, struct cpg_name * ) = 0;
+ virtual cs_error_t op ( cpg_handle_t handle, struct cpg_name * ) = 0;
virtual std::string msg(const Name&) = 0;
virtual ~CpgOp ( ) { }
};
@@ -141,7 +141,7 @@ class Cpg : public sys::IOHandle {
CpgJoinOp ( )
: CpgOp ( std::string("cpg_join") ) { }
- cpg_error_t op(cpg_handle_t handle, struct cpg_name * group) {
+ cs_error_t op(cpg_handle_t handle, struct cpg_name * group) {
return cpg_join ( handle, group );
}
@@ -152,7 +152,7 @@ class Cpg : public sys::IOHandle {
CpgLeaveOp ( )
: CpgOp ( std::string("cpg_leave") ) { }
- cpg_error_t op(cpg_handle_t handle, struct cpg_name * group) {
+ cs_error_t op(cpg_handle_t handle, struct cpg_name * group) {
return cpg_leave ( handle, group );
}
@@ -163,7 +163,7 @@ class Cpg : public sys::IOHandle {
CpgFinalizeOp ( )
: CpgOp ( std::string("cpg_finalize") ) { }
- cpg_error_t op(cpg_handle_t handle, struct cpg_name *) {
+ cs_error_t op(cpg_handle_t handle, struct cpg_name *) {
return cpg_finalize ( handle );
}
@@ -177,7 +177,7 @@ class Cpg : public sys::IOHandle {
CpgLeaveOp cpgLeaveOp;
CpgFinalizeOp cpgFinalizeOp;
- static std::string errorStr(cpg_error_t err, const std::string& msg);
+ static std::string errorStr(cs_error_t err, const std::string& msg);
static std::string cantJoinMsg(const Name&);
static std::string cantLeaveMsg(const Name&);
static std::string cantMcastMsg(const Name&);
Modified: qpid/branches/asyncstore/cpp/src/qpid/cluster/UpdateClient.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/cluster/UpdateClient.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/cluster/UpdateClient.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/cluster/UpdateClient.cpp Fri Aug 3 12:13:32 2012
@@ -74,6 +74,8 @@
namespace qpid {
namespace cluster {
+using std::string;
+
using amqp_0_10::ListCodec;
using broker::Broker;
using broker::Exchange;
@@ -87,6 +89,8 @@ using namespace framing;
namespace arg=client::arg;
using client::SessionBase_0_10Access;
+namespace _qmf = qmf::org::apache::qpid::broker;
+
// Reserved exchange/queue name for catch-up, avoid clashes with user queues/exchanges.
const std::string UpdateClient::UPDATE("x-qpid.cluster-update");
// Name for header used to carry expiration information.
@@ -226,14 +230,6 @@ template <class T> std::string encode(co
t.encode(buf);
return encoded;
}
-
-template <class T> std::string encode(const T& t, bool encodeKind) {
- std::string encoded;
- encoded.resize(t.encodedSize());
- framing::Buffer buf(const_cast<char*>(encoded.data()), encoded.size());
- t.encode(buf, encodeKind);
- return encoded;
-}
} // namespace
@@ -377,13 +373,14 @@ class MessageUpdater {
void UpdateClient::updateQueue(client::AsyncSession& s, const boost::shared_ptr<Queue>& q) {
broker::Exchange::shared_ptr alternateExchange = q->getAlternateExchange();
+ _qmf::Queue* mgmtQueue = dynamic_cast<_qmf::Queue*>(q->GetManagementObject());
s.queueDeclare(
arg::queue = q->getName(),
arg::durable = q->isDurable(),
arg::autoDelete = q->isAutoDelete(),
arg::alternateExchange = alternateExchange ? alternateExchange->getName() : "",
arg::arguments = q->getSettings(),
- arg::exclusive = q->hasExclusiveOwner()
+ arg::exclusive = mgmtQueue && mgmtQueue->get_exclusive()
);
MessageUpdater updater(q->getName(), s, expiry);
q->eachMessage(boost::bind(&MessageUpdater::updateQueuedMessage, &updater, _1));
@@ -545,7 +542,8 @@ void UpdateClient::updateConsumer(
ci->isNotifyEnabled(),
ci->getPosition(),
ci->getCredit().used().messages,
- ci->getCredit().used().bytes
+ ci->getCredit().used().bytes,
+ ci->getDeliveryCount()
);
consumerNumbering.add(ci.get());
Propchange: qpid/branches/asyncstore/cpp/src/qpid/cluster/UpdateClient.cpp
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.cpp:r1333988-1368650
Propchange: qpid/branches/asyncstore/cpp/src/qpid/cluster/UpdateClient.h
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.h:r1333988-1368650
Modified: qpid/branches/asyncstore/cpp/src/qpid/cluster/types.h
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/cluster/types.h?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/cluster/types.h (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/cluster/types.h Fri Aug 3 12:13:32 2012
@@ -34,6 +34,29 @@
extern "C" {
#if defined (HAVE_OPENAIS_CPG_H)
# include <openais/cpg.h>
+
+// Provide translations back to the deprecated definitions in openais
+typedef cpg_error_t cs_error_t;
+#define CS_DISPATCH_ONE CPG_DISPATCH_ONE
+#define CS_DISPATCH_ALL CPG_DISPATCH_ALL
+#define CS_DISPATCH_BLOCKING CPG_DISPATCH_BLOCKING
+#define CS_FLOW_CONTROL_DISABLED CPG_FLOW_CONTROL_DISABLED
+#define CS_FLOW_CONTROL_ENABLED CPG_FLOW_CONTROL_ENABLED
+#define CS_OK CPG_OK
+#define CS_ERR_LIBRARY CPG_ERR_LIBRARY
+#define CS_ERR_TIMEOUT CPG_ERR_TIMEOUT
+#define CS_ERR_TRY_AGAIN CPG_ERR_TRY_AGAIN
+#define CS_ERR_INVALID_PARAM CPG_ERR_INVALID_PARAM
+#define CS_ERR_NO_MEMORY CPG_ERR_NO_MEMORY
+#define CS_ERR_BAD_HANDLE CPG_ERR_BAD_HANDLE
+#define CS_ERR_BUSY CPG_ERR_BUSY
+#define CS_ERR_ACCESS CPG_ERR_ACCESS
+#define CS_ERR_NOT_EXIST CPG_ERR_NOT_EXIST
+#define CS_ERR_EXIST CPG_ERR_EXIST
+#define CS_ERR_NOT_SUPPORTED CPG_ERR_NOT_SUPPORTED
+#define CS_ERR_SECURITY CPG_ERR_SECURITY
+#define CS_ERR_TOO_MANY_GROUPS CPG_ERR_TOO_MANY_GROUPS
+
#elif defined (HAVE_COROSYNC_CPG_H)
# include <corosync/cpg.h>
#else
Modified: qpid/branches/asyncstore/cpp/src/qpid/console/ClassKey.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/console/ClassKey.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/console/ClassKey.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/console/ClassKey.cpp Fri Aug 3 12:13:32 2012
@@ -21,6 +21,7 @@
#include "qpid/console/ClassKey.h"
#include <string.h>
+#include <iostream>
#include <cstdio>
using namespace std;
Modified: qpid/branches/asyncstore/cpp/src/qpid/framing/AMQContentBody.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/framing/AMQContentBody.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/framing/AMQContentBody.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/framing/AMQContentBody.cpp Fri Aug 3 12:13:32 2012
@@ -24,7 +24,7 @@
qpid::framing::AMQContentBody::AMQContentBody(){
}
-qpid::framing::AMQContentBody::AMQContentBody(const string& _data) : data(_data){
+qpid::framing::AMQContentBody::AMQContentBody(const std::string& _data) : data(_data){
}
uint32_t qpid::framing::AMQContentBody::encodedSize() const{
Modified: qpid/branches/asyncstore/cpp/src/qpid/framing/AMQContentBody.h
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/framing/AMQContentBody.h?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/framing/AMQContentBody.h (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/framing/AMQContentBody.h Fri Aug 3 12:13:32 2012
@@ -31,15 +31,15 @@ namespace framing {
class QPID_COMMON_CLASS_EXTERN AMQContentBody : public AMQBody
{
- string data;
+ std::string data;
public:
QPID_COMMON_EXTERN AMQContentBody();
- QPID_COMMON_EXTERN AMQContentBody(const string& data);
+ QPID_COMMON_EXTERN AMQContentBody(const std::string& data);
inline virtual ~AMQContentBody(){}
inline uint8_t type() const { return CONTENT_BODY; };
- inline const string& getData() const { return data; }
- inline string& getData() { return data; }
+ inline const std::string& getData() const { return data; }
+ inline std::string& getData() { return data; }
QPID_COMMON_EXTERN uint32_t encodedSize() const;
QPID_COMMON_EXTERN void encode(Buffer& buffer) const;
QPID_COMMON_EXTERN void decode(Buffer& buffer, uint32_t size);
Modified: qpid/branches/asyncstore/cpp/src/qpid/framing/AMQFrame.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/framing/AMQFrame.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/framing/AMQFrame.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/framing/AMQFrame.cpp Fri Aug 3 12:13:32 2012
@@ -85,7 +85,7 @@ bool AMQFrame::decode(Buffer& buffer)
{
if(buffer.available() < frameOverhead())
return false;
- buffer.record();
+ uint32_t pos = buffer.getPosition();
uint8_t flags = buffer.getOctet();
uint8_t framing_version = (flags & 0xc0) >> 6;
@@ -115,7 +115,7 @@ bool AMQFrame::decode(Buffer& buffer)
// B,E,b,e flags
uint16_t body_size = frame_size - frameOverhead();
if (buffer.available() < body_size){
- buffer.restore();
+ buffer.setPosition(pos);
return false;
}
Modified: qpid/branches/asyncstore/cpp/src/qpid/framing/Buffer.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/framing/Buffer.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/framing/Buffer.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/framing/Buffer.cpp Fri Aug 3 12:13:32 2012
@@ -23,27 +23,17 @@
#include "qpid/Msg.h"
#include <string.h>
#include <boost/format.hpp>
+
namespace qpid {
namespace framing {
+using std::string;
+
Buffer::Buffer(char* _data, uint32_t _size)
: size(_size), data(_data), position(0) {
}
-void Buffer::record(){
- r_position = position;
-}
-
-void Buffer::restore(bool reRecord){
- uint32_t savedPosition = position;
-
- position = r_position;
-
- if (reRecord)
- r_position = savedPosition;
-}
-
void Buffer::reset(){
position = 0;
}
Modified: qpid/branches/asyncstore/cpp/src/qpid/framing/FieldTable.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/framing/FieldTable.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/framing/FieldTable.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/framing/FieldTable.cpp Fri Aug 3 12:13:32 2012
@@ -196,7 +196,7 @@ void FieldTable::setFloat(const std::str
flushRawCache();
}
-void FieldTable::setDouble(const std::string& name, double value){
+void FieldTable::setDouble(const std::string& name, const double value){
realDecode();
values[name] = ValuePtr(new DoubleValue(value));
flushRawCache();
Modified: qpid/branches/asyncstore/cpp/src/qpid/framing/ProtocolInitiation.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/framing/ProtocolInitiation.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/framing/ProtocolInitiation.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/framing/ProtocolInitiation.cpp Fri Aug 3 12:13:32 2012
@@ -20,6 +20,8 @@
*/
#include "qpid/framing/ProtocolInitiation.h"
+#include <iostream>
+
namespace qpid {
namespace framing {
Modified: qpid/branches/asyncstore/cpp/src/qpid/framing/Uuid.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/framing/Uuid.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/framing/Uuid.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/framing/Uuid.cpp Fri Aug 3 12:13:32 2012
@@ -43,6 +43,13 @@ Uuid::Uuid(const uint8_t* data) {
assign(data);
}
+Uuid::Uuid(const std::string& s) {
+ if (s.size() != UNPARSED_SIZE)
+ throw IllegalArgumentException(QPID_MSG("Invalid UUID: " << s));
+ if (uuid_parse(const_cast<char *>(&s[0]), c_array()) != 0)
+ throw IllegalArgumentException(QPID_MSG("Invalid UUID: " << s));
+}
+
void Uuid::assign(const uint8_t* data) {
// This const cast is for Solaris which has a
// uuid_copy that takes a non const 2nd argument
Modified: qpid/branches/asyncstore/cpp/src/qpid/ha/Backup.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/ha/Backup.cpp?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/ha/Backup.cpp (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/ha/Backup.cpp Fri Aug 3 12:13:32 2012
@@ -20,7 +20,6 @@
*/
#include "Backup.h"
#include "BrokerReplicator.h"
-#include "ConnectionExcluder.h"
#include "HaBroker.h"
#include "ReplicatingSubscription.h"
#include "Settings.h"
@@ -34,6 +33,7 @@
#include "qpid/framing/AMQFrame.h"
#include "qpid/framing/FieldTable.h"
#include "qpid/framing/MessageTransferBody.h"
+#include "qpid/sys/SystemInfo.h"
#include "qpid/types/Variant.h"
namespace qpid {
@@ -45,47 +45,85 @@ using types::Variant;
using std::string;
Backup::Backup(HaBroker& hb, const Settings& s) :
- haBroker(hb), broker(hb.getBroker()), settings(s), excluder(new ConnectionExcluder())
+ logPrefix("Backup: "), haBroker(hb), broker(hb.getBroker()), settings(s)
{
- // Exclude client connections before starting the link to avoid self-connection.
- broker.getConnectionObservers().add(excluder);
- // Empty brokerUrl means delay initialization until setUrl() is called.
+ // Empty brokerUrl means delay initialization until seBrokertUrl() is called.
if (!s.brokerUrl.empty()) initialize(Url(s.brokerUrl));
}
-void Backup::initialize(const Url& url) {
- if (url.empty()) throw Url::Invalid("HA broker URL is empty");
- QPID_LOG(notice, "HA: Backup initialized: " << url);
+bool Backup::isSelf(const Address& a) const {
+ return sys::SystemInfo::isLocalHost(a.host) &&
+ a.port == haBroker.getBroker().getPort(a.protocol);
+}
+
+// Remove my own address from the URL if possible.
+// This isn't 100% reliable given the many ways to specify a host,
+// but should work in most cases. We have additional measures to prevent
+// self-connection in ConnectionObserver
+Url Backup::removeSelf(const Url& brokers) const {
+ Url url;
+ for (Url::const_iterator i = brokers.begin(); i != brokers.end(); ++i)
+ if (!isSelf(*i)) url.push_back(*i);
+ if (url.empty())
+ throw Url::Invalid(logPrefix+"Failover URL is empty");
+ QPID_LOG(debug, logPrefix << "Failover URL (excluding self): " << url);
+ return url;
+}
+
+void Backup::initialize(const Url& brokers) {
+ if (brokers.empty()) throw Url::Invalid("HA broker URL is empty");
+ QPID_LOG(info, logPrefix << "Connecting to cluster, broker URL: " << brokers);
+ Url url = removeSelf(brokers);
string protocol = url[0].protocol.empty() ? "tcp" : url[0].protocol;
+ types::Uuid uuid(true);
// Declare the link
std::pair<Link::shared_ptr, bool> result = broker.getLinks().declare(
+ broker::QPID_NAME_PREFIX + string("ha.link.") + uuid.str(),
url[0].host, url[0].port, protocol,
- false, // durable
- settings.mechanism, settings.username, settings.password);
- link = result.first;
- link->setUrl(url);
- replicator.reset(new BrokerReplicator(haBroker, link));
- broker.getExchanges().registerExchange(replicator);
+ false, // durable
+ settings.mechanism, settings.username, settings.password,
+ false); // no amq.failover - don't want to use client URL.
+ {
+ sys::Mutex::ScopedLock l(lock);
+ link = result.first;
+ replicator.reset(new BrokerReplicator(haBroker, link));
+ replicator->initialize();
+ broker.getExchanges().registerExchange(replicator);
+ }
+ link->setUrl(url); // Outside the lock, once set link doesn't change.
}
Backup::~Backup() {
if (link) link->close();
if (replicator.get()) broker.getExchanges().destroy(replicator->getName());
- replicator.reset();
- broker.getConnectionObservers().remove(excluder); // This allows client connections.
}
-
+// Called via management.
void Backup::setBrokerUrl(const Url& url) {
// Ignore empty URLs seen during start-up for some tests.
if (url.empty()) return;
- sys::Mutex::ScopedLock l(lock);
- if (link) { // URL changed after we initialized.
- QPID_LOG(info, "HA: Backup broker URL set to " << url);
- link->setUrl(url);
+ bool linkSet = false;
+ {
+ sys::Mutex::ScopedLock l(lock);
+ linkSet = link;
+ }
+ if (linkSet) {
+ QPID_LOG(info, logPrefix << "Broker URL set to: " << url);
+ link->setUrl(removeSelf(url)); // Outside lock, once set link doesn't change
}
- else {
+ else
initialize(url); // Deferred initialization
+}
+
+void Backup::setStatus(BrokerStatus status) {
+ switch (status) {
+ case READY:
+ QPID_LOG(notice, logPrefix << "Ready to become primary.");
+ break;
+ case CATCHUP:
+ QPID_LOG(notice, logPrefix << "Catching up on primary, cannot be promoted.");
+ default:
+ assert(0);
}
}
Modified: qpid/branches/asyncstore/cpp/src/qpid/ha/Backup.h
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/ha/Backup.h?rev=1368910&r1=1368909&r2=1368910&view=diff
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/ha/Backup.h (original)
+++ qpid/branches/asyncstore/cpp/src/qpid/ha/Backup.h Fri Aug 3 12:13:32 2012
@@ -36,7 +36,6 @@ class Link;
namespace ha {
class Settings;
-class ConnectionExcluder;
class BrokerReplicator;
class HaBroker;
@@ -51,9 +50,13 @@ class Backup
Backup(HaBroker&, const Settings&);
~Backup();
void setBrokerUrl(const Url&);
+ void setStatus(BrokerStatus);
private:
+ bool isSelf(const Address& a) const;
+ Url removeSelf(const Url&) const;
void initialize(const Url&);
+ std::string logPrefix;
sys::Mutex lock;
HaBroker& haBroker;
@@ -61,7 +64,6 @@ class Backup
Settings settings;
boost::shared_ptr<broker::Link> link;
boost::shared_ptr<BrokerReplicator> replicator;
- boost::shared_ptr<ConnectionExcluder> excluder;
};
}} // namespace qpid::ha
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org