You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ac...@apache.org on 2009/02/12 20:01:51 UTC
svn commit: r743839 - in /qpid/trunk/qpid/cpp: src/ src/qpid/broker/
src/qpid/client/ src/qpid/cluster/ src/tests/ xml/
Author: aconway
Date: Thu Feb 12 19:01:51 2009
New Revision: 743839
URL: http://svn.apache.org/viewvc?rev=743839&view=rev
Log:
Cluster security support:
- Set correct user ID on update connections.
- Allow configuration of user, pass and mechanism used for update connections.
Added:
qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterSettings.h (with props)
Modified:
qpid/trunk/qpid/cpp/src/cluster.mk
qpid/trunk/qpid/cpp/src/qpid/broker/SessionAdapter.cpp
qpid/trunk/qpid/cpp/src/qpid/client/Connection.cpp
qpid/trunk/qpid/cpp/src/qpid/client/Connection.h
qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp
qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.h
qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp
qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp
qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h
qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.cpp
qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.h
qpid/trunk/qpid/cpp/src/tests/BrokerFixture.h
qpid/trunk/qpid/cpp/src/tests/ClusterFixture.cpp
qpid/trunk/qpid/cpp/src/tests/ClusterFixture.h
qpid/trunk/qpid/cpp/src/tests/cluster_test.cpp
qpid/trunk/qpid/cpp/xml/cluster.xml
Modified: qpid/trunk/qpid/cpp/src/cluster.mk
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/cluster.mk?rev=743839&r1=743838&r2=743839&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/cluster.mk (original)
+++ qpid/trunk/qpid/cpp/src/cluster.mk Thu Feb 12 19:01:51 2009
@@ -44,6 +44,7 @@
qpid/cluster/ClusterMap.cpp \
qpid/cluster/ClusterMap.h \
qpid/cluster/ClusterPlugin.cpp \
+ qpid/cluster/ClusterSettings.h \
qpid/cluster/Connection.cpp \
qpid/cluster/Connection.h \
qpid/cluster/ConnectionCodec.cpp \
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/SessionAdapter.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/SessionAdapter.cpp?rev=743839&r1=743838&r2=743839&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/SessionAdapter.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/SessionAdapter.cpp Thu Feb 12 19:01:51 2009
@@ -70,7 +70,7 @@
params.insert(make_pair(acl::PROP_PASSIVE, std::string(passive ? "true" : "false") ));
params.insert(make_pair(acl::PROP_DURABLE, std::string(durable ? "true" : "false")));
if (!acl->authorise(getConnection().getUserId(),acl::ACT_CREATE,acl::OBJ_EXCHANGE,exchange,¶ms) )
- throw NotAllowedException("ACL denied exhange declare request");
+ throw NotAllowedException(QPID_MSG("ACL denied exhange declare request from " << getConnection().getUserId()));
}
//TODO: implement autoDelete
@@ -130,7 +130,7 @@
AclModule* acl = getBroker().getAcl();
if (acl) {
if (!acl->authorise(getConnection().getUserId(),acl::ACT_DELETE,acl::OBJ_EXCHANGE,name,NULL) )
- throw NotAllowedException("ACL denied exhange delete request");
+ throw NotAllowedException(QPID_MSG("ACL denied exhange delete request from " << getConnection().getUserId()));
}
//TODO: implement unused
@@ -150,7 +150,7 @@
AclModule* acl = getBroker().getAcl();
if (acl) {
if (!acl->authorise(getConnection().getUserId(),acl::ACT_ACCESS,acl::OBJ_EXCHANGE,name,NULL) )
- throw NotAllowedException("ACL denied exhange query request");
+ throw NotAllowedException(QPID_MSG("ACL denied exhange query request from " << getConnection().getUserId()));
}
try {
@@ -168,7 +168,7 @@
AclModule* acl = getBroker().getAcl();
if (acl) {
if (!acl->authorise(getConnection().getUserId(),acl::ACT_BIND,acl::OBJ_EXCHANGE,exchangeName,routingKey) )
- throw NotAllowedException("ACL denied exhange bind request");
+ throw NotAllowedException(QPID_MSG("ACL denied exhange bind request from " << getConnection().getUserId()));
}
Queue::shared_ptr queue = getQueue(queueName);
@@ -200,7 +200,7 @@
params.insert(make_pair(acl::PROP_QUEUENAME, queueName));
params.insert(make_pair(acl::PROP_ROUTINGKEY, routingKey));
if (!acl->authorise(getConnection().getUserId(),acl::ACT_UNBIND,acl::OBJ_EXCHANGE,exchangeName,¶ms) )
- throw NotAllowedException("ACL denied exchange unbind request");
+ throw NotAllowedException(QPID_MSG("ACL denied exchange unbind request from " << getConnection().getUserId()));
}
Queue::shared_ptr queue = getQueue(queueName);
@@ -231,7 +231,7 @@
params.insert(make_pair(acl::PROP_QUEUENAME, queueName));
params.insert(make_pair(acl::PROP_ROUTINGKEY, key));
if (!acl->authorise(getConnection().getUserId(),acl::ACT_CREATE,acl::OBJ_EXCHANGE,exchangeName,¶ms) )
- throw NotAllowedException("ACL denied exhange bound request");
+ throw NotAllowedException(QPID_MSG("ACL denied exhange bound request from " << getConnection().getUserId()));
}
Exchange::shared_ptr exchange;
@@ -297,7 +297,7 @@
AclModule* acl = getBroker().getAcl();
if (acl) {
if (!acl->authorise(getConnection().getUserId(),acl::ACT_ACCESS,acl::OBJ_QUEUE,name,NULL) )
- throw NotAllowedException("ACL denied queue query request");
+ throw NotAllowedException(QPID_MSG("ACL denied queue query request from " << getConnection().getUserId()));
}
Queue::shared_ptr queue = session.getBroker().getQueues().find(name);
@@ -331,7 +331,7 @@
params.insert(make_pair(acl::PROP_EXCLUSIVE, std::string(exclusive ? "true" : "false")));
params.insert(make_pair(acl::PROP_AUTODELETE, std::string(autoDelete ? "true" : "false")));
if (!acl->authorise(getConnection().getUserId(),acl::ACT_CREATE,acl::OBJ_QUEUE,name,¶ms) )
- throw NotAllowedException("ACL denied queue create request");
+ throw NotAllowedException(QPID_MSG("ACL denied queue create request from " << getConnection().getUserId()));
}
Exchange::shared_ptr alternate;
@@ -394,7 +394,7 @@
if (acl)
{
if (!acl->authorise(getConnection().getUserId(),acl::ACT_PURGE,acl::OBJ_QUEUE,queue,NULL) )
- throw NotAllowedException("ACL denied queue purge request");
+ throw NotAllowedException(QPID_MSG("ACL denied queue purge request from " << getConnection().getUserId()));
}
getQueue(queue)->purge();
}
@@ -405,7 +405,7 @@
if (acl)
{
if (!acl->authorise(getConnection().getUserId(),acl::ACT_DELETE,acl::OBJ_QUEUE,queue,NULL) )
- throw NotAllowedException("ACL denied queue delete request");
+ throw NotAllowedException(QPID_MSG("ACL denied queue delete request from " << getConnection().getUserId()));
}
Queue::shared_ptr q = getQueue(queue);
@@ -474,7 +474,7 @@
{
// add flags as needed
if (!acl->authorise(getConnection().getUserId(),acl::ACT_CONSUME,acl::OBJ_QUEUE,queueName,NULL) )
- throw NotAllowedException("ACL denied Queue subscribe request");
+ throw NotAllowedException(QPID_MSG("ACL denied Queue subscribe request from " << getConnection().getUserId()));
}
Queue::shared_ptr queue = getQueue(queueName);
Modified: qpid/trunk/qpid/cpp/src/qpid/client/Connection.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/Connection.cpp?rev=743839&r1=743838&r2=743839&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/Connection.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/Connection.cpp Thu Feb 12 19:01:51 2009
@@ -54,6 +54,15 @@
const std::string& vhost,
uint16_t maxFrameSize)
{
+ ConnectionSettings settings;
+ settings.username = uid;
+ settings.password = pwd;
+ settings.virtualhost = vhost;
+ settings.maxFrameSize = maxFrameSize;
+ open(url, settings);
+}
+
+void Connection::open(const Url& url, const ConnectionSettings& settings) {
if (url.empty())
throw Exception(QPID_MSG("Attempt to open URL with no addresses."));
Url::const_iterator i = url.begin();
@@ -62,14 +71,10 @@
i++;
if (tcp) {
try {
- ConnectionSettings settings;
- settings.host = tcp->host;
- settings.port = tcp->port;
- settings.username = uid;
- settings.password = pwd;
- settings.virtualhost = vhost;
- settings.maxFrameSize = maxFrameSize;
- open(settings);
+ ConnectionSettings cs(settings);
+ cs.host = tcp->host;
+ cs.port = tcp->port;
+ open(cs);
break;
}
catch (const Exception& /*e*/) {
Modified: qpid/trunk/qpid/cpp/src/qpid/client/Connection.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/Connection.h?rev=743839&r1=743838&r2=743839&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/Connection.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/Connection.h Thu Feb 12 19:01:51 2009
@@ -107,6 +107,18 @@
const std::string& virtualhost = "/", uint16_t maxFrameSize=65535);
/**
+ * Opens a connection to a broker using a URL.
+ * If the URL contains multiple addresses, try each in turn
+ * till connection is successful.
+ *
+ * @url address of the broker to connect to.
+ *
+ * @param settings used for any settings not provided by the URL.
+ * Settings provided by the url (e.g. host, port) are ignored.
+ */
+ void open(const Url& url, const ConnectionSettings& settings);
+
+ /**
* Opens a connection to a broker.
*
* @param the settings to use (host, port etc). @see ConnectionSettings.
Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp?rev=743839&r1=743838&r2=743839&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp Thu Feb 12 19:01:51 2009
@@ -17,6 +17,7 @@
*/
#include "Cluster.h"
+#include "ClusterSettings.h"
#include "Connection.h"
#include "UpdateClient.h"
#include "FailoverExchange.h"
@@ -82,16 +83,17 @@
bool invoke(AMQBody& body) { return framing::invoke(*this, body).wasHandled(); }
};
-Cluster::Cluster(const std::string& name_, const Url& url_, broker::Broker& b, bool quorum_, size_t readMax_, size_t writeEstimate_) :
+Cluster::Cluster(const ClusterSettings& set, broker::Broker& b) :
+ settings(set),
broker(b),
mgmtObject(0),
poller(b.getPoller()),
cpg(*this),
- name(name_),
- myUrl(url_),
+ name(settings.name),
+ myUrl(settings.url.empty() ? Url() : Url(settings.url)),
myId(cpg.self()),
- readMax(readMax_),
- writeEstimate(writeEstimate_),
+ readMax(settings.readMax),
+ writeEstimate(settings.writeEstimate),
mcast(cpg, poller, boost::bind(&Cluster::leave, this)),
dispatcher(cpg, poller, boost::bind(&Cluster::leave, this)),
deliverEventQueue(boost::bind(&Cluster::deliveredEvent, this, _1),
@@ -121,7 +123,7 @@
}
failoverExchange.reset(new FailoverExchange(this));
- if (quorum_) quorum.init();
+ if (settings.quorum) quorum.init();
cpg.join(name);
// pump the CPG dispatch manually till we get initialized.
while (!initialized)
@@ -425,10 +427,15 @@
QPID_LOG(info, *this << " stall for update to " << updatee << " at " << url);
deliverFrameQueue.stop();
if (updateThread.id()) updateThread.join(); // Join the previous updatethread.
+ client::ConnectionSettings cs;
+ cs.username = settings.username;
+ cs.password = settings.password;
+ cs.mechanism = settings.mechanism;
updateThread = Thread(
new UpdateClient(myId, updatee, url, broker, map, frameId, connections.values(),
boost::bind(&Cluster::updateOutDone, this),
- boost::bind(&Cluster::updateOutError, this, _1)));
+ boost::bind(&Cluster::updateOutError, this, _1),
+ cs));
}
// Called in update thread.
Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.h?rev=743839&r1=743838&r2=743839&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.h Thu Feb 12 19:01:51 2009
@@ -19,6 +19,7 @@
*
*/
+#include "ClusterSettings.h"
#include "ClusterMap.h"
#include "ConnectionMap.h"
#include "Cpg.h"
@@ -67,9 +68,7 @@
typedef std::vector<ConnectionPtr> Connections;
/** Construct the cluster in plugin earlyInitialize */
- Cluster(const std::string& name, const Url& url, broker::Broker&, bool useQuorum,
- size_t readMax, size_t writeEstimate);
-
+ Cluster(const ClusterSettings&, broker::Broker&);
virtual ~Cluster();
/** Join the cluster in plugin initialize. Requires transport
@@ -178,6 +177,7 @@
void setClusterId(const framing::Uuid&);
// Immutable members set on construction, never changed.
+ ClusterSettings settings;
broker::Broker& broker;
qmf::org::apache::qpid::cluster::Cluster* mgmtObject; // mgnt owns lifecycle
boost::shared_ptr<sys::Poller> poller;
Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp?rev=743839&r1=743838&r2=743839&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp Thu Feb 12 19:01:51 2009
@@ -18,6 +18,7 @@
#include "Connection.h"
#include "ConnectionCodec.h"
+#include "ClusterSettings.h"
#include "qpid/cluster/Cluster.h"
#include "qpid/cluster/ConnectionCodec.h"
@@ -35,6 +36,7 @@
#include "qpid/broker/Exchange.h"
#include "qpid/broker/Queue.h"
#include "qpid/broker/SessionState.h"
+#include "qpid/client/ConnectionSettings.h"
#include <boost/utility/in_place_factory.hpp>
#include <boost/scoped_ptr.hpp>
@@ -48,40 +50,31 @@
using management::ManagementAgent;
using management::ManagementBroker;
-struct ClusterValues {
- string name;
- string url;
- bool quorum;
- size_t readMax, writeEstimate;
-
- ClusterValues() : quorum(false), readMax(10), writeEstimate(64) {}
-
- Url getUrl(uint16_t port) const {
- if (url.empty()) return Url::getIpAddressesUrl(port);
- return Url(url);
- }
-};
-/** Note separating options from values to work around boost version differences.
+/** Note separating options from settings to work around boost version differences.
* Old boost takes a reference to options objects, but new boost makes a copy.
* New boost allows a shared_ptr but that's not compatible with old boost.
*/
struct ClusterOptions : public Options {
- ClusterValues& values;
+ ClusterSettings& settings;
- ClusterOptions(ClusterValues& v) : Options("Cluster Options"), values(v) {
+ ClusterOptions(ClusterSettings& v) : Options("Cluster Options"), settings(v) {
addOptions()
- ("cluster-name", optValue(values.name, "NAME"), "Name of cluster to join")
- ("cluster-url", optValue(values.url,"URL"),
+ ("cluster-name", optValue(settings.name, "NAME"), "Name of cluster to join")
+ ("cluster-url", optValue(settings.url,"URL"),
"URL of this broker, advertized to the cluster.\n"
"Defaults to a URL listing all the local IP addresses\n")
+ ("cluster-username", optValue(settings.username, ""), "Username for connections between brokers")
+ ("cluster-password", optValue(settings.password, ""), "Password for connections between brokers")
+ ("cluster-mechanism", optValue(settings.mechanism, ""), "Authentication mechanism for connections between brokers")
#if HAVE_LIBCMAN
- ("cluster-cman", optValue(values.quorum), "Integrate with Cluster Manager (CMAN) cluster.")
+ ("cluster-cman", optValue(settings.quorum), "Integrate with Cluster Manager (CMAN) cluster.")
#endif
- ("cluster-read-max", optValue(values.readMax,"N"),
+ ("cluster-read-max", optValue(settings.readMax,"N"),
"Experimental: Limit per-client-connection queue of read buffers. 0=no limit.")
- ("cluster-write-estimate", optValue(values.writeEstimate, "Kb"),
- "Experimental: initial estimate for connection write rate per multicast cycle");
+ ("cluster-write-estimate", optValue(settings.writeEstimate, "Kb"),
+ "Experimental: initial estimate for connection write rate per multicast cycle")
+ ;
}
};
@@ -127,26 +120,20 @@
struct ClusterPlugin : public Plugin {
- ClusterValues values;
+ ClusterSettings settings;
ClusterOptions options;
Cluster* cluster;
boost::scoped_ptr<ConnectionCodec::Factory> factory;
- ClusterPlugin() : options(values), cluster(0) {}
+ ClusterPlugin() : options(settings), cluster(0) {}
Options* getOptions() { return &options; }
void earlyInitialize(Plugin::Target& target) {
- if (values.name.empty()) return; // Only if --cluster-name option was specified.
+ if (settings.name.empty()) return; // Only if --cluster-name option was specified.
Broker* broker = dynamic_cast<Broker*>(&target);
if (!broker) return;
- cluster = new Cluster(
- values.name,
- values.url.empty() ? Url() : Url(values.url),
- *broker,
- values.quorum,
- values.readMax, values.writeEstimate*1024
- );
+ cluster = new Cluster(settings, *broker);
broker->setConnectionFactory(
boost::shared_ptr<sys::ConnectionCodec::Factory>(
new ConnectionCodec::Factory(broker->getConnectionFactory(), *cluster)));
Added: qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterSettings.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterSettings.h?rev=743839&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterSettings.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterSettings.h Thu Feb 12 19:01:51 2009
@@ -0,0 +1,48 @@
+#ifndef QPID_CLUSTER_CLUSTERSETTINGS_H
+#define QPID_CLUSTER_CLUSTERSETTINGS_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * 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
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include <qpid/Url.h>
+#include <string>
+
+namespace qpid {
+namespace cluster {
+
+struct ClusterSettings {
+ std::string name;
+ std::string url;
+ bool quorum;
+ size_t readMax, writeEstimate;
+ std::string username, password, mechanism;
+
+ ClusterSettings() : quorum(false), readMax(10), writeEstimate(64), username("guest"), password("guest") {}
+
+ Url getUrl(uint16_t port) const {
+ if (url.empty()) return Url::getIpAddressesUrl(port);
+ return Url(url);
+ }
+};
+
+}} // namespace qpid::cluster
+
+#endif /*!QPID_CLUSTER_CLUSTERSETTINGS_H*/
Propchange: qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterSettings.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterSettings.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp?rev=743839&r1=743838&r2=743839&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp Thu Feb 12 19:01:51 2009
@@ -253,10 +253,11 @@
QPID_LOG(debug, cluster << " received session state update for " << sessionState().getId());
}
-void Connection::shadowReady(uint64_t memberId, uint64_t connectionId) {
+void Connection::shadowReady(uint64_t memberId, uint64_t connectionId, const string& username) {
ConnectionId shadow = ConnectionId(memberId, connectionId);
QPID_LOG(debug, cluster << " catch-up connection " << *this << " becomes shadow " << shadow);
self = shadow;
+ connection.setUserId(username);
}
void Connection::membership(const FieldTable& joiners, const FieldTable& members, uint64_t frameId) {
Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h?rev=743839&r1=743838&r2=743839&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.h Thu Feb 12 19:01:51 2009
@@ -117,7 +117,7 @@
const framing::SequenceNumber& received,
const framing::SequenceSet& unknownCompleted, const SequenceSet& receivedIncomplete);
- void shadowReady(uint64_t memberId, uint64_t connectionId);
+ void shadowReady(uint64_t memberId, uint64_t connectionId, const std::string& username);
void membership(const framing::FieldTable&, const framing::FieldTable&, uint64_t frameId);
Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.cpp?rev=743839&r1=743838&r2=743839&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.cpp Thu Feb 12 19:01:51 2009
@@ -89,21 +89,22 @@
broker::Broker& broker, const ClusterMap& m, uint64_t frameId_,
const Cluster::Connections& cons,
const boost::function<void()>& ok,
- const boost::function<void(const std::exception&)>& fail)
+ const boost::function<void(const std::exception&)>& fail,
+ const client::ConnectionSettings& cs
+)
: updaterId(updater), updateeId(updatee), updateeUrl(url), updaterBroker(broker), map(m),
frameId(frameId_), connections(cons),
connection(catchUpConnection()), shadowConnection(catchUpConnection()),
- done(ok), failed(fail)
+ done(ok), failed(fail)
{
- connection.open(url);
+ connection.open(url, cs);
session = connection.newSession("update_shared");
}
UpdateClient::~UpdateClient() {}
// Reserved exchange/queue name for catch-up, avoid clashes with user queues/exchanges.
-static const char UPDATE_CHARS[] = "qpid.qpid-update";
-const std::string UpdateClient::UPDATE(UPDATE_CHARS, sizeof(UPDATE_CHARS));
+const std::string UpdateClient::UPDATE("qpid.qpid-update");
void UpdateClient::update() {
QPID_LOG(debug, updaterId << " updating state to " << updateeId << " at " << updateeUrl);
@@ -232,7 +233,9 @@
bc.eachSessionHandler(boost::bind(&UpdateClient::updateSession, this, _1));
ClusterConnectionProxy(shadowConnection).shadowReady(
updateConnection->getId().getMember(),
- reinterpret_cast<uint64_t>(updateConnection->getId().getPointer()));
+ reinterpret_cast<uint64_t>(updateConnection->getId().getPointer()),
+ updateConnection->getBrokerConnection().getUserId()
+ );
shadowConnection.close();
QPID_LOG(debug, updaterId << " updated connection " << *updateConnection);
}
Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.h?rev=743839&r1=743838&r2=743839&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/UpdateClient.h Thu Feb 12 19:01:51 2009
@@ -66,7 +66,9 @@
broker::Broker& donor, const ClusterMap& map, uint64_t sequence,
const std::vector<boost::intrusive_ptr<Connection> >& ,
const boost::function<void()>& done,
- const boost::function<void(const std::exception&)>& fail);
+ const boost::function<void(const std::exception&)>& fail,
+ const client::ConnectionSettings&
+ );
~UpdateClient();
void update();
Modified: qpid/trunk/qpid/cpp/src/tests/BrokerFixture.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/BrokerFixture.h?rev=743839&r1=743838&r2=743839&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/BrokerFixture.h (original)
+++ qpid/trunk/qpid/cpp/src/tests/BrokerFixture.h Thu Feb 12 19:01:51 2009
@@ -88,6 +88,7 @@
/** Connection that opens in its constructor */
struct LocalConnection : public qpid::client::Connection {
LocalConnection(uint16_t port) { open("localhost", port); }
+ LocalConnection(const qpid::client::ConnectionSettings& s) { open(s); }
};
/** A local client connection via a socket proxy. */
@@ -96,6 +97,11 @@
ProxyConnection(int brokerPort) : proxy(brokerPort) {
open("localhost", proxy.getPort());
}
+ ProxyConnection(const qpid::client::ConnectionSettings& s) : proxy(s.port) {
+ qpid::client::ConnectionSettings proxySettings(s);
+ proxySettings.port = proxy.getPort();
+ open(proxySettings);
+ }
~ProxyConnection() { close(); }
};
@@ -110,6 +116,8 @@
qpid::client::LocalQueue lq;
ClientT(uint16_t port, const std::string& name=std::string())
: connection(port), session(connection.newSession(name)), subs(session) {}
+ ClientT(const qpid::client::ConnectionSettings& settings, const std::string& name=std::string())
+ : connection(settings), session(connection.newSession(name)), subs(session) {}
~ClientT() { connection.close(); }
};
Modified: qpid/trunk/qpid/cpp/src/tests/ClusterFixture.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/ClusterFixture.cpp?rev=743839&r1=743838&r2=743839&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/ClusterFixture.cpp (original)
+++ qpid/trunk/qpid/cpp/src/tests/ClusterFixture.cpp Thu Feb 12 19:01:51 2009
@@ -67,13 +67,14 @@
add(n);
}
+const ClusterFixture::Args ClusterFixture::DEFAULT_ARGS =
+ list_of<string>("--auth=no")("--no-data-dir");
+
ClusterFixture::Args ClusterFixture::makeArgs(const std::string& prefix) {
Args args = list_of<string>("qpidd " __FILE__)
("--no-module-dir")
("--load-module=../.libs/cluster.so")
("--cluster-name")(name)
- ("--auth=no")
- ("--no-data-dir")
("--log-prefix")(prefix);
args.insert(args.end(), userArgs.begin(), userArgs.end());
return args;
Modified: qpid/trunk/qpid/cpp/src/tests/ClusterFixture.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/ClusterFixture.h?rev=743839&r1=743838&r2=743839&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/ClusterFixture.h (original)
+++ qpid/trunk/qpid/cpp/src/tests/ClusterFixture.h Thu Feb 12 19:01:51 2009
@@ -72,7 +72,7 @@
/** @param localIndex can be -1 meaning don't automatically start a local broker.
* A local broker can be started with addLocal().
*/
- ClusterFixture(size_t n, int localIndex=0, const Args& args=Args());
+ ClusterFixture(size_t n, int localIndex=0, const Args& args=DEFAULT_ARGS);
void add(size_t n) { for (size_t i=0; i < n; ++i) add(); }
void add(); // Add a broker.
void setup();
@@ -86,6 +86,8 @@
void killWithSilencer(size_t n, client::Connection& c, int sig=SIGINT);
private:
+ static const Args DEFAULT_ARGS;
+
void addLocal(); // Add a local broker.
Args makeArgs(const std::string& prefix);
string name;
Modified: qpid/trunk/qpid/cpp/src/tests/cluster_test.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/cluster_test.cpp?rev=743839&r1=743838&r2=743839&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/cluster_test.cpp (original)
+++ qpid/trunk/qpid/cpp/src/tests/cluster_test.cpp Thu Feb 12 19:01:51 2009
@@ -23,6 +23,7 @@
#include "ClusterFixture.h"
#include "qpid/client/Connection.h"
+#include "qpid/client/ConnectionSettings.h"
#include "qpid/client/ConnectionAccess.h"
#include "qpid/client/Session.h"
#include "qpid/client/FailoverListener.h"
@@ -41,6 +42,7 @@
#include <string>
#include <iostream>
+#include <fstream>
#include <iterator>
#include <vector>
#include <set>
@@ -86,11 +88,11 @@
template <class C> set<uint16_t> makeSet(const C& c) {
set<uint16_t> s;
- std::copy(c.begin(), c.end(), std::inserter(s, s.begin()));
+ copy(c.begin(), c.end(), inserter(s, s.begin()));
return s;
}
-template <class T> std::set<uint16_t> knownBrokerPorts(T& source, int n=-1) {
+template <class T> set<uint16_t> knownBrokerPorts(T& source, int n=-1) {
vector<Url> urls = source.getKnownBrokers();
if (n >= 0 && unsigned(n) != urls.size()) {
BOOST_MESSAGE("knownBrokerPorts waiting for " << n << ": " << urls);
@@ -129,13 +131,13 @@
return m.getMessageProperties().getApplicationHeaders().getAsInt64("qpid.msg_sequence");
}
-Message ttlMessage(const std::string& data, const std::string& key, uint64_t ttl) {
+Message ttlMessage(const string& data, const string& key, uint64_t ttl) {
Message m(data, key);
m.getDeliveryProperties().setTtl(ttl);
return m;
}
-vector<std::string> browse(Client& c, const std::string& q, int n) {
+vector<string> browse(Client& c, const string& q, int n) {
SubscriptionSettings browseSettings(
FlowControl::unlimited(),
ACCEPT_MODE_NONE,
@@ -144,7 +146,7 @@
);
LocalQueue lq;
c.subs.subscribe(lq, q, browseSettings);
- vector<std::string> result;
+ vector<string> result;
for (int i = 0; i < n; ++i) {
result.push_back(lq.get(TIMEOUT).getData());
}
@@ -152,6 +154,44 @@
return result;
}
+
+// FIXME aconway 2009-02-12: need to figure out how to test this properly.
+// Current problems:
+// - all brokers share the same data-dir (set ACL without data dir?)
+// - updater's user name not making it through to updatee for ACL checks.
+//
+// QPID_AUTO_TEST_CASE(testAcl) {
+// ofstream policyFile("cluster_test.acl");
+// // FIXME aconway 2009-02-12: guest -> qpidd?
+// policyFile << "acl allow guest@QPID all all" << endl
+// << "acl allow foo@QPID create queue name=foo" << endl
+// << "acl allow bar@QPID create queue name=bar" << endl
+// << "acl deny all create queue" << endl
+// << "acl allow all all" << endl;
+// policyFile.close();
+// ClusterFixture cluster(2,-1, list_of<string>
+// ("--data-dir=.") ("--auth=no")
+// ("--acl-file=cluster_test.acl")
+// ("--cluster-mechanism=PLAIN")
+// ("--load-module=../.libs/acl.so"));
+// Client c0(cluster[0], "c0");
+// Client c1(cluster[1], "c1");
+
+// ConnectionSettings settings;
+// settings.port = cluster[0];
+// settings.username = "foo";
+// Client foo(settings, "foo");
+
+// foo.session.queueDeclare("foo");
+// BOOST_CHECK_EQUAL(c0.session.queueQuery("foo").getQueue(), "foo");
+// BOOST_CHECK_EQUAL(c1.session.queueQuery("foo").getQueue(), "foo");
+
+// BOOST_CHECK_THROW(foo.session.queueDeclare("bar"), int);
+// BOOST_CHECK_EQUAL(c0.session.queueQuery("bar").getQueue(), "");
+// BOOST_CHECK_EQUAL(c1.session.queueQuery("bar").getQueue(), "");
+// }
+
+
QPID_AUTO_TEST_CASE(testMessageTimeToLive) {
// Note: this doesn't actually test for cluster race conditions around TTL,
// it just verifies that basic TTL functionality works.
@@ -162,10 +202,10 @@
c0.session.queueDeclare("q");
c0.session.messageTransfer(arg::content=ttlMessage("a", "q", 200));
c0.session.messageTransfer(arg::content=Message("b", "q"));
- BOOST_CHECK_EQUAL(browse(c1, "q", 2), list_of<std::string>("a")("b"));
+ BOOST_CHECK_EQUAL(browse(c1, "q", 2), list_of<string>("a")("b"));
sys::usleep(300*1000);
- BOOST_CHECK_EQUAL(browse(c0, "q", 1), list_of<std::string>("b"));
- BOOST_CHECK_EQUAL(browse(c1, "q", 1), list_of<std::string>("b"));
+ BOOST_CHECK_EQUAL(browse(c0, "q", 1), list_of<string>("b"));
+ BOOST_CHECK_EQUAL(browse(c1, "q", 1), list_of<string>("b"));
}
QPID_AUTO_TEST_CASE(testSequenceOptions) {
@@ -349,7 +389,7 @@
// Send first 2 frames of message.
MessageTransferBody transfer(
- ProtocolVersion(), std::string(), // default exchange.
+ ProtocolVersion(), string(), // default exchange.
framing::message::ACCEPT_MODE_NONE,
framing::message::ACQUIRE_MODE_PRE_ACQUIRED);
sender.send(transfer, true, false, true, true);
Modified: qpid/trunk/qpid/cpp/xml/cluster.xml
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/xml/cluster.xml?rev=743839&r1=743838&r2=743839&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/xml/cluster.xml (original)
+++ qpid/trunk/qpid/cpp/xml/cluster.xml Thu Feb 12 19:01:51 2009
@@ -124,6 +124,7 @@
<control name="shadow-ready" code="0x20" label="End of shadow connection update.">
<field name="member-id" type="uint64"/>
<field name="connection-id" type="uint64"/>
+ <field name="user-name" type="str8"/>
</control>
<!-- Complete a cluster state update. -->
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org