You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ro...@apache.org on 2012/11/19 13:48:08 UTC
svn commit: r1411155 [3/6] - in
/qpid/branches/java-broker-config-qpid-4390/qpid: ./ cpp/ cpp/bindings/
cpp/bindings/qpid/dotnet/
cpp/bindings/qpid/dotnet/examples/csharp.map.callback.receiver/
cpp/bindings/qpid/dotnet/msvc9/ cpp/bindings/qpid/dotnet/s...
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/BrokerReplicator.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/BrokerReplicator.cpp?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/BrokerReplicator.cpp (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/BrokerReplicator.cpp Mon Nov 19 12:47:53 2012
@@ -29,6 +29,7 @@
#include "qpid/broker/Link.h"
#include "qpid/broker/amqp_0_10/MessageTransfer.h"
#include "qpid/framing/FieldTable.h"
+#include "qpid/framing/FieldValue.h"
#include "qpid/log/Statement.h"
#include "qpid/amqp_0_10/Codecs.h"
#include "qpid/broker/SessionHandler.h"
@@ -76,8 +77,9 @@ const string EVENT("_event");
const string OBJECT_NAME("_object_name");
const string PACKAGE_NAME("_package_name");
const string QUERY_RESPONSE("_query_response");
-const string SCHEMA_ID("_schema_id");
const string VALUES("_values");
+const string SCHEMA_ID("_schema_id");
+const string WHAT("_what");
const string ALTEX("altEx");
const string ALTEXCHANGE("altExchange");
@@ -114,10 +116,6 @@ const string QMF_CONTENT("qmf.content");
const string QMF_DEFAULT_TOPIC("qmf.default.topic");
const string QMF_OPCODE("qmf.opcode");
-const string _WHAT("_what");
-const string _CLASS_NAME("_class_name");
-const string _PACKAGE_NAME("_package_name");
-const string _SCHEMA_ID("_schema_id");
const string OBJECT("OBJECT");
const string ORG_APACHE_QPID_BROKER("org.apache.qpid.broker");
const string ORG_APACHE_QPID_HA("org.apache.qpid.ha");
@@ -132,11 +130,11 @@ void sendQuery(const string& packageName
{
framing::AMQP_ServerProxy peer(sessionHandler.out);
Variant::Map request;
- request[_WHAT] = OBJECT;
+ request[WHAT] = OBJECT;
Variant::Map schema;
- schema[_CLASS_NAME] = className;
- schema[_PACKAGE_NAME] = packageName;
- request[_SCHEMA_ID] = schema;
+ schema[CLASS_NAME] = className;
+ schema[PACKAGE_NAME] = packageName;
+ request[SCHEMA_ID] = schema;
AMQFrame method((MessageTransferBody(ProtocolVersion(), QMF_DEFAULT_DIRECT, 0, 0)));
method.setBof(true);
@@ -214,31 +212,87 @@ class BrokerReplicator::ConnectionObserv
BrokerReplicator& brokerReplicator;
};
-template <class E> pair<string,string> eventKey() {
- return make_pair(E::PACKAGE_NAME, E::EVENT_NAME);
-}
+/** Keep track of queues or exchanges during the update process to solve 2
+ * problems.
+ *
+ * 1. Once all responses are processed, remove any queues/exchanges
+ * that were not mentioned as they no longer exist on the primary.
+ *
+ * 2. During the update if we see an event for an object we should
+ * ignore any subsequent responses for that object as they are out
+ * of date.
+ */
+class BrokerReplicator::UpdateTracker {
+ public:
+ typedef std::set<std::string> Names;
+ typedef boost::function<void (const std::string&)> CleanFn;
+
+ UpdateTracker(CleanFn f, const ReplicationTest& rt) : cleanFn(f), repTest(rt) {}
+
+ /** Destructor cleans up remaining initial queues. */
+ ~UpdateTracker() {
+ // Don't throw in a destructor.
+ try { for_each(initial.begin(), initial.end(), cleanFn); }
+ catch (const std::exception& e) {
+ QPID_LOG(error, "Error in cleanup of lost objects: " << e.what());
+ }
+ }
+
+ /** Add an exchange name */
+ void addExchange(Exchange::shared_ptr ex) {
+ if (repTest.isReplicated(CONFIGURATION, *ex)) initial.insert(ex->getName());
+ }
+
+ /** Add a queue name. */
+ void addQueue(Queue::shared_ptr q) {
+ if (repTest.isReplicated(CONFIGURATION, *q)) initial.insert(q->getName());
+ }
+
+ /** Received an event for name */
+ void event(const std::string& name) {
+ initial.erase(name); // no longer a candidate for deleting
+ events.insert(name); // we have seen an event for this name
+ }
+
+ /** Received a response for name.
+ *@return true if this response should be processed, false if we have
+ *already seen an event for this object.
+ */
+ bool response(const std::string& name) {
+ initial.erase(name); // no longer a candidate for deleting
+ return events.find(name) == events.end(); // true if no event seen yet.
+ }
+
+ private:
+ Names initial, events;
+ CleanFn cleanFn;
+ ReplicationTest repTest;
+};
BrokerReplicator::BrokerReplicator(HaBroker& hb, const boost::shared_ptr<Link>& l)
: Exchange(QPID_CONFIGURATION_REPLICATOR),
logPrefix("Backup: "), replicationTest(hb.getReplicationTest()),
- haBroker(hb), broker(hb.getBroker()), link(l),
+ haBroker(hb), broker(hb.getBroker()),
+ exchanges(broker.getExchanges()), queues(broker.getQueues()),
+ link(l),
initialized(false),
alternates(hb.getBroker().getExchanges()),
- cleaner(*this),
connection(0)
{
broker.getConnectionObservers().add(
boost::shared_ptr<broker::ConnectionObserver>(new ConnectionObserver(*this)));
- getArgs().setString(QPID_REPLICATE, printable(NONE).str());
-
- dispatch[eventKey<EventQueueDeclare>()] = &BrokerReplicator::doEventQueueDeclare;
- dispatch[eventKey<EventQueueDelete>()] = &BrokerReplicator::doEventQueueDelete;
- dispatch[eventKey<EventExchangeDeclare>()] = &BrokerReplicator::doEventExchangeDeclare;
- dispatch[eventKey<EventExchangeDelete>()] = &BrokerReplicator::doEventExchangeDelete;
- dispatch[eventKey<EventBind>()] = &BrokerReplicator::doEventBind;
- dispatch[eventKey<EventUnbind>()] = &BrokerReplicator::doEventUnbind;
- dispatch[eventKey<EventMembersUpdate>()] = &BrokerReplicator::doEventMembersUpdate;
- dispatch[eventKey<EventSubscribe>()] = &BrokerReplicator::doEventSubscribe;
+ framing::FieldTable args = getArgs();
+ args.setString(QPID_REPLICATE, printable(NONE).str());
+ setArgs(args);
+
+ dispatch[EventQueueDeclare::getFullName()] = &BrokerReplicator::doEventQueueDeclare;
+ dispatch[EventQueueDelete::getFullName()] = &BrokerReplicator::doEventQueueDelete;
+ dispatch[EventExchangeDeclare::getFullName()] = &BrokerReplicator::doEventExchangeDeclare;
+ dispatch[EventExchangeDelete::getFullName()] = &BrokerReplicator::doEventExchangeDelete;
+ dispatch[EventBind::getFullName()] = &BrokerReplicator::doEventBind;
+ dispatch[EventUnbind::getFullName()] = &BrokerReplicator::doEventUnbind;
+ dispatch[EventMembersUpdate::getFullName()] = &BrokerReplicator::doEventMembersUpdate;
+ dispatch[EventSubscribe::getFullName()] = &BrokerReplicator::doEventSubscribe;
}
void BrokerReplicator::initialize() {
@@ -298,9 +352,16 @@ void BrokerReplicator::initializeBridge(
<< " status:" << printable(haBroker.getStatus()));
initialized = true;
- // Scan for existing replicated queues and exchanges. Any that have not been seen by
- // the time all reponses are received will be cleaned up.
- cleaner.start();
+ exchangeTracker.reset(
+ new UpdateTracker(boost::bind(&BrokerReplicator::deleteExchange, this, _1),
+ replicationTest));
+ exchanges.eachExchange(
+ boost::bind(&UpdateTracker::addExchange, exchangeTracker.get(), _1));
+
+ queueTracker.reset(
+ new UpdateTracker(boost::bind(&BrokerReplicator::deleteQueue, this, _1, true),
+ replicationTest));
+ queues.eachQueue(boost::bind(&UpdateTracker::addQueue, queueTracker.get(), _1));
framing::AMQP_ServerProxy peer(sessionHandler.out);
const qmf::org::apache::qpid::broker::ArgsLinkBridge& args(bridge.getArgs());
@@ -369,12 +430,12 @@ void BrokerReplicator::route(Deliverable
}
if (MessageTransfer::isLastQMFResponse(msg.getMessage(), EXCHANGE)) {
QPID_LOG(debug, logPrefix << "All exchange responses received.")
- cleaner.cleanExchanges(); // Clean up exchanges that no longer exist in the primary
+ exchangeTracker.reset(); // Clean up exchanges that no longer exist in the primary
alternates.clear();
}
if (MessageTransfer::isLastQMFResponse(msg.getMessage(), QUEUE)) {
- QPID_LOG(debug, logPrefix << "All queue responses received.")
- cleaner.cleanQueues(); // Clean up queues that no longer exist in the primary
+ QPID_LOG(debug, logPrefix << "All queue responses received.");
+ queueTracker.reset(); // Clean up queues that no longer exist in the primary
}
}
} catch (const std::exception& e) {
@@ -394,13 +455,14 @@ void BrokerReplicator::doEventQueueDecla
string name = values[QNAME].asString();
QueueSettings settings(values[DURABLE].asBool(), values[AUTODEL].asBool());
QPID_LOG(debug, logPrefix << "Queue declare event: " << name);
- cleaner.forgetQueue(name);
+ if (queueTracker.get()) queueTracker->event(name);
framing::FieldTable args;
qpid::amqp_0_10::translate(argsMap, args);
// If we already have a queue with this name, replace it.
// The queue was definitely created on the primary.
- if (broker.getQueues().find(name)) {
- QPID_LOG(warning, logPrefix << "Replacing exsiting queue: " << name);
+ if (queues.find(name)) {
+ QPID_LOG(warning, logPrefix << "Declare event, replacing exsiting queue: "
+ << name);
deleteQueue(name);
}
replicateQueue(name, values[DURABLE].asBool(), autoDel, args,
@@ -412,7 +474,7 @@ boost::shared_ptr<QueueReplicator> Broke
const std::string& qname)
{
string rname = QueueReplicator::replicatorName(qname);
- boost::shared_ptr<broker::Exchange> ex = broker.getExchanges().find(rname);
+ boost::shared_ptr<broker::Exchange> ex = exchanges.find(rname);
return boost::dynamic_pointer_cast<QueueReplicator>(ex);
}
@@ -420,10 +482,10 @@ void BrokerReplicator::doEventQueueDelet
// The remote queue has already been deleted so replicator
// sessions may be closed by a "queue deleted" exception.
string name = values[QNAME].asString();
- boost::shared_ptr<Queue> queue = broker.getQueues().find(name);
+ boost::shared_ptr<Queue> queue = queues.find(name);
if (queue && replicationTest.replicateLevel(queue->getSettings().storeSettings)) {
QPID_LOG(debug, logPrefix << "Queue delete event: " << name);
- cleaner.forgetQueue(name);
+ if (queueTracker.get()) queueTracker->event(name);
deleteQueue(name);
}
}
@@ -434,14 +496,15 @@ void BrokerReplicator::doEventExchangeDe
if (values[DISP] == CREATED && replicationTest.replicateLevel(argsMap)) {
string name = values[EXNAME].asString();
QPID_LOG(debug, logPrefix << "Exchange declare event: " << name);
- cleaner.forgetExchange(name);
+ if (exchangeTracker.get()) exchangeTracker->event(name);
framing::FieldTable args;
qpid::amqp_0_10::translate(argsMap, args);
// If we already have a exchange with this name, replace it.
// The exchange was definitely created on the primary.
- if (broker.getExchanges().find(name)) {
+ if (exchanges.find(name)) {
deleteExchange(name);
- QPID_LOG(warning, logPrefix << "Replaced existing exchange: " << name);
+ QPID_LOG(warning, logPrefix << "Declare event, replacing existing exchange: "
+ << name);
}
CreateExchangeResult result = createExchange(
name, values[EXTYPE].asString(), values[DURABLE].asBool(), args,
@@ -453,14 +516,14 @@ void BrokerReplicator::doEventExchangeDe
void BrokerReplicator::doEventExchangeDelete(Variant::Map& values) {
string name = values[EXNAME].asString();
- boost::shared_ptr<Exchange> exchange = broker.getExchanges().find(name);
+ boost::shared_ptr<Exchange> exchange = exchanges.find(name);
if (!exchange) {
QPID_LOG(warning, logPrefix << "Exchange delete event, not found: " << name);
} else if (!replicationTest.replicateLevel(exchange->getArgs())) {
QPID_LOG(warning, logPrefix << "Exchange delete event, not replicated: " << name);
} else {
QPID_LOG(debug, logPrefix << "Exchange delete event:" << name);
- cleaner.forgetExchange(name);
+ if (exchangeTracker.get()) exchangeTracker->event(name);
deleteExchange(name);
replicatedExchanges.erase(name);
}
@@ -468,9 +531,9 @@ void BrokerReplicator::doEventExchangeDe
void BrokerReplicator::doEventBind(Variant::Map& values) {
boost::shared_ptr<Exchange> exchange =
- broker.getExchanges().find(values[EXNAME].asString());
+ exchanges.find(values[EXNAME].asString());
boost::shared_ptr<Queue> queue =
- broker.getQueues().find(values[QNAME].asString());
+ queues.find(values[QNAME].asString());
// We only replicate binds for a replicated queue to replicated
// exchange that both exist locally.
if (exchange && replicationTest.replicateLevel(exchange->getArgs()) &&
@@ -488,9 +551,9 @@ void BrokerReplicator::doEventBind(Varia
void BrokerReplicator::doEventUnbind(Variant::Map& values) {
boost::shared_ptr<Exchange> exchange =
- broker.getExchanges().find(values[EXNAME].asString());
+ exchanges.find(values[EXNAME].asString());
boost::shared_ptr<Queue> queue =
- broker.getQueues().find(values[QNAME].asString());
+ queues.find(values[QNAME].asString());
// We only replicate unbinds for a replicated queue to replicated
// exchange that both exist locally.
if (exchange && replicationTest.replicateLevel(exchange->getArgs()) &&
@@ -533,8 +596,15 @@ string getAltExchange(const types::Varia
}
else return string();
}
+
+Variant getHaUuid(const Variant::Map& map) {
+ Variant::Map::const_iterator i = map.find(QPID_HA_UUID);
+ return i == map.end() ? Variant() : i->second;
}
+} // namespace
+
+
void BrokerReplicator::doResponseQueue(Variant::Map& values) {
Variant::Map argsMap(asMapVoid(values[ARGUMENTS]));
if (!replicationTest.isReplicated(
@@ -544,15 +614,17 @@ void BrokerReplicator::doResponseQueue(V
values[EXCLUSIVE].asBool()))
return;
string name(values[NAME].asString());
- cleaner.forgetQueue(name);
+ if (!queueTracker.get())
+ throw Exception(QPID_MSG("Unexpected queue response: " << values));
+ if (!queueTracker->response(name)) return; // Response is out-of-date
QPID_LOG(debug, logPrefix << "Queue response: " << name);
- if (broker.getQueues().find(name)) { // Already exists
- if (findQueueReplicator(name))
- return; // Already replicated
- else {
- QPID_LOG(debug, logPrefix << "Deleting queue to make way for replica: " << name);
- broker.deleteQueue(name, userId, remoteHost);
- }
+ // If we see a queue with the same name as one we have, but not the same UUID,
+ // then replace the one we have.
+ boost::shared_ptr<Queue> queue = queues.find(name);
+ if (queue && getHaUuid(queue->getSettings().original) != getHaUuid(argsMap)) {
+ QPID_LOG(warning, logPrefix << "UUID mismatch, replacing queue: "
+ << name);
+ deleteQueue(name);
}
framing::FieldTable args;
qpid::amqp_0_10::translate(argsMap, args);
@@ -571,23 +643,25 @@ void BrokerReplicator::doResponseExchang
Variant::Map argsMap(asMapVoid(values[ARGUMENTS]));
if (!replicationTest.replicateLevel(argsMap)) return;
string name = values[NAME].asString();
- cleaner.forgetExchange(name);
+ if (!exchangeTracker.get())
+ throw Exception(QPID_MSG("Unexpected exchange response: " << values));
+ if (!exchangeTracker->response(name)) return; // Response is out of date.
QPID_LOG(debug, logPrefix << "Exchange response: " << name);
- if (broker.getExchanges().find(name)) {
- if (replicatedExchanges.find(name) != replicatedExchanges.end())
- return; // Already replicated
- else {
- QPID_LOG(debug, logPrefix << "Deleting exchange to make way for replica: "
- << name);
- broker.deleteExchange(name, userId, remoteHost);
- }
- }
framing::FieldTable args;
qpid::amqp_0_10::translate(argsMap, args);
+ // If we see an exchange with the same name as one we have, but not the same UUID,
+ // then replace the one we have.
+ boost::shared_ptr<Exchange> exchange = exchanges.find(name);
+ if (exchange &&
+ exchange->getArgs().getAsString(QPID_HA_UUID) != args.getAsString(QPID_HA_UUID))
+ {
+ QPID_LOG(warning, logPrefix << "UUID mismatch, replacing exchange: "
+ << name);
+ deleteExchange(name);
+ }
CreateExchangeResult result = createExchange(
name, values[TYPE].asString(), values[DURABLE].asBool(), args,
getAltExchange(values[ALTEXCHANGE]));
- assert(result.second);
replicatedExchanges.insert(name);
}
@@ -615,8 +689,8 @@ const std::string QUEUE_REF("queueRef");
void BrokerReplicator::doResponseBind(Variant::Map& values) {
std::string exName = getRefName(EXCHANGE_REF_PREFIX, values[EXCHANGE_REF]);
std::string qName = getRefName(QUEUE_REF_PREFIX, values[QUEUE_REF]);
- boost::shared_ptr<Exchange> exchange = broker.getExchanges().find(exName);
- boost::shared_ptr<Queue> queue = broker.getQueues().find(qName);
+ boost::shared_ptr<Exchange> exchange = exchanges.find(exName);
+ boost::shared_ptr<Queue> queue = queues.find(qName);
// Automatically replicate binding if queue and exchange exist and are replicated
if (exchange && replicationTest.replicateLevel(exchange->getArgs()) &&
@@ -628,7 +702,7 @@ void BrokerReplicator::doResponseBind(Va
<< " key:" << key);
framing::FieldTable args;
qpid::amqp_0_10::translate(asMapVoid(values[ARGUMENTS]), args);
- exchange->bind(queue, key, &args);
+ queue->bind(exchange, key, args);
}
}
@@ -661,7 +735,7 @@ boost::shared_ptr<QueueReplicator> Broke
if (replicationTest.replicateLevel(queue->getSettings().storeSettings) == ALL) {
boost::shared_ptr<QueueReplicator> qr(
new QueueReplicator(haBroker, queue, link));
- if (!broker.getExchanges().registerExchange(qr))
+ if (!exchanges.registerExchange(qr))
throw Exception(QPID_MSG("Duplicate queue replicator " << qr->getName()));
qr->activate();
return qr;
@@ -670,7 +744,7 @@ boost::shared_ptr<QueueReplicator> Broke
}
void BrokerReplicator::deleteQueue(const std::string& name, bool purge) {
- Queue::shared_ptr queue = broker.getQueues().find(name);
+ Queue::shared_ptr queue = queues.find(name);
if (queue) {
// Purge before deleting to ensure that we don't reroute any
// messages. Any reroutes will be done at the primary and
@@ -709,11 +783,9 @@ boost::shared_ptr<QueueReplicator> Broke
remoteHost);
boost::shared_ptr<QueueReplicator> qr;
if (!findQueueReplicator(name)) qr = startQueueReplicator(result.first);
- if (result.second) {
- if (!alternateExchange.empty()) {
- alternates.setAlternate(
- alternateExchange, boost::bind(&Queue::setAlternateExchange, result.first, _1));
- }
+ if (result.second && !alternateExchange.empty()) {
+ alternates.setAlternate(
+ alternateExchange, boost::bind(&Queue::setAlternateExchange, result.first, _1));
}
return qr;
}
@@ -734,7 +806,6 @@ BrokerReplicator::CreateExchangeResult B
args,
userId,
remoteHost);
-
alternates.addExchange(result.first);
if (!alternateExchange.empty()) {
alternates.setAlternate(
@@ -749,51 +820,7 @@ bool BrokerReplicator::isBound(boost::sh
string BrokerReplicator::getType() const { return QPID_CONFIGURATION_REPLICATOR; }
-
-// BrokerReplicator::Cleaner
-
-BrokerReplicator::Cleaner::Cleaner(BrokerReplicator& br) : brokerReplicator(br) {}
-
-void BrokerReplicator::Cleaner::start() {
- queues.clear();
- exchanges.clear();
- brokerReplicator.broker.getExchanges().eachExchange(boost::bind(&Cleaner::addExchange, this, _1));
- brokerReplicator.broker.getQueues().eachQueue(boost::bind(&Cleaner::addQueue, this, _1));
-}
-
-void BrokerReplicator::Cleaner::addExchange(boost::shared_ptr<broker::Exchange> ex) {
- if (brokerReplicator.replicationTest.isReplicated(CONFIGURATION, *ex))
- exchanges.insert(ex->getName());
-}
-
-void BrokerReplicator::Cleaner::addQueue(boost::shared_ptr<broker::Queue> q) {
- if (brokerReplicator.replicationTest.isReplicated(CONFIGURATION, *q))
- queues.insert(q->getName());
-}
-
-void BrokerReplicator::Cleaner::forgetExchange(const std::string& name) {
- exchanges.erase(name);
-}
-
-void BrokerReplicator::Cleaner::forgetQueue(const std::string& name) {
- queues.erase(name);
-}
-
-void BrokerReplicator::Cleaner::cleanExchanges() {
- for_each(exchanges.begin(), exchanges.end(),
- boost::bind(&BrokerReplicator::deleteExchange, &brokerReplicator, _1));
- exchanges.clear();
-}
-
-void BrokerReplicator::Cleaner::cleanQueues() {
- for_each(queues.begin(), queues.end(),
- boost::bind(&BrokerReplicator::deleteQueue, &brokerReplicator, _1, true));
- queues.clear();
-}
-
-void BrokerReplicator::autoDeleteCheck(
- boost::shared_ptr<Exchange> ex, set<string>& result)
-{
+void BrokerReplicator::autoDeleteCheck(boost::shared_ptr<Exchange> ex) {
boost::shared_ptr<QueueReplicator> qr(boost::dynamic_pointer_cast<QueueReplicator>(ex));
if (!qr) return;
assert(qr);
@@ -803,8 +830,9 @@ void BrokerReplicator::autoDeleteCheck(
Queue::tryAutoDelete(broker, qr->getQueue(), remoteHost, userId);
}
else {
- // Mark for immediate deletion.
- result.insert(qr->getQueue()->getName());
+ // Delete immediately. Don't purge, the primary is gone so we need
+ // to reroute the deleted messages.
+ deleteQueue(qr->getQueue()->getName(), false);
}
}
}
@@ -813,13 +841,12 @@ void BrokerReplicator::disconnected() {
QPID_LOG(info, logPrefix << "Disconnected");
connection = 0;
// Clean up auto-delete queues
- set<string> deleteQueues;
- broker.getExchanges().eachExchange(boost::bind(&BrokerReplicator::autoDeleteCheck,
- this, _1, boost::ref(deleteQueues)));
- // Don't purge before deleting, the primary is gone so we need to
- // reroute the deleted messages.
- for_each(deleteQueues.begin(), deleteQueues.end(),
- boost::bind(&BrokerReplicator::deleteQueue, this, _1, false));
+ vector<boost::shared_ptr<Exchange> > collect;
+ // Make a copy so we can work outside the ExchangeRegistry lock
+ exchanges.eachExchange(
+ boost::bind(&vector<boost::shared_ptr<Exchange> >::push_back, ref(collect), _1));
+ for_each(collect.begin(), collect.end(),
+ boost::bind(&BrokerReplicator::autoDeleteCheck, this, _1));
}
}} // namespace broker
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/BrokerReplicator.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/BrokerReplicator.h?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/BrokerReplicator.h (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/BrokerReplicator.h Mon Nov 19 12:47:53 2012
@@ -41,6 +41,8 @@ class Link;
class Bridge;
class SessionHandler;
class Connection;
+class QueueRegistry;
+class ExchangeRegistry;
}
namespace framing {
@@ -90,34 +92,7 @@ class BrokerReplicator : public broker::
typedef std::map<std::string, QueueReplicatorPtr> QueueReplicatorMap;
-
- /** Keep track of queues and exchanges that need to be cleaned up. */
- class Cleaner {
- public:
- Cleaner(BrokerReplicator&);
-
- /** Scan for existing queues and exchanges. */
- void start();
-
- // Forget a queue/exchange that does not need cleaning
- void forgetExchange(const std::string& name);
- void forgetQueue(const std::string& name);
- // Clean up queues/exchange that are no longer on primary
- void cleanExchanges();
- void cleanQueues();
-
- private:
- typedef std::set<std::string> Names;
-
- // add a queue/exchange that may need cleaning.
- void addExchange(boost::shared_ptr<broker::Exchange>);
- void addQueue(boost::shared_ptr<broker::Queue>);
-
- BrokerReplicator& brokerReplicator;
- Names queues, exchanges;
- };
- friend class Cleaner;
-
+ class UpdateTracker;
class ErrorListener;
class ConnectionObserver;
@@ -158,7 +133,7 @@ class BrokerReplicator : public broker::
void deleteQueue(const std::string& name, bool purge=true);
void deleteExchange(const std::string& name);
- void autoDeleteCheck(boost::shared_ptr<broker::Exchange>, std::set<std::string>&);
+ void autoDeleteCheck(boost::shared_ptr<broker::Exchange>);
void disconnected();
std::string logPrefix;
@@ -166,15 +141,18 @@ class BrokerReplicator : public broker::
ReplicationTest replicationTest;
HaBroker& haBroker;
broker::Broker& broker;
+ broker::ExchangeRegistry& exchanges;
+ broker::QueueRegistry& queues;
boost::shared_ptr<broker::Link> link;
bool initialized;
AlternateExchangeSetter alternates;
qpid::Address primary;
typedef std::set<std::string> StringSet;
StringSet replicatedExchanges; // exchanges that have been replicated.
- Cleaner cleaner;
broker::Connection* connection;
EventDispatchMap dispatch;
+ std::auto_ptr<UpdateTracker> queueTracker;
+ std::auto_ptr<UpdateTracker> exchangeTracker;
};
}} // namespace qpid::broker
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/HaBroker.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/HaBroker.cpp?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/HaBroker.cpp (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/HaBroker.cpp Mon Nov 19 12:47:53 2012
@@ -123,7 +123,7 @@ void HaBroker::initialize() {
}
HaBroker::~HaBroker() {
- QPID_LOG(notice, logPrefix << "Shut down: " << brokerInfo);
+ QPID_LOG(notice, logPrefix << "Shut down");
broker.getConnectionObservers().remove(observer);
}
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/HaPlugin.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/HaPlugin.cpp?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/HaPlugin.cpp (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/HaPlugin.cpp Mon Nov 19 12:47:53 2012
@@ -69,10 +69,16 @@ struct HaPlugin : public Plugin {
void earlyInitialize(Plugin::Target& target) {
broker::Broker* broker = dynamic_cast<broker::Broker*>(&target);
if (broker) {
- // Must create the HaBroker in earlyInitialize so it can set up its
- // connection observer before clients start connecting.
- haBroker.reset(new ha::HaBroker(*broker, settings));
- broker->addFinalizer(boost::bind(&HaPlugin::finalize, this));
+ if (!broker->getManagementAgent()) {
+ QPID_LOG(info, "HA plugin disabled because management is disabled");
+ if (settings.cluster)
+ throw Exception("Cannot start HA: management is disabled");
+ } else {
+ // Must create the HaBroker in earlyInitialize so it can set up its
+ // connection observer before clients start connecting.
+ haBroker.reset(new ha::HaBroker(*broker, settings));
+ broker->addFinalizer(boost::bind(&HaPlugin::finalize, this));
+ }
}
}
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/Primary.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/Primary.cpp?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/Primary.cpp (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/Primary.cpp Mon Nov 19 12:47:53 2012
@@ -31,6 +31,8 @@
#include "qpid/broker/Connection.h"
#include "qpid/broker/Queue.h"
#include "qpid/framing/FieldTable.h"
+#include "qpid/framing/FieldValue.h"
+#include "qpid/framing/Uuid.h"
#include "qpid/log/Statement.h"
#include "qpid/sys/Timer.h"
#include <boost/bind.hpp>
@@ -39,6 +41,8 @@ namespace qpid {
namespace ha {
using sys::Mutex;
+using namespace std;
+using namespace framing;
namespace {
@@ -58,6 +62,8 @@ class PrimaryConfigurationObserver : pub
PrimaryConfigurationObserver(Primary& p) : primary(p) {}
void queueCreate(const Primary::QueuePtr& q) { primary.queueCreate(q); }
void queueDestroy(const Primary::QueuePtr& q) { primary.queueDestroy(q); }
+ void exchangeCreate(const Primary::ExchangePtr& q) { primary.exchangeCreate(q); }
+ void exchangeDestroy(const Primary::ExchangePtr& q) { primary.exchangeDestroy(q); }
private:
Primary& primary;
};
@@ -178,9 +184,12 @@ void Primary::readyReplica(const Replica
}
}
+// NOTE: Called with queue registry lock held.
void Primary::queueCreate(const QueuePtr& q) {
- // Throw if there is an invalid replication level in the queue settings.
- haBroker.getReplicationTest().replicateLevel(q->getSettings().storeSettings);
+ if (haBroker.getReplicationTest().isReplicated(CONFIGURATION, *q)) {
+ // Give each queue a unique id to avoid confusion of same-named queues.
+ q->addArgument(QPID_HA_UUID, types::Variant(Uuid(true)));
+ }
Mutex::ScopedLock l(lock);
for (BackupMap::iterator i = backups.begin(); i != backups.end(); ++i) {
i->second->queueCreate(q);
@@ -188,6 +197,7 @@ void Primary::queueCreate(const QueuePtr
}
}
+// NOTE: Called with queue registry lock held.
void Primary::queueDestroy(const QueuePtr& q) {
Mutex::ScopedLock l(lock);
for (BackupMap::iterator i = backups.begin(); i != backups.end(); ++i)
@@ -195,6 +205,21 @@ void Primary::queueDestroy(const QueuePt
checkReady(l);
}
+// NOTE: Called with exchange registry lock held.
+void Primary::exchangeCreate(const ExchangePtr& ex) {
+ if (haBroker.getReplicationTest().isReplicated(CONFIGURATION, *ex)) {
+ // Give each exchange a unique id to avoid confusion of same-named exchanges.
+ FieldTable args = ex->getArgs();
+ args.set(QPID_HA_UUID, FieldTable::ValuePtr(new UuidValue(&Uuid(true)[0])));
+ ex->setArgs(args);
+ }
+}
+
+// NOTE: Called with exchange registry lock held.
+void Primary::exchangeDestroy(const ExchangePtr&) {
+ // Do nothing
+ }
+
void Primary::opened(broker::Connection& connection) {
BrokerInfo info;
if (ha::ConnectionObserver::getBrokerInfo(connection, info)) {
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/Primary.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/Primary.h?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/Primary.h (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/Primary.h Mon Nov 19 12:47:53 2012
@@ -60,6 +60,7 @@ class Primary
{
public:
typedef boost::shared_ptr<broker::Queue> QueuePtr;
+ typedef boost::shared_ptr<broker::Exchange> ExchangePtr;
static Primary* get() { return instance; }
@@ -72,6 +73,8 @@ class Primary
// Called via ConfigurationObserver
void queueCreate(const QueuePtr&);
void queueDestroy(const QueuePtr&);
+ void exchangeCreate(const ExchangePtr&);
+ void exchangeDestroy(const ExchangePtr&);
// Called via ConnectionObserver
void opened(broker::Connection& connection);
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/QueueReplicator.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/QueueReplicator.cpp?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/QueueReplicator.cpp (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/QueueReplicator.cpp Mon Nov 19 12:47:53 2012
@@ -40,6 +40,7 @@
namespace {
const std::string QPID_REPLICATOR_("qpid.replicator-");
const std::string TYPE_NAME("qpid.queue-replicator");
+const std::string QPID_HA("qpid.ha-");
}
namespace qpid {
@@ -49,9 +50,8 @@ using namespace framing;
using namespace std;
using sys::Mutex;
-const std::string QPID_HA_EVENT_PREFIX("qpid.ha-");
-const std::string QueueReplicator::DEQUEUE_EVENT_KEY(QPID_HA_EVENT_PREFIX+"dequeue");
-const std::string QueueReplicator::POSITION_EVENT_KEY(QPID_HA_EVENT_PREFIX+"position");
+const std::string QueueReplicator::DEQUEUE_EVENT_KEY(QPID_HA+"dequeue");
+const std::string QueueReplicator::POSITION_EVENT_KEY(QPID_HA+"position");
const std::string QueueReplicator::QPID_SYNC_FREQUENCY("qpid.sync_frequency");
std::string QueueReplicator::replicatorName(const std::string& queueName) {
@@ -63,7 +63,7 @@ bool QueueReplicator::isReplicatorName(c
}
bool QueueReplicator::isEventKey(const std::string key) {
- const std::string& prefix = QPID_HA_EVENT_PREFIX;
+ const std::string& prefix = QPID_HA;
bool ret = key.size() > prefix.size() && key.compare(0, prefix.size(), prefix) == 0;
return ret;
}
@@ -114,7 +114,9 @@ QueueReplicator::QueueReplicator(HaBroke
args.setString(QPID_REPLICATE, printable(NONE).str());
Uuid uuid(true);
bridgeName = replicatorName(q->getName()) + std::string(".") + uuid.str();
- getArgs().setString(QPID_REPLICATE, printable(NONE).str());
+ framing::FieldTable args = getArgs();
+ args.setString(QPID_REPLICATE, printable(NONE).str());
+ setArgs(args);
}
// This must be separate from the constructor so we can call shared_from_this.
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/types.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/types.cpp?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/types.cpp (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/types.cpp Mon Nov 19 12:47:53 2012
@@ -33,6 +33,7 @@ namespace ha {
using namespace std;
const string QPID_REPLICATE("qpid.replicate");
+const string QPID_HA_UUID("qpid.ha-uuid");
string EnumBase::str() const {
assert(value < count);
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/types.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/types.h?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/types.h (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/ha/types.h Mon Nov 19 12:47:53 2012
@@ -99,6 +99,7 @@ inline bool isBackup(BrokerStatus s) { r
// String constants.
extern const std::string QPID_REPLICATE;
+extern const std::string QPID_HA_UUID;
/** Define IdSet type, not a typedef so we can overload operator << */
class IdSet : public std::set<types::Uuid> {};
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/management/ManagementAgent.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/management/ManagementAgent.cpp?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/management/ManagementAgent.cpp (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/management/ManagementAgent.cpp Mon Nov 19 12:47:53 2012
@@ -34,6 +34,7 @@
#include "qpid/broker/amqp_0_10/MessageTransfer.h"
#include "qpid/sys/Time.h"
#include "qpid/sys/Thread.h"
+#include "qpid/sys/PollableQueue.h"
#include "qpid/broker/ConnectionState.h"
#include "qpid/broker/AclModule.h"
#include "qpid/types/Variant.h"
@@ -62,23 +63,23 @@ namespace _qmf = qmf::org::apache::qpid:
namespace {
- const size_t qmfV1BufferSize(65536);
- const string defaultVendorName("vendor");
- const string defaultProductName("product");
-
- // Create a valid binding key substring by
- // replacing all '.' chars with '_'
- const string keyifyNameStr(const string& name)
- {
- string n2 = name;
-
- size_t pos = n2.find('.');
- while (pos != n2.npos) {
- n2.replace(pos, 1, "_");
- pos = n2.find('.', pos);
- }
- return n2;
+const size_t qmfV1BufferSize(65536);
+const string defaultVendorName("vendor");
+const string defaultProductName("product");
+
+// Create a valid binding key substring by
+// replacing all '.' chars with '_'
+const string keyifyNameStr(const string& name)
+{
+ string n2 = name;
+
+ size_t pos = n2.find('.');
+ while (pos != n2.npos) {
+ n2.replace(pos, 1, "_");
+ pos = n2.find('.', pos);
}
+ return n2;
+}
struct ScopedManagementContext
{
@@ -166,6 +167,9 @@ void ManagementAgent::configure(const st
broker = _broker;
threadPoolSize = _threads;
ManagementObject::maxThreads = threadPoolSize;
+ sendQueue.reset(
+ new EventQueue(boost::bind(&ManagementAgent::sendEvents, this, _1), broker->getPoller()));
+ sendQueue->start();
// Get from file or generate and save to file.
if (dataDir.empty())
@@ -235,13 +239,13 @@ void ManagementAgent::setName(const stri
} else
inst = instance;
- name_address = vendor + ":" + product + ":" + inst;
- attrMap["_instance"] = inst;
- attrMap["_name"] = name_address;
-
- vendorNameKey = keyifyNameStr(vendor);
- productNameKey = keyifyNameStr(product);
- instanceNameKey = keyifyNameStr(inst);
+ name_address = vendor + ":" + product + ":" + inst;
+ attrMap["_instance"] = inst;
+ attrMap["_name"] = name_address;
+
+ vendorNameKey = keyifyNameStr(vendor);
+ productNameKey = keyifyNameStr(product);
+ instanceNameKey = keyifyNameStr(inst);
}
@@ -541,14 +545,9 @@ void ManagementAgent::sendBuffer(Buffer&
dp->setRoutingKey(routingKey);
transfer->getFrames().append(content);
-
Message msg(transfer, transfer);
msg.setIsManagementMessage(true);
-
- DeliverableMessage deliverable (msg, 0);
- try {
- exchange->route(deliverable);
- } catch(exception&) {}
+ sendQueue->push(make_pair(exchange, msg));
buf.reset();
}
@@ -617,10 +616,7 @@ void ManagementAgent::sendBuffer(const s
msg.setIsManagementMessage(true);
msg.computeExpiration(broker->getExpiryPolicy());
- DeliverableMessage deliverable (msg,0);
- try {
- exchange->route(deliverable);
- } catch(exception&) {}
+ sendQueue->push(make_pair(exchange, msg));
}
@@ -1418,7 +1414,7 @@ void ManagementAgent::handleMethodReques
if (!acl->authorise(userId, acl::ACT_ACCESS, acl::OBJ_METHOD, methodName, ¶ms)) {
sendException(rte, rtk, cid, Manageable::StatusText(Manageable::STATUS_FORBIDDEN),
- Manageable::STATUS_FORBIDDEN, viaLocal);
+ Manageable::STATUS_FORBIDDEN, viaLocal);
return;
}
}
@@ -2123,7 +2119,7 @@ bool ManagementAgent::authorizeAgentMess
qpid::broker::amqp_0_10::MessageTransfer& transfer(qpid::broker::amqp_0_10::MessageTransfer::get(msg));
const framing::MessageProperties* p =
- transfer.getFrames().getHeaders()->get<framing::MessageProperties>();
+ transfer.getFrames().getHeaders()->get<framing::MessageProperties>();
const framing::FieldTable *headers = p ? &p->getApplicationHeaders() : 0;
@@ -2939,6 +2935,21 @@ bool ManagementAgent::moveDeletedObjects
return !deleteList.empty();
}
+ManagementAgent::EventQueue::Batch::const_iterator ManagementAgent::sendEvents(
+ const EventQueue::Batch& batch)
+{
+ EventQueue::Batch::const_iterator i;
+ for (i = batch.begin(); i != batch.end(); ++i) {
+ DeliverableMessage deliverable (i->second, 0);
+ try {
+ i->first->route(deliverable);
+ } catch(exception& e) {
+ QPID_LOG(warning, "ManagementAgent failed to route event: " << e.what());
+ }
+ }
+ return i;
+}
+
namespace {
QPID_TSS const qpid::broker::ConnectionState* executionContext = 0;
}
Propchange: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/management/ManagementAgent.cpp
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/cpp/src/qpid/management/ManagementAgent.cpp:r1401294-1411033
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/management/ManagementAgent.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/management/ManagementAgent.h?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/management/ManagementAgent.h (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/management/ManagementAgent.h Mon Nov 19 12:47:53 2012
@@ -34,6 +34,7 @@
#include "qmf/org/apache/qpid/broker/Agent.h"
#include "qmf/org/apache/qpid/broker/Memory.h"
#include "qpid/sys/MemStat.h"
+#include "qpid/sys/PollableQueue.h"
#include "qpid/types/Variant.h"
#include <qpid/framing/AMQFrame.h>
#include <qpid/framing/ResizableBuffer.h>
@@ -340,6 +341,11 @@ private:
typedef std::map<std::string, DeletedObjectList> PendingDeletedObjsMap;
PendingDeletedObjsMap pendingDeletedObjs;
+ // Pollable queue to serialize event messages
+ typedef std::pair<boost::shared_ptr<broker::Exchange>,
+ broker::Message> ExchangeAndMessage;
+ typedef sys::PollableQueue<ExchangeAndMessage> EventQueue;
+
//
// Memory statistics object
//
@@ -350,6 +356,7 @@ private:
void deleteObjectNow(const ObjectId& oid);
void encodeHeader (framing::Buffer& buf, uint8_t opcode, uint32_t seq = 0);
bool checkHeader (framing::Buffer& buf, uint8_t *opcode, uint32_t *seq);
+ EventQueue::Batch::const_iterator sendEvents(const EventQueue::Batch& batch);
void sendBuffer(framing::Buffer& buf,
qpid::broker::Exchange::shared_ptr exchange,
const std::string& routingKey);
@@ -417,6 +424,7 @@ private:
std::string summarizeAgents();
void debugSnapshot(const char* title);
+ std::auto_ptr<EventQueue> sendQueue;
};
void setManagementExecutionContext(const qpid::broker::ConnectionState*);
Propchange: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/management/ManagementAgent.h
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/cpp/src/qpid/management/ManagementAgent.h:r1401294-1411033
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp Mon Nov 19 12:47:53 2012
@@ -250,8 +250,7 @@ void ConnectionContext::acknowledge(boos
void ConnectionContext::attach(boost::shared_ptr<SessionContext> ssn, boost::shared_ptr<SenderContext> lnk)
{
- pn_terminus_t* target = pn_link_target((pn_link_t*) lnk->sender);
- pn_terminus_set_address(target, lnk->getTarget().c_str());
+ lnk->configure();
attach(ssn->session, (pn_link_t*) lnk->sender);
if (!pn_link_remote_target((pn_link_t*) lnk->sender)) {
std::string msg("No such target : ");
@@ -262,10 +261,9 @@ void ConnectionContext::attach(boost::sh
void ConnectionContext::attach(boost::shared_ptr<SessionContext> ssn, boost::shared_ptr<ReceiverContext> lnk)
{
- pn_terminus_t* source = pn_link_source((pn_link_t*) lnk->receiver);
- pn_terminus_set_address(source, lnk->getSource().c_str());
- attach(ssn->session, (pn_link_t*) lnk->receiver, lnk->capacity);
- if (!pn_link_remote_source((pn_link_t*) lnk->receiver)) {
+ lnk->configure();
+ attach(ssn->session, lnk->receiver, lnk->capacity);
+ if (!pn_link_remote_source(lnk->receiver)) {
std::string msg("No such source : ");
msg += lnk->getSource();
throw qpid::messaging::NotFound(msg);
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/DriverImpl.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/DriverImpl.cpp?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/DriverImpl.cpp (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/DriverImpl.cpp Mon Nov 19 12:47:53 2012
@@ -53,7 +53,7 @@ void DriverImpl::stop()
boost::shared_ptr<Transport> DriverImpl::getTransport(const std::string& protocol, TransportContext& connection)
{
boost::shared_ptr<Transport> t(Transport::create(protocol, connection, poller));
- if (!t) throw new qpid::messaging::ConnectionError("No such transport: " + protocol);
+ if (!t) throw qpid::messaging::ConnectionError("No such transport: " + protocol);
return t;
}
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.cpp?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.cpp (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.cpp Mon Nov 19 12:47:53 2012
@@ -18,9 +18,11 @@
* under the License.
*
*/
-#include "ReceiverContext.h"
+#include "qpid/messaging/amqp/ReceiverContext.h"
+#include "qpid/messaging/amqp/AddressHelper.h"
#include "qpid/messaging/Duration.h"
#include "qpid/messaging/Message.h"
+#include "qpid/amqp/descriptors.h"
extern "C" {
#include <proton/engine.h>
}
@@ -29,10 +31,10 @@ namespace qpid {
namespace messaging {
namespace amqp {
//TODO: proper conversion to wide string for address
-ReceiverContext::ReceiverContext(pn_session_t* session, const std::string& n, const std::string& s)
+ReceiverContext::ReceiverContext(pn_session_t* session, const std::string& n, const qpid::messaging::Address& a)
: name(n),
- source(s),
- receiver(pn_receiver(session, source.c_str())),
+ address(a),
+ receiver(pn_receiver(session, name.c_str())),
capacity(0) {}
ReceiverContext::~ReceiverContext()
{
@@ -84,7 +86,52 @@ const std::string& ReceiverContext::getN
const std::string& ReceiverContext::getSource() const
{
- return source;
+ return address.getName();
+}
+namespace {
+pn_bytes_t convert(const std::string& s)
+{
+ pn_bytes_t result;
+ result.start = const_cast<char*>(s.data());
+ result.size = s.size();
+ return result;
+}
+bool hasWildcards(const std::string& key)
+{
+ return key.find('*') != std::string::npos || key.find('#') != std::string::npos;
+}
+
+uint64_t getFilterDescriptor(const std::string& key)
+{
+ return hasWildcards(key) ? qpid::amqp::filters::LEGACY_TOPIC_FILTER_CODE : qpid::amqp::filters::LEGACY_DIRECT_FILTER_CODE;
+}
+}
+
+void ReceiverContext::configure() const
+{
+ configure(pn_link_source(receiver));
+}
+void ReceiverContext::configure(pn_terminus_t* source) const
+{
+ pn_terminus_set_address(source, address.getName().c_str());
+ //dynamic create:
+ AddressHelper helper(address);
+ if (helper.createEnabled(AddressHelper::FOR_RECEIVER)) {
+ helper.setNodeProperties(source);
+ }
+
+ //filter:
+ pn_data_t* filter = pn_terminus_filter(source);
+ pn_data_put_map(filter);
+ pn_data_enter(filter);
+ pn_data_put_symbol(filter, convert("subject"));
+ //TODO: At present inserting described values into the map doesn't seem to work; correct this once resolved
+ //pn_data_put_described(filter);
+ //pn_data_enter(filter);
+ //pn_data_put_ulong(filter, getFilterDescriptor(address.getSubject()));
+ pn_data_put_string(filter, convert(address.getSubject()));
+ //pn_data_exit(filter);
+ pn_data_exit(filter);
}
bool ReceiverContext::isClosed() const
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.h?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.h (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.h Mon Nov 19 12:47:53 2012
@@ -21,11 +21,13 @@
* under the License.
*
*/
+#include "qpid/messaging/Address.h"
#include <string>
#include "qpid/sys/IntegerTypes.h"
struct pn_link_t;
struct pn_session_t;
+struct pn_terminus_t;
namespace qpid {
namespace messaging {
@@ -41,22 +43,25 @@ namespace amqp {
class ReceiverContext
{
public:
- ReceiverContext(pn_session_t* session, const std::string& name, const std::string& source);
+ ReceiverContext(pn_session_t* session, const std::string& name, const qpid::messaging::Address& source);
~ReceiverContext();
void setCapacity(uint32_t);
uint32_t getCapacity();
uint32_t getAvailable();
uint32_t getUnsettled();
+ void attach();
void close();
const std::string& getName() const;
const std::string& getSource() const;
bool isClosed() const;
+ void configure() const;
private:
friend class ConnectionContext;
const std::string name;
- const std::string source;
+ const Address address;
pn_link_t* receiver;
uint32_t capacity;
+ void configure(pn_terminus_t*) const;
};
}}} // namespace qpid::messaging::amqp
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/SenderContext.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/SenderContext.cpp?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/SenderContext.cpp (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/SenderContext.cpp Mon Nov 19 12:47:53 2012
@@ -20,6 +20,7 @@
*/
#include "qpid/messaging/amqp/SenderContext.h"
#include "qpid/messaging/amqp/EncodedMessage.h"
+#include "qpid/messaging/amqp/AddressHelper.h"
#include "qpid/amqp/descriptors.h"
#include "qpid/amqp/MessageEncoder.h"
#include "qpid/messaging/exceptions.h"
@@ -36,10 +37,10 @@ namespace qpid {
namespace messaging {
namespace amqp {
//TODO: proper conversion to wide string for address
-SenderContext::SenderContext(pn_session_t* session, const std::string& n, const std::string& t)
+SenderContext::SenderContext(pn_session_t* session, const std::string& n, const qpid::messaging::Address& a)
: name(n),
- target(t),
- sender(pn_sender(session, target.c_str())), capacity(1000) {}
+ address(a),
+ sender(pn_sender(session, n.c_str())), capacity(1000) {}
SenderContext::~SenderContext()
{
@@ -74,7 +75,7 @@ const std::string& SenderContext::getNam
const std::string& SenderContext::getTarget() const
{
- return target;
+ return address.getName();
}
SenderContext::Delivery* SenderContext::send(const qpid::messaging::Message& message)
@@ -82,7 +83,7 @@ SenderContext::Delivery* SenderContext::
if (processUnsettled() < capacity) {
deliveries.push_back(Delivery(nextId++));
Delivery& delivery = deliveries.back();
- delivery.encode(MessageImplAccess::get(message));
+ delivery.encode(MessageImplAccess::get(message), address);
delivery.send(sender);
return &delivery;
} else {
@@ -135,7 +136,7 @@ const std::string EMPTY;
class PropertiesAdapter : public qpid::amqp::MessageEncoder::Properties
{
public:
- PropertiesAdapter(const qpid::messaging::MessageImpl& impl) : msg(impl) {}
+ PropertiesAdapter(const qpid::messaging::MessageImpl& impl, const std::string& s) : msg(impl), subject(s) {}
bool hasMessageId() const
{
return getMessageId().size();
@@ -167,12 +168,12 @@ class PropertiesAdapter : public qpid::a
bool hasSubject() const
{
- return getSubject().size();
+ return subject.size() || getSubject().size();
}
std::string getSubject() const
{
- return msg.getSubject();
+ return subject.size() ? subject : msg.getSubject();
}
bool hasReplyTo() const
@@ -266,16 +267,23 @@ class PropertiesAdapter : public qpid::a
}
private:
const qpid::messaging::MessageImpl& msg;
+ const std::string subject;
};
+
+bool changedSubject(const qpid::messaging::MessageImpl& msg, const qpid::messaging::Address& address)
+{
+ return address.getSubject().size() && address.getSubject() != msg.getSubject();
+}
+
}
SenderContext::Delivery::Delivery(int32_t i) : id(i), token(0) {}
-void SenderContext::Delivery::encode(const qpid::messaging::MessageImpl& msg)
+void SenderContext::Delivery::encode(const qpid::messaging::MessageImpl& msg, const qpid::messaging::Address& address)
{
boost::shared_ptr<const EncodedMessage> original = msg.getEncoded();
- if (original) { //still have the content as received, send at least the bare message unaltered
+ if (original && !changedSubject(msg, address)) { //still have the content as received, send at least the bare message unaltered
//do we need to alter the header? are durable, priority, ttl, first-acquirer, delivery-count different from what was received?
if (original->hasHeaderChanged(msg)) {
//since as yet have no annotations, just write the revised header then the rest of the message as received
@@ -293,7 +301,7 @@ void SenderContext::Delivery::encode(con
}
} else {
HeaderAdapter header(msg);
- PropertiesAdapter properties(msg);
+ PropertiesAdapter properties(msg, address.getSubject());
//compute size:
encoded.resize(qpid::amqp::MessageEncoder::getEncodedSize(header, properties, msg.getHeaders(), msg.getBytes()));
QPID_LOG(debug, "Sending message, buffer is " << encoded.getSize() << " bytes")
@@ -329,4 +337,17 @@ bool SenderContext::Delivery::accepted()
return pn_delivery_remote_state(token) == PN_ACCEPTED;
}
+void SenderContext::configure() const
+{
+ configure(pn_link_target(sender));
+}
+void SenderContext::configure(pn_terminus_t* target) const
+{
+ pn_terminus_set_address(target, address.getName().c_str());
+ //dynamic create:
+ AddressHelper helper(address);
+ if (helper.createEnabled(AddressHelper::FOR_SENDER)) {
+ helper.setNodeProperties(target);
+ }
+}
}}} // namespace qpid::messaging::amqp
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/SenderContext.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/SenderContext.h?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/SenderContext.h (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/SenderContext.h Mon Nov 19 12:47:53 2012
@@ -25,11 +25,13 @@
#include <string>
#include <vector>
#include "qpid/sys/IntegerTypes.h"
+#include "qpid/messaging/Address.h"
#include "qpid/messaging/amqp/EncodedMessage.h"
struct pn_delivery_t;
struct pn_link_t;
struct pn_session_t;
+struct pn_terminus_t;
namespace qpid {
namespace messaging {
@@ -48,7 +50,7 @@ class SenderContext
{
public:
Delivery(int32_t id);
- void encode(const qpid::messaging::MessageImpl& message);
+ void encode(const qpid::messaging::MessageImpl& message, const qpid::messaging::Address&);
void send(pn_link_t*);
bool accepted();
private:
@@ -57,7 +59,7 @@ class SenderContext
EncodedMessage encoded;
};
- SenderContext(pn_session_t* session, const std::string& name, const std::string& target);
+ SenderContext(pn_session_t* session, const std::string& name, const qpid::messaging::Address& target);
~SenderContext();
void close();
void setCapacity(uint32_t);
@@ -66,18 +68,20 @@ class SenderContext
const std::string& getName() const;
const std::string& getTarget() const;
Delivery* send(const qpid::messaging::Message& message);
+ void configure() const;
private:
friend class ConnectionContext;
typedef std::deque<Delivery> Deliveries;
const std::string name;
- const std::string target;
+ const qpid::messaging::Address address;
pn_link_t* sender;
int32_t nextId;
Deliveries deliveries;
uint32_t capacity;
uint32_t processUnsettled();
+ void configure(pn_terminus_t*) const;
};
}}} // namespace qpid::messaging::amqp
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/SessionContext.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/SessionContext.cpp?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/SessionContext.cpp (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/SessionContext.cpp Mon Nov 19 12:47:53 2012
@@ -49,7 +49,7 @@ boost::shared_ptr<SenderContext> Session
for (SenderMap::const_iterator i = senders.find(name); i != senders.end(); i = senders.find(name)) {
name = (boost::format("%1%_%2%") % address.getName() % ++count).str();
}
- boost::shared_ptr<SenderContext> s(new SenderContext(session, name, address.str()));
+ boost::shared_ptr<SenderContext> s(new SenderContext(session, name, address));
senders[name] = s;
return s;
}
@@ -62,7 +62,7 @@ boost::shared_ptr<ReceiverContext> Sessi
for (ReceiverMap::const_iterator i = receivers.find(name); i != receivers.end(); i = receivers.find(name)) {
name = (boost::format("%1%_%2%") % address.getName() % ++count).str();
}
- boost::shared_ptr<ReceiverContext> r(new ReceiverContext(session, name, address.str()));
+ boost::shared_ptr<ReceiverContext> r(new ReceiverContext(session, name, address));
receivers[name] = r;
return r;
}
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/SslTransport.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/SslTransport.cpp?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/SslTransport.cpp (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/SslTransport.cpp Mon Nov 19 12:47:53 2012
@@ -20,7 +20,8 @@
*/
#include "SslTransport.h"
#include "TransportContext.h"
-#include "qpid/sys/ssl/SslIo.h"
+#include "qpid/sys/ssl/SslSocket.h"
+#include "qpid/sys/AsynchIO.h"
#include "qpid/sys/ConnectionCodec.h"
#include "qpid/sys/Poller.h"
#include "qpid/log/Statement.h"
@@ -51,18 +52,19 @@ struct StaticInit
}
-SslTransport::SslTransport(TransportContext& c, boost::shared_ptr<Poller> p) : context(c), aio(0), poller(p) {}
+SslTransport::SslTransport(TransportContext& c, boost::shared_ptr<Poller> p) : context(c), connector(0), aio(0), poller(p) {}
void SslTransport::connect(const std::string& host, const std::string& port)
{
+ assert(!connector);
assert(!aio);
- try {
- socket.connect(host, port);
- connected(socket);
- } catch (const std::exception& e) {
- failed(e.what());
- }
+ connector = AsynchConnector::create(
+ socket,
+ host, port,
+ boost::bind(&SslTransport::connected, this, _1),
+ boost::bind(&SslTransport::failed, this, _3));
+ connector->start(poller);
}
void SslTransport::failed(const std::string& msg)
@@ -72,22 +74,22 @@ void SslTransport::failed(const std::str
context.closed();
}
-void SslTransport::connected(const SslSocket&)
+void SslTransport::connected(const Socket&)
{
context.opened();
- aio = new SslIO(socket,
- boost::bind(&SslTransport::read, this, _1, _2),
- boost::bind(&SslTransport::eof, this, _1),
- boost::bind(&SslTransport::disconnected, this, _1),
- boost::bind(&SslTransport::socketClosed, this, _1, _2),
- 0, // nobuffs
- boost::bind(&SslTransport::write, this, _1));
+ aio = AsynchIO::create(socket,
+ boost::bind(&SslTransport::read, this, _1, _2),
+ boost::bind(&SslTransport::eof, this, _1),
+ boost::bind(&SslTransport::disconnected, this, _1),
+ boost::bind(&SslTransport::socketClosed, this, _1, _2),
+ 0, // nobuffs
+ boost::bind(&SslTransport::write, this, _1));
aio->createBuffers(std::numeric_limits<uint16_t>::max());//note: AMQP 1.0 _can_ handle large frame sizes
id = boost::str(boost::format("[%1%]") % socket.getFullAddress());
aio->start(poller);
}
-void SslTransport::read(SslIO&, SslIO::BufferBase* buffer)
+void SslTransport::read(AsynchIO&, AsynchIO::BufferBase* buffer)
{
int32_t decoded = context.getCodec().decode(buffer->bytes+buffer->dataStart, buffer->dataCount);
if (decoded < buffer->dataCount) {
@@ -101,10 +103,10 @@ void SslTransport::read(SslIO&, SslIO::B
}
}
-void SslTransport::write(SslIO&)
+void SslTransport::write(AsynchIO&)
{
if (context.getCodec().canEncode()) {
- SslIO::BufferBase* buffer = aio->getQueuedBuffer();
+ AsynchIO::BufferBase* buffer = aio->getQueuedBuffer();
if (buffer) {
size_t encoded = context.getCodec().encode(buffer->bytes, buffer->byteCount);
@@ -123,18 +125,18 @@ void SslTransport::close()
aio->queueWriteClose();
}
-void SslTransport::eof(SslIO&)
+void SslTransport::eof(AsynchIO&)
{
close();
}
-void SslTransport::disconnected(SslIO&)
+void SslTransport::disconnected(AsynchIO&)
{
close();
socketClosed(*aio, socket);
}
-void SslTransport::socketClosed(SslIO&, const SslSocket&)
+void SslTransport::socketClosed(AsynchIO&, const Socket&)
{
if (aio)
aio->queueForDeletion();
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/SslTransport.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/SslTransport.h?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/SslTransport.h (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/SslTransport.h Mon Nov 19 12:47:53 2012
@@ -30,10 +30,9 @@ namespace qpid {
namespace sys {
class ConnectionCodec;
class Poller;
-namespace ssl {
-class SslIO;
-class SslIOBufferBase;
-}
+class AsynchConnector;
+class AsynchIO;
+class AsynchIOBufferBase;
}
namespace messaging {
@@ -54,18 +53,19 @@ class SslTransport : public Transport
private:
qpid::sys::ssl::SslSocket socket;
TransportContext& context;
- qpid::sys::ssl::SslIO* aio;
+ qpid::sys::AsynchConnector* connector;
+ qpid::sys::AsynchIO* aio;
boost::shared_ptr<qpid::sys::Poller> poller;
bool closed;
std::string id;
- void connected(const qpid::sys::ssl::SslSocket&);
+ void connected(const qpid::sys::Socket&);
void failed(const std::string& msg);
- void read(qpid::sys::ssl::SslIO&, qpid::sys::ssl::SslIOBufferBase*);
- void write(qpid::sys::ssl::SslIO&);
- void eof(qpid::sys::ssl::SslIO&);
- void disconnected(qpid::sys::ssl::SslIO&);
- void socketClosed(qpid::sys::ssl::SslIO&, const qpid::sys::ssl::SslSocket&);
+ void read(qpid::sys::AsynchIO&, qpid::sys::AsynchIOBufferBase*);
+ void write(qpid::sys::AsynchIO&);
+ void eof(qpid::sys::AsynchIO&);
+ void disconnected(qpid::sys::AsynchIO&);
+ void socketClosed(qpid::sys::AsynchIO&, const qpid::sys::Socket&);
friend class DriverImpl;
};
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/TcpTransport.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/TcpTransport.cpp?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/TcpTransport.cpp (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/TcpTransport.cpp Mon Nov 19 12:47:53 2012
@@ -48,14 +48,14 @@ struct StaticInit
} init;
}
-TcpTransport::TcpTransport(TransportContext& c, boost::shared_ptr<Poller> p) : context(c), connector(0), aio(0), poller(p) {}
+TcpTransport::TcpTransport(TransportContext& c, boost::shared_ptr<Poller> p) : socket(createSocket()), context(c), connector(0), aio(0), poller(p) {}
void TcpTransport::connect(const std::string& host, const std::string& port)
{
assert(!connector);
assert(!aio);
connector = AsynchConnector::create(
- socket,
+ *socket,
host, port,
boost::bind(&TcpTransport::connected, this, _1),
boost::bind(&TcpTransport::failed, this, _3));
@@ -67,7 +67,7 @@ void TcpTransport::failed(const std::str
{
QPID_LOG(debug, "Failed to connect: " << msg);
connector = 0;
- socket.close();
+ socket->close();
context.closed();
}
@@ -75,7 +75,7 @@ void TcpTransport::connected(const Socke
{
context.opened();
connector = 0;
- aio = AsynchIO::create(socket,
+ aio = AsynchIO::create(*socket,
boost::bind(&TcpTransport::read, this, _1, _2),
boost::bind(&TcpTransport::eof, this, _1),
boost::bind(&TcpTransport::disconnected, this, _1),
@@ -83,7 +83,7 @@ void TcpTransport::connected(const Socke
0, // nobuffs
boost::bind(&TcpTransport::write, this, _1));
aio->createBuffers(std::numeric_limits<uint16_t>::max());//note: AMQP 1.0 _can_ handle large frame sizes
- id = boost::str(boost::format("[%1%]") % socket.getFullAddress());
+ id = boost::str(boost::format("[%1%]") % socket->getFullAddress());
aio->start(poller);
}
@@ -131,7 +131,7 @@ void TcpTransport::eof(AsynchIO&)
void TcpTransport::disconnected(AsynchIO&)
{
close();
- socketClosed(*aio, socket);
+ socketClosed(*aio, *socket);
}
void TcpTransport::socketClosed(AsynchIO&, const Socket&)
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/TcpTransport.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/TcpTransport.h?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/TcpTransport.h (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/messaging/amqp/TcpTransport.h Mon Nov 19 12:47:53 2012
@@ -24,6 +24,7 @@
#include "qpid/messaging/amqp/Transport.h"
#include "qpid/sys/Mutex.h"
#include "qpid/sys/Socket.h"
+#include <boost/scoped_ptr.hpp>
#include <boost/shared_ptr.hpp>
namespace qpid {
@@ -51,7 +52,7 @@ class TcpTransport : public Transport
void giveReadCredit(int32_t) {}
private:
- qpid::sys::Socket socket;
+ boost::scoped_ptr<qpid::sys::Socket> socket;
TransportContext& context;
qpid::sys::AsynchConnector* connector;
qpid::sys::AsynchIO* aio;
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/AggregateOutput.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/AggregateOutput.cpp?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/AggregateOutput.cpp (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/AggregateOutput.cpp Mon Nov 19 12:47:53 2012
@@ -51,6 +51,7 @@ bool AggregateOutput::doOutput() {
while (!tasks.empty()) {
OutputTask* t=tasks.front();
tasks.pop_front();
+ taskSet.erase(t);
bool didOutput;
{
// Allow concurrent call to addOutputTask.
@@ -59,7 +60,9 @@ bool AggregateOutput::doOutput() {
didOutput = t->doOutput();
}
if (didOutput) {
- tasks.push_back(t);
+ if (taskSet.insert(t).second) {
+ tasks.push_back(t);
+ }
return true;
}
}
@@ -68,12 +71,15 @@ bool AggregateOutput::doOutput() {
void AggregateOutput::addOutputTask(OutputTask* task) {
Mutex::ScopedLock l(lock);
- tasks.push_back(task);
+ if (taskSet.insert(task).second) {
+ tasks.push_back(task);
+ }
}
void AggregateOutput::removeOutputTask(OutputTask* task) {
Mutex::ScopedLock l(lock);
while (busy) lock.wait();
+ taskSet.erase(task);
tasks.erase(std::remove(tasks.begin(), tasks.end(), task), tasks.end());
}
@@ -81,6 +87,7 @@ void AggregateOutput::removeAll()
{
Mutex::ScopedLock l(lock);
while (busy) lock.wait();
+ taskSet.clear();
tasks.clear();
}
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/AggregateOutput.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/AggregateOutput.h?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/AggregateOutput.h (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/AggregateOutput.h Mon Nov 19 12:47:53 2012
@@ -28,6 +28,7 @@
#include <algorithm>
#include <deque>
+#include <set>
namespace qpid {
namespace sys {
@@ -44,9 +45,11 @@ namespace sys {
class QPID_COMMON_CLASS_EXTERN AggregateOutput : public OutputTask, public OutputControl
{
typedef std::deque<OutputTask*> TaskList;
+ typedef std::set<OutputTask*> TaskSet;
Monitor lock;
TaskList tasks;
+ TaskSet taskSet;
bool busy;
OutputControl& control;
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/AsynchIO.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/AsynchIO.h?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/AsynchIO.h (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/AsynchIO.h Mon Nov 19 12:47:53 2012
@@ -21,9 +21,11 @@
*
*/
-#include "qpid/sys/IntegerTypes.h"
#include "qpid/CommonImportExport.h"
+#include "qpid/sys/IntegerTypes.h"
+#include "qpid/sys/SecuritySettings.h"
+
#include <string.h>
#include <boost/function.hpp>
@@ -160,6 +162,8 @@ public:
virtual void requestCallback(RequestCallback) = 0;
virtual BufferBase* getQueuedBuffer() = 0;
+ virtual SecuritySettings getSecuritySettings() = 0;
+
protected:
// Derived class manages lifetime; must be constructed using the
// static create() method. Deletes not allowed from outside.
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/AsynchIOHandler.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/AsynchIOHandler.cpp?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/AsynchIOHandler.cpp (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/AsynchIOHandler.cpp Mon Nov 19 12:47:53 2012
@@ -51,7 +51,7 @@ struct ProtocolTimeoutTask : public sys:
}
};
-AsynchIOHandler::AsynchIOHandler(const std::string& id, ConnectionCodec::Factory* f) :
+AsynchIOHandler::AsynchIOHandler(const std::string& id, ConnectionCodec::Factory* f, bool nodict0) :
identifier(id),
aio(0),
factory(f),
@@ -59,6 +59,7 @@ AsynchIOHandler::AsynchIOHandler(const s
reads(0),
readError(false),
isClient(false),
+ nodict(nodict0),
readCredit(InfiniteCredit)
{}
@@ -118,6 +119,15 @@ void AsynchIOHandler::giveReadCredit(int
aio->startReading();
}
+namespace {
+ SecuritySettings getSecuritySettings(AsynchIO* aio, bool nodict)
+ {
+ SecuritySettings settings = aio->getSecuritySettings();
+ settings.nodict = nodict;
+ return settings;
+ }
+}
+
void AsynchIOHandler::readbuff(AsynchIO& , AsynchIO::BufferBase* buff) {
if (readError) {
return;
@@ -168,7 +178,7 @@ void AsynchIOHandler::readbuff(AsynchIO&
QPID_LOG(debug, "RECV [" << identifier << "]: INIT(" << protocolInit << ")");
try {
- codec = factory->create(protocolInit.getVersion(), *this, identifier, SecuritySettings());
+ codec = factory->create(protocolInit.getVersion(), *this, identifier, getSecuritySettings(aio, nodict));
if (!codec) {
//TODO: may still want to revise this...
//send valid version header & close connection.
@@ -226,7 +236,7 @@ void AsynchIOHandler::nobuffs(AsynchIO&)
void AsynchIOHandler::idle(AsynchIO&){
if (isClient && codec == 0) {
- codec = factory->create(*this, identifier, SecuritySettings());
+ codec = factory->create(*this, identifier, getSecuritySettings(aio, nodict));
write(framing::ProtocolInitiation(codec->getVersion()));
// We've just sent the protocol negotiation so we can cancel the timeout for that
// This is not ideal, because we've not received anything yet, but heartbeats will
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/AsynchIOHandler.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/AsynchIOHandler.h?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/AsynchIOHandler.h (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/AsynchIOHandler.h Mon Nov 19 12:47:53 2012
@@ -51,6 +51,7 @@ class AsynchIOHandler : public OutputCon
uint32_t reads;
bool readError;
bool isClient;
+ bool nodict;
AtomicValue<int32_t> readCredit;
static const int32_t InfiniteCredit = -1;
Mutex creditLock;
@@ -59,7 +60,7 @@ class AsynchIOHandler : public OutputCon
void write(const framing::ProtocolInitiation&);
public:
- QPID_COMMON_EXTERN AsynchIOHandler(const std::string& id, qpid::sys::ConnectionCodec::Factory* f );
+ QPID_COMMON_EXTERN AsynchIOHandler(const std::string& id, qpid::sys::ConnectionCodec::Factory* f, bool nodict);
QPID_COMMON_EXTERN ~AsynchIOHandler();
QPID_COMMON_EXTERN void init(AsynchIO* a, Timer& timer, uint32_t maxTime);
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/SecuritySettings.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/SecuritySettings.h?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/SecuritySettings.h (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/SecuritySettings.h Mon Nov 19 12:47:53 2012
@@ -21,6 +21,8 @@
* under the License.
*
*/
+#include <string>
+
namespace qpid {
namespace sys {
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/Socket.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/Socket.h?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/Socket.h (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/Socket.h Mon Nov 19 12:47:53 2012
@@ -22,7 +22,6 @@
*
*/
-#include "qpid/sys/IOHandle.h"
#include "qpid/sys/IntegerTypes.h"
#include "qpid/CommonImportExport.h"
#include <string>
@@ -31,45 +30,43 @@ namespace qpid {
namespace sys {
class Duration;
+class IOHandle;
class SocketAddress;
-class QPID_COMMON_CLASS_EXTERN Socket : public IOHandle
+class Socket
{
public:
- /** Create a socket wrapper for descriptor. */
- QPID_COMMON_EXTERN Socket();
+ virtual ~Socket() {};
- /** Create a new Socket which is the same address family as this one */
- QPID_COMMON_EXTERN Socket* createSameTypeSocket() const;
+ virtual operator const IOHandle&() const = 0;
/** Set socket non blocking */
- void setNonblocking() const;
+ virtual void setNonblocking() const = 0;
- QPID_COMMON_EXTERN void setTcpNoDelay() const;
+ virtual void setTcpNoDelay() const = 0;
- QPID_COMMON_EXTERN void connect(const std::string& host, const std::string& port) const;
- QPID_COMMON_EXTERN void connect(const SocketAddress&) const;
+ virtual void connect(const SocketAddress&) const = 0;
+ virtual void finishConnect(const SocketAddress&) const = 0;
- QPID_COMMON_EXTERN void close() const;
+ virtual void close() const = 0;
/** Bind to a port and start listening.
*@param port 0 means choose an available port.
*@param backlog maximum number of pending connections.
*@return The bound port.
*/
- QPID_COMMON_EXTERN int listen(const std::string& host = "", const std::string& port = "0", int backlog = 10) const;
- QPID_COMMON_EXTERN int listen(const SocketAddress&, int backlog = 10) const;
+ virtual int listen(const SocketAddress&, int backlog = 10) const = 0;
/**
* Returns an address (host and port) for the remote end of the
* socket
*/
- QPID_COMMON_EXTERN std::string getPeerAddress() const;
+ virtual std::string getPeerAddress() const = 0;
/**
* Returns an address (host and port) for the local end of the
* socket
*/
- QPID_COMMON_EXTERN std::string getLocalAddress() const;
+ virtual std::string getLocalAddress() const = 0;
/**
* Returns the full address of the connection: local and remote host and port.
@@ -80,31 +77,24 @@ public:
* Returns the error code stored in the socket. This may be used
* to determine the result of a non-blocking connect.
*/
- QPID_COMMON_EXTERN int getError() const;
+ virtual int getError() const = 0;
/** Accept a connection from a socket that is already listening
* and has an incoming connection
*/
- QPID_COMMON_EXTERN Socket* accept() const;
+ virtual Socket* accept() const = 0;
- // TODO The following are raw operations, maybe they need better wrapping?
- QPID_COMMON_EXTERN int read(void *buf, size_t count) const;
- QPID_COMMON_EXTERN int write(const void *buf, size_t count) const;
-
-private:
- /** Create socket */
- void createSocket(const SocketAddress&) const;
+ virtual int read(void *buf, size_t count) const = 0;
+ virtual int write(const void *buf, size_t count) const = 0;
-public:
- /** Construct socket with existing handle */
- Socket(IOHandlePrivate*);
-
-protected:
- mutable std::string localname;
- mutable std::string peername;
- mutable bool nonblocking;
- mutable bool nodelay;
+ /* Transport security related: */
+ virtual int getKeyLen() const = 0;
+ virtual std::string getClientAuthId() const = 0;
};
+/** Make the default socket for whatever platform we are executing on
+ */
+QPID_COMMON_EXTERN Socket* createSocket();
+
}}
#endif /*!_sys_Socket_h*/
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/SocketAddress.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/SocketAddress.h?rev=1411155&r1=1411154&r2=1411155&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/SocketAddress.h (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/cpp/src/qpid/sys/SocketAddress.h Mon Nov 19 12:47:53 2012
@@ -44,11 +44,12 @@ public:
QPID_COMMON_EXTERN bool nextAddress();
QPID_COMMON_EXTERN std::string asString(bool numeric=true) const;
+ QPID_COMMON_EXTERN std::string getHost() const;
QPID_COMMON_EXTERN void setAddrInfoPort(uint16_t port);
QPID_COMMON_EXTERN static std::string asString(::sockaddr const * const addr, size_t addrlen);
QPID_COMMON_EXTERN static uint16_t getPort(::sockaddr const * const addr);
-
+
private:
std::string host;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org