Author: aconway
Date: Tue Jul 24 12:39:27 2007
New Revision: 559171
URL: http://svn.apache.org/viewvc?view=rev&rev=559171
Log:
* Summary:
- Wiring (declare/delete/bind) is replicated via AIS.
- TestOptions includes all logging options.
- Logger automatically parses env vars so logging can be enabled
for any program linked with libqpidcommon e.g. by setting QPID_TRACE=1.
* src/qpid/cluster/SessionManager.cpp: Handle frames from cluster
- Forward to BrokerAdapter for execution.
- Suppress responses in proxy.
* src/tests/TestOptions.h (Options): Logging options, --help option.
* src/qpid/client/ClientConnection.cpp: Removed log initialization.
Logs are initialized either in TestOptions or automatically from env
vars,
e.g. QPID_TRACE,
* src/qpid/QpidError.h (class QpidError): Initialize Exception in
constructor so messages can be logged.
* src/qpid/framing/ChannelAdapter.h: Made send() virtual.
* src/tests/Cluster_child.cpp: UUID corrected.
* src/qpid/broker/Broker.cpp: Pass chains to updater by ref.
* src/qpid/Options.cpp (parse): Fix log settings from environment.
Modified:
incubator/qpid/trunk/qpid/cpp/src/qpid/Options.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/QpidError.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/QpidError.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/client/ClientConnection.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ClassifierHandler.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/SessionManager.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/SessionManager.h
incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ChannelAdapter.h
incubator/qpid/trunk/qpid/cpp/src/tests/Cluster.h
incubator/qpid/trunk/qpid/cpp/src/tests/Cluster_child.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/TestOptions.h
incubator/qpid/trunk/qpid/cpp/src/tests/cluster_client.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/start_cluster
incubator/qpid/trunk/qpid/cpp/src/tests/stop_cluster
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/Options.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/Options.cpp?view=diff&rev=559171&r1=559170&r2=559171
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/Options.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/Options.cpp Tue Jul 24 12:39:27 2007
@@ -18,6 +18,9 @@
#include "Options.h"
#include "qpid/Exception.h"
+
+#include <boost/bind.hpp>
+
#include <fstream>
#include <algorithm>
#include <iostream>
@@ -28,18 +31,26 @@
namespace {
-char env2optchar(char env) { return (env=='_') ? '-' : tolower(env); }
+struct EnvOptMapper {
+ static bool matchChar(char env, char opt) {
+ return (env==toupper(opt)) || (strchr("-.", opt) && env=='_');
+ }
-struct Mapper {
- Mapper(const Options& o) : opts(o) {}
- string operator()(const string& env) {
+ static bool matchStr(const string& env,
boost::shared_ptr<po::option_description> desc) {
+ return std::equal(env.begin(), env.end(), desc->long_name().begin(),
&matchChar);
+ }
+
+ EnvOptMapper(const Options& o) : opts(o) {}
+
+ string operator()(const string& envVar) {
static const std::string prefix("QPID_");
- if (env.substr(0, prefix.size()) == prefix) {
- string opt = env.substr(prefix.size());
- transform(opt.begin(), opt.end(), opt.begin(), env2optchar);
- // Ignore env vars that don't match to known options.
- if (opts.find_nothrow(opt, false))
- return opt;
+ if (envVar.substr(0, prefix.size()) == prefix) {
+ string env = envVar.substr(prefix.size());
+ typedef const std::vector<
boost::shared_ptr<po::option_description> > OptDescs;
+ OptDescs::const_iterator i =
+ find_if(opts.options().begin(), opts.options().end(),
boost::bind(matchStr, env, _1));
+ if (i != opts.options().end())
+ return (*i)->long_name();
}
return string();
}
@@ -62,7 +73,7 @@
if (argc > 0 && argv != 0)
po::store(po::parse_command_line(argc, argv, *this), vm);
parsing="environment variables";
- po::store(po::parse_environment(*this, Mapper(*this)), vm);
+ po::store(po::parse_environment(*this, EnvOptMapper(*this)), vm);
po::notify(vm); // configFile may be updated from arg/env options.
if (!configFile.empty()) {
parsing="configuration file "+configFile;
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/QpidError.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/QpidError.cpp?view=diff&rev=559171&r1=559170&r2=559171
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/QpidError.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/QpidError.cpp Tue Jul 24 12:39:27
2007
@@ -34,9 +34,8 @@
void QpidError::throwSelf() const { throw *this; }
-void QpidError::init() {
- whatStr = boost::str(boost::format("Error [%d] %s (%s:%d)")
- % code % msg % loc.file % loc.line);
+std::string QpidError::message(int code, const std::string& msg, const char*
file, int line) {
+ return (boost::format("Error [%d] %s (%s:%d)") % code % msg % file %
line).str();
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/QpidError.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/QpidError.h?view=diff&rev=559171&r1=559170&r2=559171
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/QpidError.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/QpidError.h Tue Jul 24 12:39:27 2007
@@ -48,16 +48,15 @@
template <class T>
QpidError(int code_, const T& msg_, const SrcLine& loc_) throw()
- : code(code_), loc(loc_), msg(boost::lexical_cast<std::string>(msg_))
- { init(); }
+ : Exception(message(code_, boost::lexical_cast<std::string>(msg_),
loc_.file.c_str(), loc_.line)),
+ code(code_), loc(loc_), msg(boost::lexical_cast<std::string>(msg_))
{}
~QpidError() throw();
Exception::auto_ptr clone() const throw();
void throwSelf() const;
- private:
-
- void init();
+ /** Format message for exception. */
+ static std::string message(int code, const std::string& msg, const char*
file, int line);
};
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp?view=diff&rev=559171&r1=559170&r2=559171
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp Tue Jul 24
12:39:27 2007
@@ -164,7 +164,7 @@
void Broker::update(FrameHandler::Chains& chains) {
for_each(handlerUpdaters.begin(), handlerUpdaters.end(),
- boost::bind(&HandlerUpdater::update, _1, chains));
+ boost::bind(&HandlerUpdater::update, _1, boost::ref(chains)));
}
}} // namespace qpid::broker
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/ClientConnection.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/ClientConnection.cpp?view=diff&rev=559171&r1=559170&r2=559171
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/ClientConnection.cpp
(original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/ClientConnection.cpp Tue Jul
24 12:39:27 2007
@@ -51,9 +51,6 @@
isOpen(false), debug(_debug)
{
setConnector(defaultConnector);
- qpid::log::Options o;
- o.trace = debug;
- qpid::log::Logger::instance().configure(o, "qpid-c++-client");
}
Connection::~Connection(){}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ClassifierHandler.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ClassifierHandler.cpp?view=diff&rev=559171&r1=559170&r2=559171
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ClassifierHandler.cpp
(original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ClassifierHandler.cpp Tue
Jul 24 12:39:27 2007
@@ -61,6 +61,9 @@
Chain chosen;
shared_ptr<AMQMethodBody> method =
dynamic_pointer_cast<AMQMethodBody>(frame.getBody());
+ // FIXME aconway 2007-07-05: Need to stop bypassed frames
+ // from overtaking mcast frames.
+ //
if (method)
chosen=map[fullId(method)];
if (chosen)
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp?view=diff&rev=559171&r1=559170&r2=559171
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp Tue Jul 24
12:39:27 2007
@@ -50,9 +50,10 @@
// Only provide to a Broker, and only if the --cluster config is set.
if (broker && !options.clusterName.empty()) {
assert(!cluster); // A process can only belong to one cluster.
- sessions.reset(new SessionManager());
+
+ sessions.reset(new SessionManager(*broker));
cluster.reset(new Cluster(options.clusterName, broker->getUrl(),
sessions));
- sessions->setClusterSend(cluster); // FIXME aconway 2007-07-10:
+ sessions->setClusterSend(cluster);
broker->add(sessions);
}
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/SessionManager.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/SessionManager.cpp?view=diff&rev=559171&r1=559170&r2=559171
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/SessionManager.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/SessionManager.cpp Tue Jul
24 12:39:27 2007
@@ -16,17 +16,59 @@
*
*/
+#include "SessionManager.h"
+#include "ClassifierHandler.h"
+
#include "qpid/log/Statement.h"
#include "qpid/framing/amqp_types.h"
#include "qpid/framing/AMQFrame.h"
-#include "SessionManager.h"
-#include "ClassifierHandler.h"
+#include "qpid/framing/AMQP_ServerOperations.h"
+#include "qpid/broker/BrokerAdapter.h"
+#include "qpid/broker/Connection.h"
+#include "qpid/broker/BrokerChannel.h"
+#include "qpid/framing/ChannelAdapter.h"
namespace qpid {
namespace cluster {
using namespace framing;
using namespace sys;
+using namespace broker;
+
+/** Handler to send frames direct to local broker (bypass correlation etc.) */
+struct BrokerHandler : public FrameHandler, private ChannelAdapter {
+ Connection connection;
+ Channel channel;
+ BrokerAdapter adapter;
+
+ // TODO aconway 2007-07-23: Lots of needless flab here (Channel,
+ // Connection, ChannelAdapter) As these classes are untangled the
+ // flab can be reduced. The real requirements are:
+ // - Dispatch methods direct to broker bypassing all the correlation muck
+ // - Efficiently suppress responses
+ // For the latter we are now using a ChannelAdapter with noop send()
+ // A more efficient solution would be a no-op proxy.
+ //
+ BrokerHandler(Broker& broker) :
+ connection(0, broker),
+ channel(connection, 1, 0),
+ adapter(channel, connection, broker, *this) {}
+
+ void handle(AMQFrame& frame) {
+ AMQMethodBody* body=dynamic_cast<AMQMethodBody*>(frame.body.get());
+ assert(body);
+ body->invoke(adapter, MethodContext()); // TODO aconway 2007-07-24:
Remove MethodContext
+ }
+
+ // Dummy methods.
+ virtual void handleHeader(boost::shared_ptr<AMQHeaderBody>){}
+ virtual void handleContent(boost::shared_ptr<AMQContentBody>){}
+ virtual void handleHeartbeat(boost::shared_ptr<AMQHeartbeatBody>){}
+ virtual bool isOpen() const{ return true; }
+ virtual void handleMethodInContext(shared_ptr<AMQMethodBody>, const
MethodContext&){}
+ // No-op send.
+ virtual RequestId send(shared_ptr<AMQBody>, Correlator::Action) { return
0; }
+};
/** Wrap plain AMQFrames in SessionFrames */
struct FrameWrapperHandler : public FrameHandler {
@@ -47,17 +89,15 @@
SessionFrameHandler::Chain next;
};
-SessionManager::SessionManager() {}
+SessionManager::SessionManager(Broker& b) : localBroker(new BrokerHandler(b))
{}
-void SessionManager::update(FrameHandler::Chains& chains)
-{
+void SessionManager::update(FrameHandler::Chains& chains) {
Mutex::ScopedLock l(lock);
// Create a new local session, store local chains.
Uuid uuid(true);
sessions[uuid] = chains;
- // Replace local incoming chain. Build from the back.
- //
+ // Replace local in chain. Build from the back.
// TODO aconway 2007-07-05: Currently mcast wiring, bypass
// everythign else.
assert(clusterSend);
@@ -65,39 +105,26 @@
FrameHandler::Chain classify(new ClassifierHandler(wiring, chains.in));
chains.in = classify;
- // FIXME aconway 2007-07-05: Need to stop bypassed frames
- // from overtaking mcast frames.
- //
-
- // Leave outgoing chain unmodified.
+ // Leave out chain unmodified.
// TODO aconway 2007-07-05: Failover will require replication of
// outgoing frames to session replicas.
-
}
void SessionManager::handle(SessionFrame& frame) {
- // Incoming from frame.
- FrameHandler::Chains chains;
+ // Incoming from cluster.
{
Mutex::ScopedLock l(lock);
+ assert(frame.isIncoming); // FIXME aconway 2007-07-24: Drop isIncoming?
SessionMap::iterator i = sessions.find(frame.uuid);
if (i == sessions.end()) {
- QPID_LOG(trace, "Non-local frame cluster: " << frame.frame);
- chains = nonLocal;
+ // Non local method frame, invoke.
+ localBroker->handle(frame.frame);
}
else {
- QPID_LOG(trace, "Local frame from cluster: " << frame.frame);
- chains = i->second;
+ // Local frame, continue on local chain
+ i->second.in->handle(frame.frame);
}
}
- FrameHandler::Chain chain =
- chain = frame.isIncoming ? chains.in : chains.out;
- // TODO aconway 2007-07-11: Should this be assert(chain)
- if (chain)
- chain->handle(frame.frame);
-
- // TODO aconway 2007-07-05: Here's where we should unblock frame
- // dispatch for the channel.
}
}} // namespace qpid::cluster
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/SessionManager.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/SessionManager.h?view=diff&rev=559171&r1=559170&r2=559171
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/SessionManager.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/SessionManager.h Tue Jul 24
12:39:27 2007
@@ -19,25 +19,33 @@
*
*/
-#include "qpid/broker/BrokerChannel.h"
#include "qpid/cluster/SessionFrame.h"
#include "qpid/framing/HandlerUpdater.h"
+#include "qpid/framing/FrameHandler.h"
#include "qpid/framing/Uuid.h"
#include "qpid/sys/Mutex.h"
+#include <boost/noncopyable.hpp>
+
#include <map>
namespace qpid {
+
+namespace broker {
+class Broker;
+}
+
namespace cluster {
/**
* Manage sessions and handler chains for the cluster.
*
*/
-class SessionManager : public framing::HandlerUpdater, public
SessionFrameHandler
+class SessionManager : public framing::HandlerUpdater, public
SessionFrameHandler,
+ private boost::noncopyable
{
public:
- SessionManager();
+ SessionManager(broker::Broker& broker);
/** Set the handler to send to the cluster */
void setClusterSend(const SessionFrameHandler::Chain& send) {
clusterSend=send; }
@@ -52,12 +60,13 @@
framing::ChannelId getChannelId(const framing::Uuid&) const;
private:
+ class SessionOperations;
typedef std::map<framing::Uuid,framing::FrameHandler::Chains> SessionMap;
sys::Mutex lock;
SessionFrameHandler::Chain clusterSend;
+ framing::FrameHandler::Chain localBroker;
SessionMap sessions;
- framing::FrameHandler::Chains nonLocal;
};
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ChannelAdapter.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ChannelAdapter.h?view=diff&rev=559171&r1=559170&r2=559171
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ChannelAdapter.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ChannelAdapter.h Tue Jul 24
12:39:27 2007
@@ -75,8 +75,8 @@
*response to this frame. Ignored if body is not a Request.
[EMAIL PROTECTED] If body is a request, the ID assigned else 0.
*/
- RequestId send(shared_ptr<AMQBody> body,
- Correlator::Action action=Correlator::Action());
+ virtual RequestId send(shared_ptr<AMQBody> body,
+ Correlator::Action action=Correlator::Action());
virtual bool isOpen() const = 0;
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/Cluster.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/Cluster.h?view=diff&rev=559171&r1=559170&r2=559171
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/Cluster.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/Cluster.h Tue Jul 24 12:39:27 2007
@@ -64,9 +64,9 @@
Mutex::ScopedLock l(lock);
BOOST_MESSAGE(getpid()<<" TestHandler::waitFor("<<n<<")
"<<this->size());
AbsTime deadline(now(), 2*TIME_SEC);
- while (vector<T>::size() < n && lock.wait(deadline))
+ while (this->size() < n && lock.wait(deadline))
;
- return vector<T>::size() >= n;
+ return this->size() >= n;
}
};
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/Cluster_child.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/Cluster_child.cpp?view=diff&rev=559171&r1=559170&r2=559171
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/Cluster_child.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/Cluster_child.cpp Tue Jul 24
12:39:27 2007
@@ -32,15 +32,14 @@
/** Chlid part of Cluster::clusterTwo test */
void clusterTwo() {
- TestCluster cluster("clusterTwo", "amqp::2");
+ TestCluster cluster("clusterTwo", "amqp:child:2");
BOOST_REQUIRE(cluster.received.waitFor(1)); // Frame from parent.
BOOST_CHECK(cluster.received[0].isIncoming);
BOOST_CHECK_TYPEID_EQUAL(ChannelPingBody,
*cluster.received[0].frame.getBody());
BOOST_CHECK_EQUAL(2u, cluster.size()); // Me and parent
AMQFrame frame(VER, 1, new ChannelOkBody(VER));
- Uuid id(true);
- SessionFrame sf(id, frame, false);
+ SessionFrame sf(cluster.received[0].uuid, frame, false);
cluster.handle(sf);
BOOST_REQUIRE(cluster.received.waitFor(2));
BOOST_CHECK(!cluster.received[1].isIncoming);
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/TestOptions.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/TestOptions.h?view=diff&rev=559171&r1=559170&r2=559171
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/TestOptions.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/TestOptions.h Tue Jul 24 12:39:27
2007
@@ -22,13 +22,18 @@
*/
#include "qpid/Options.h"
+#include "qpid/log/Options.h"
#include "qpid/Url.h"
+#include "qpid/log/Logger.h"
+
+#include <iostream>
+#include <exception>
namespace qpid {
struct TestOptions : public qpid::Options
{
- TestOptions() : Options("Test Options"), host("localhost"),
port(TcpAddress::DEFAULT_PORT), clientid("cpp"), trace(false), help(false)
+ TestOptions() : Options("Test Options"), host("localhost"),
port(TcpAddress::DEFAULT_PORT), clientid("cpp"), help(false)
{
addOptions()
("host,h", optValue(host, "HOST"), "Broker host to connect to")
@@ -39,10 +44,26 @@
("clientname,n", optValue(clientid, "ID"), "unique client
identifier")
("username", optValue(username, "USER"), "user name for broker log
in.")
("password", optValue(password, "USER"), "password for broker log
in.")
- ("trace,t", optValue(trace), "Turn on debug tracing.")
("help", optValue(help), "print this usage statement");
+ add(log);
}
+ /** As well as parsing, print help & exit if required */
+ void parse(int argc, char** argv) {
+ try {
+ qpid::Options::parse(argc, argv);
+ } catch (const std::exception& e) {
+ std::cout << e.what() << std::endl << *this << std::endl;
+ exit(1);
+ }
+ if (help) {
+ std::cout << *this << std::endl;
+ exit(0);
+ }
+ trace = log.trace;
+ qpid::log::Logger::instance().configure(log, argv[0]);
+ }
+
std::string host;
uint16_t port;
std::string virtualhost;
@@ -51,6 +72,7 @@
std::string password;
bool trace;
bool help;
+ log::Options log;
};
}
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/cluster_client.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/cluster_client.cpp?view=diff&rev=559171&r1=559170&r2=559171
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/cluster_client.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/cluster_client.cpp Tue Jul 24
12:39:27 2007
@@ -52,6 +52,7 @@
};
BOOST_AUTO_TEST_CASE(testWiringReplication) {
+ // Declare on one broker, use on others.
ClusterConnections cluster;
BOOST_REQUIRE(cluster.size() > 1);
@@ -63,13 +64,17 @@
broker0.declareExchange(fooEx);
broker0.declareQueue(fooQ);
broker0.bind(fooEx, fooQ, "FooKey");
-
- Channel broker1;
- cluster[1]->openChannel(broker1);
- broker1.publish(Message("hello"), fooEx, "FooKey");
- Message m;
- BOOST_REQUIRE(broker1.get(m, fooQ));
- BOOST_REQUIRE_EQUAL(m.getData(), "hello");
+ broker0.close();
+
+ for (size_t i = 1; i < cluster.size(); ++i) {
+ Channel ch;
+ cluster[i]->openChannel(ch);
+ ch.publish(Message("hello"), fooEx, "FooKey");
+ Message m;
+ BOOST_REQUIRE(ch.get(m, fooQ));
+ BOOST_REQUIRE_EQUAL(m.getData(), "hello");
+ ch.close();
+ }
}
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/start_cluster
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/start_cluster?view=diff&rev=559171&r1=559170&r2=559171
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/start_cluster (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/start_cluster Tue Jul 24 12:39:27
2007
@@ -6,14 +6,14 @@
test -f cluster.ports && { echo "cluster.ports file already exists" ; exit 1; }
test -z "$*" && { echo "Usage: $0 cluster-size [options]"; exit 1; }
+rm -f cluster*.log cluster.ports
SIZE=$1
shift
OPTS=$*
-
+CLUSTER=`whoami` # Cluster name=user name, avoid clashes.
for (( i=0; i<SIZE; ++i )); do
- PORT=`../qpidd -dp0 --log.output=broker$i.log $OPTS` || exit 1
- PORTS="$PORT $PORTS"
+ PORT=`../qpidd -dp0 --log.output=cluster$i.log --cluster $CLUSTER $OPTS`
|| exit 1
+ echo $PORT >> cluster.ports
done
-echo $PORTS > cluster.ports
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/stop_cluster
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/stop_cluster?view=diff&rev=559171&r1=559170&r2=559171
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/stop_cluster (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/stop_cluster Tue Jul 24 12:39:27
2007
@@ -6,11 +6,9 @@
for PORT in $PORTS ; do
../qpidd -qp $PORT || ERROR="$ERROR $PORT"
done
+rm -f cluster.ports
if [ -n "$ERROR" ]; then
echo "Errors stopping brokers on ports: $ERROR"
- echo $ERROR > cluster.ports
exit 1
-else
- rm cluster.ports
fi
