Author: gsim
Date: Tue May 8 10:00:00 2007
New Revision: 536253
URL: http://svn.apache.org/viewvc?view=rev&rev=536253
Log:
Initial interop tests for c++.
Added:
incubator/qpid/branches/M2/cpp/tests/BasicP2PTest.h (with props)
incubator/qpid/branches/M2/cpp/tests/BasicPubSubTest.h (with props)
incubator/qpid/branches/M2/cpp/tests/SimpleTestCaseBase.h (with props)
incubator/qpid/branches/M2/cpp/tests/TestCase.h (with props)
incubator/qpid/branches/M2/cpp/tests/TestOptions.h (with props)
incubator/qpid/branches/M2/cpp/tests/interop_runner.cpp (with props)
Modified:
incubator/qpid/branches/M2/cpp/lib/broker/SessionHandlerImpl.cpp
incubator/qpid/branches/M2/cpp/tests/ (props changed)
incubator/qpid/branches/M2/cpp/tests/Makefile.am
Modified: incubator/qpid/branches/M2/cpp/lib/broker/SessionHandlerImpl.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/branches/M2/cpp/lib/broker/SessionHandlerImpl.cpp?view=diff&rev=536253&r1=536252&r2=536253
==============================================================================
--- incubator/qpid/branches/M2/cpp/lib/broker/SessionHandlerImpl.cpp (original)
+++ incubator/qpid/branches/M2/cpp/lib/broker/SessionHandlerImpl.cpp Tue May 8
10:00:00 2007
@@ -233,8 +233,12 @@
}
}
-void SessionHandlerImpl::ChannelHandlerImpl::flow(u_int16_t /*channel*/, bool
/*active*/){}
-void SessionHandlerImpl::ChannelHandlerImpl::flowOk(u_int16_t /*channel*/,
bool /*active*/){}
+void SessionHandlerImpl::ChannelHandlerImpl::flow(u_int16_t /*channel*/, bool
/*active*/){
+
+}
+void SessionHandlerImpl::ChannelHandlerImpl::flowOk(u_int16_t /*channel*/,
bool /*active*/){
+
+}
void SessionHandlerImpl::ChannelHandlerImpl::close(u_int16_t channel,
u_int16_t /*replyCode*/, const string& /*replyText*/,
u_int16_t /*classId*/,
u_int16_t /*methodId*/){
Propchange: incubator/qpid/branches/M2/cpp/tests/
------------------------------------------------------------------------------
--- svn:ignore (original)
+++ svn:ignore Tue May 8 10:00:00 2007
@@ -8,3 +8,4 @@
topic_listener
topic_publisher
qpidd.log
+interop_runner
Added: incubator/qpid/branches/M2/cpp/tests/BasicP2PTest.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/branches/M2/cpp/tests/BasicP2PTest.h?view=auto&rev=536253
==============================================================================
--- incubator/qpid/branches/M2/cpp/tests/BasicP2PTest.h (added)
+++ incubator/qpid/branches/M2/cpp/tests/BasicP2PTest.h Tue May 8 10:00:00 2007
@@ -0,0 +1,84 @@
+#ifndef _BasicP2PTest_
+#define _BasicP2PTest_
+/*
+ *
+ * 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 <memory>
+#include <sstream>
+
+#include <ClientChannel.h>
+#include <ClientMessage.h>
+#include <Connection.h>
+#include <Exception.h>
+#include <MessageListener.h>
+#include "SimpleTestCaseBase.h"
+
+
+namespace qpid {
+
+using namespace qpid::client;
+
+class BasicP2PTest : public SimpleTestCaseBase
+{
+
+ class Receiver : public Worker, public MessageListener
+ {
+ const std::string queue;
+ std::string tag;
+ public:
+ Receiver(TestOptions& options, const std::string& _queue, const int
_messages) : Worker(options, _messages), queue(_queue){}
+
+ void init()
+ {
+ Queue q(queue, true);
+ channel.declareQueue(q);
+ channel.consume(q, tag, this);
+ channel.start();
+ }
+
+ void start(){
+ }
+
+ void received(Message&)
+ {
+ count++;
+ }
+ };
+
+public:
+ void assign(const std::string& role, framing::FieldTable& params,
TestOptions& options)
+ {
+ std::string queue = params.getString("P2P_QUEUE_AND_KEY_NAME");
+ int messages = params.getInt("P2P_NUM_MESSAGES");
+ if (role == "SENDER") {
+ worker = std::auto_ptr<Worker>(new Sender(options,
Exchange::DEFAULT_EXCHANGE, queue, messages));
+ } else if(role == "RECEIVER"){
+ worker = std::auto_ptr<Worker>(new Receiver(options, queue,
messages));
+ } else {
+ throw Exception("unrecognised role");
+ }
+ worker->init();
+ }
+};
+
+}
+
+#endif
Propchange: incubator/qpid/branches/M2/cpp/tests/BasicP2PTest.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/branches/M2/cpp/tests/BasicP2PTest.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: incubator/qpid/branches/M2/cpp/tests/BasicPubSubTest.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/branches/M2/cpp/tests/BasicPubSubTest.h?view=auto&rev=536253
==============================================================================
--- incubator/qpid/branches/M2/cpp/tests/BasicPubSubTest.h (added)
+++ incubator/qpid/branches/M2/cpp/tests/BasicPubSubTest.h Tue May 8 10:00:00
2007
@@ -0,0 +1,144 @@
+#ifndef _BasicPubSubTest_
+#define _BasicPubSubTest_
+/*
+ *
+ * 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 <memory>
+#include <sstream>
+
+#include <ClientChannel.h>
+#include <ClientMessage.h>
+#include <Connection.h>
+#include <Exception.h>
+#include <MessageListener.h>
+#include "SimpleTestCaseBase.h"
+#include <boost/ptr_container/ptr_vector.hpp>
+#include <boost/format.hpp>
+
+
+namespace qpid {
+
+using namespace qpid::client;
+
+class BasicPubSubTest : public SimpleTestCaseBase
+{
+
+ class Receiver : public Worker, public MessageListener
+ {
+ const Exchange& exchange;
+ const std::string queue;
+ const std::string key;
+ std::string tag;
+ public:
+ Receiver(TestOptions& options, const Exchange& _exchange, const
std::string& _queue, const std::string& _key, const int _messages)
+ : Worker(options, _messages), exchange(_exchange), queue(_queue),
key(_key){}
+
+ void init()
+ {
+ Queue q(queue, true);
+ channel.declareQueue(q);
+ framing::FieldTable args;
+ channel.bind(exchange, q, key, args);
+ channel.consume(q, tag, this);
+ channel.start();
+ }
+
+ void start(){
+ }
+
+ void received(Message&)
+ {
+ count++;
+ }
+ };
+
+ class MultiReceiver : public Worker, public MessageListener
+ {
+ typedef boost::ptr_vector<Receiver> ReceiverList;
+ ReceiverList receivers;
+
+ public:
+ MultiReceiver(TestOptions& options, const Exchange& exchange, const
std::string& key, const int _messages, int receiverCount)
+ : Worker(options, _messages)
+ {
+ for (int i = 0; i != receiverCount; i++) {
+ std::string queue = (boost::format("%1%_%2%") %
options.clientid % i).str();
+ receivers.push_back(new Receiver(options, exchange, queue,
key, _messages));
+ }
+ }
+
+ void init()
+ {
+ for (ReceiverList::size_type i = 0; i != receivers.size(); i++) {
+ receivers[i].init();
+ }
+ }
+
+ void start()
+ {
+ for (ReceiverList::size_type i = 0; i != receivers.size(); i++) {
+ receivers[i].start();
+ }
+ }
+
+ void received(Message& msg)
+ {
+ for (ReceiverList::size_type i = 0; i != receivers.size(); i++) {
+ receivers[i].received(msg);
+ }
+ }
+
+ virtual int getCount()
+ {
+ count = 0;
+ for (ReceiverList::size_type i = 0; i != receivers.size(); i++) {
+ count += receivers[i].getCount();
+ }
+ return count;
+ }
+ virtual void stop()
+ {
+ for (ReceiverList::size_type i = 0; i != receivers.size(); i++) {
+ receivers[i].stop();
+ }
+ }
+ };
+
+public:
+ void assign(const std::string& role, framing::FieldTable& params,
TestOptions& options)
+ {
+ std::string key = params.getString("PUBSUB_KEY");
+ int messages = params.getInt("PUBSUB_NUM_MESSAGES");
+ int receivers = params.getInt("PUBSUB_NUM_RECEIVERS");
+ if (role == "SENDER") {
+ worker = std::auto_ptr<Worker>(new Sender(options,
Exchange::STANDARD_TOPIC_EXCHANGE, key, messages));
+ } else if(role == "RECEIVER"){
+ worker = std::auto_ptr<Worker>(new MultiReceiver(options,
Exchange::STANDARD_TOPIC_EXCHANGE, key, messages, receivers));
+ } else {
+ throw Exception("unrecognised role");
+ }
+ worker->init();
+ }
+};
+
+}
+
+#endif
Propchange: incubator/qpid/branches/M2/cpp/tests/BasicPubSubTest.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/branches/M2/cpp/tests/BasicPubSubTest.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: incubator/qpid/branches/M2/cpp/tests/Makefile.am
URL:
http://svn.apache.org/viewvc/incubator/qpid/branches/M2/cpp/tests/Makefile.am?view=diff&rev=536253&r1=536252&r2=536253
==============================================================================
--- incubator/qpid/branches/M2/cpp/tests/Makefile.am (original)
+++ incubator/qpid/branches/M2/cpp/tests/Makefile.am Tue May 8 10:00:00 2007
@@ -23,7 +23,8 @@
client_test \
echo_service \
topic_listener \
- topic_publisher
+ topic_publisher \
+ interop_runner
broker_tests = \
AccumulatedAckTest \
Added: incubator/qpid/branches/M2/cpp/tests/SimpleTestCaseBase.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/branches/M2/cpp/tests/SimpleTestCaseBase.h?view=auto&rev=536253
==============================================================================
--- incubator/qpid/branches/M2/cpp/tests/SimpleTestCaseBase.h (added)
+++ incubator/qpid/branches/M2/cpp/tests/SimpleTestCaseBase.h Tue May 8
10:00:00 2007
@@ -0,0 +1,137 @@
+#ifndef _SimpleTestCaseBase_
+#define _SimpleTestCaseBase_
+/*
+ *
+ * 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 <memory>
+#include <sstream>
+
+#include <ClientChannel.h>
+#include <ClientMessage.h>
+#include <Connection.h>
+#include <Exception.h>
+#include <MessageListener.h>
+#include "TestCase.h"
+
+
+namespace qpid {
+
+using namespace qpid::client;
+
+class SimpleTestCaseBase : public TestCase
+{
+protected:
+ class Worker
+ {
+ protected:
+ client::Connection connection;
+ client::Channel channel;
+ const int messages;
+ int count;
+
+ public:
+
+ Worker(TestOptions& options, const int _messages) :
+ connection(options.trace), messages(_messages), count(0)
+ {
+ connection.open(options.broker, options.port);
+ connection.openChannel(&channel);
+ }
+
+ virtual ~Worker(){}
+
+ virtual void stop()
+ {
+ channel.close();
+ connection.close();
+ }
+
+ virtual int getCount()
+ {
+ return count;
+ }
+
+ virtual void init() = 0;
+ virtual void start() = 0;
+ };
+
+ class Sender : public Worker
+ {
+ const Exchange& exchange;
+ const std::string key;
+ public:
+ Sender(TestOptions& options,
+ const Exchange& _exchange,
+ const std::string& _key,
+ const int _messages)
+ : Worker(options, _messages), exchange(_exchange), key(_key) {}
+
+ void init()
+ {
+ channel.start();
+ }
+
+ void start(){
+ Message msg;
+ while (count < messages) {
+ channel.publish(msg, exchange, key);
+ count++;
+ }
+ stop();
+ }
+ };
+
+ std::auto_ptr<Worker> worker;
+
+public:
+ virtual void assign(const std::string& role, framing::FieldTable& params,
TestOptions& options) = 0;
+
+ virtual ~SimpleTestCaseBase() {}
+
+ void start()
+ {
+ if (worker.get()) {
+ worker->start();
+ }
+ }
+
+ void stop()
+ {
+ if (worker.get()) {
+ worker->stop();
+ }
+ }
+
+ void report(client::Message& report)
+ {
+ if (worker.get()) {
+ report.getHeaders().setInt("MESSAGE_COUNT", worker->getCount());
+ //add number of messages sent or received
+ std::stringstream reportstr;
+ reportstr << worker->getCount();
+ report.setData(reportstr.str());
+ }
+ }
+};
+
+}
+
+#endif
Propchange: incubator/qpid/branches/M2/cpp/tests/SimpleTestCaseBase.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/branches/M2/cpp/tests/SimpleTestCaseBase.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: incubator/qpid/branches/M2/cpp/tests/TestCase.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/branches/M2/cpp/tests/TestCase.h?view=auto&rev=536253
==============================================================================
--- incubator/qpid/branches/M2/cpp/tests/TestCase.h (added)
+++ incubator/qpid/branches/M2/cpp/tests/TestCase.h Tue May 8 10:00:00 2007
@@ -0,0 +1,64 @@
+#ifndef _TestCase_
+#define _TestCase_
+/*
+ *
+ * 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 <ClientMessage.h>
+#include "TestOptions.h"
+
+
+namespace qpid {
+
+/**
+ * Interface to be implemented by test cases for use with the test
+ * runner.
+ */
+class TestCase
+{
+public:
+ /**
+ * Directs the test case to act in a particular role. Some roles
+ * may be 'activated' at this stage others may require an explicit
+ * start request.
+ */
+ virtual void assign(const std::string& role, framing::FieldTable& params,
TestOptions& options) = 0;
+ /**
+ * Each test will be started on its own thread, which should block
+ * until the test completes (this may or may not require an
+ * explicit stop() request).
+ */
+ virtual void start() = 0;
+ /**
+ * Requests that the test be stopped if still running.
+ */
+ virtual void stop() = 0;
+ /**
+ * Allows the test to fill in details on the final report
+ * message. Will be called only after start has returned.
+ */
+ virtual void report(client::Message& report) = 0;
+
+ virtual ~TestCase() {}
+};
+
+}
+
+#endif
Propchange: incubator/qpid/branches/M2/cpp/tests/TestCase.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/branches/M2/cpp/tests/TestCase.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: incubator/qpid/branches/M2/cpp/tests/TestOptions.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/branches/M2/cpp/tests/TestOptions.h?view=auto&rev=536253
==============================================================================
--- incubator/qpid/branches/M2/cpp/tests/TestOptions.h (added)
+++ incubator/qpid/branches/M2/cpp/tests/TestOptions.h Tue May 8 10:00:00 2007
@@ -0,0 +1,69 @@
+#ifndef _TestOptions_
+#define _TestOptions_
+/*
+ *
+ * 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 <CommonOptions.h>
+
+namespace qpid {
+
+struct TestOptions : public qpid::CommonOptions
+{
+ TestOptions() : desc("Options"), broker("localhost"), virtualhost(""),
clientid("cpp"), help(false)
+ {
+ using namespace qpid::program_options;
+ using namespace boost::program_options;
+ CommonOptions::addTo(desc);
+ desc.add_options()
+ ("broker,b", optValue(broker, "HOSTNAME"), "the hostname to
connect to")
+ ("virtualhost,v", optValue(virtualhost, "VIRTUAL_HOST"), "virtual
host")
+ ("clientid,i", optValue(clientid, "ID"), "unique client
identifier")
+ ("help,h", optValue(help), "print this usage statement");
+ }
+
+ void parse(int argc, char** argv)
+ {
+ using namespace boost::program_options;
+ try {
+ variables_map vm;
+ store(parse_command_line(argc, argv, desc), vm);
+ notify(vm);
+ } catch(const error& e) {
+ std::cerr << "Error: " << e.what() << std::endl
+ << "Specify '--help' for usage." << std::endl;
+ }
+ }
+
+ void usage()
+ {
+ std::cout << desc << std::endl;
+ }
+
+ boost::program_options::options_description desc;
+ std::string broker;
+ std::string virtualhost;
+ std::string clientid;
+ bool help;
+};
+
+}
+
+#endif
Propchange: incubator/qpid/branches/M2/cpp/tests/TestOptions.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/branches/M2/cpp/tests/TestOptions.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: incubator/qpid/branches/M2/cpp/tests/interop_runner.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/branches/M2/cpp/tests/interop_runner.cpp?view=auto&rev=536253
==============================================================================
--- incubator/qpid/branches/M2/cpp/tests/interop_runner.cpp (added)
+++ incubator/qpid/branches/M2/cpp/tests/interop_runner.cpp Tue May 8 10:00:00
2007
@@ -0,0 +1,252 @@
+/*
+ *
+ * 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 <CommonOptions.h>
+#include <Connection.h>
+#include <ClientChannel.h>
+#include <ClientExchange.h>
+#include <ClientQueue.h>
+#include <Exception.h>
+#include <MessageListener.h>
+#include <QpidError.h>
+#include <sys/Thread.h>
+#include <sys/Time.h>
+#include <iostream>
+#include <memory>
+#include "BasicP2PTest.h"
+#include "BasicPubSubTest.h"
+#include "TestCase.h"
+#include <boost/ptr_container/ptr_map.hpp>
+
+/**
+ * Framework for interop tests.
+ *
+ * [see
http://cwiki.apache.org/confluence/display/qpid/Interop+Testing+Specification
for details].
+ */
+
+using namespace qpid::client;
+using namespace qpid::sys;
+using qpid::TestCase;
+using qpid::TestOptions;
+using qpid::framing::FieldTable;
+using std::string;
+
+class DummyRun : public TestCase
+{
+public:
+ DummyRun() {}
+ void assign(const std::string&, FieldTable&, TestOptions&) {}
+ void start() {}
+ void stop() {}
+ void report(qpid::client::Message&) {}
+};
+
+string parse_next_word(const string& input, const string& delims,
string::size_type& position);
+
+/**
+ */
+class Listener : public MessageListener, private Runnable{
+ typedef boost::ptr_map<std::string, TestCase> TestMap;
+
+ Channel& channel;
+ TestOptions& options;
+ TestMap tests;
+ const string name;
+ const string topic;
+ TestMap::iterator test;
+ std::auto_ptr<Thread> runner;
+ string reportTo;
+ string reportCorrelator;
+
+ void shutdown();
+ bool invite(const string& name);
+ void run();
+
+ void sendResponse(Message& response, string replyTo);
+ void sendResponse(Message& response, Message& request);
+ void sendSimpleResponse(const string& type, Message& request);
+ void sendReport();
+public:
+ Listener(Channel& channel, TestOptions& options);
+ void received(Message& msg);
+ void bindAndConsume();
+ void registerTest(std::string name, TestCase* test);
+};
+
+/**
+ */
+int main(int argc, char** argv){
+ TestOptions options;
+ options.parse(argc, argv);
+
+ if (options.help) {
+ options.usage();
+ } else {
+ try{
+ Connection connection(options.trace);
+ connection.open(options.broker, options.port, "guest", "guest",
options.virtualhost);
+
+ Channel channel;
+ connection.openChannel(&channel);
+
+ Listener listener(channel, options);
+ listener.registerTest("TC1_DummyRun", new DummyRun());
+ listener.registerTest("TC2_BasicP2P", new qpid::BasicP2PTest());
+ listener.registerTest("TC3_BasicPubSub", new
qpid::BasicPubSubTest());
+
+ listener.bindAndConsume();
+
+ channel.run();
+ connection.close();
+ } catch(qpid::Exception error) {
+ std::cout << error.what() << std::endl;
+ }
+ }
+}
+
+Listener::Listener(Channel& _channel, TestOptions& _options) :
channel(_channel), options(_options), name(options.clientid),
topic("iop.control." + name)
+{}
+
+void Listener::registerTest(std::string name, TestCase* test)
+{
+ tests.insert(name, test);
+}
+
+void Listener::bindAndConsume()
+{
+ Queue control(name, true);
+ channel.declareQueue(control);
+ qpid::framing::FieldTable bindArgs;
+ //replace these separate binds with a wildcard once that is supported on
java broker
+ channel.bind(Exchange::STANDARD_TOPIC_EXCHANGE, control, "iop.control",
bindArgs);
+ channel.bind(Exchange::STANDARD_TOPIC_EXCHANGE, control, topic, bindArgs);
+
+ std::string tag;
+ channel.consume(control, tag, this);
+}
+
+void Listener::sendSimpleResponse(const string& type, Message& request)
+{
+ Message response;
+ response.getHeaders().setString("CONTROL_TYPE", type);
+ response.getHeaders().setString("CLIENT_NAME", name);
+ response.getHeaders().setString("CLIENT_PRIVATE_CONTROL_KEY", topic);
+ response.setCorrelationId(request.getCorrelationId());
+ sendResponse(response, request);
+}
+
+void Listener::sendResponse(Message& response, Message& request)
+{
+ sendResponse(response, request.getReplyTo());
+}
+
+void Listener::sendResponse(Message& response, string replyTo)
+{
+ //Exchange and routing key need to be extracted from the reply-to
+ //field. Format is assumed to be:
+ //
+ // <exchange type>://<exchange name>/<routing key>?<options>
+ //
+ //and all we need is the exchange name and routing key
+ //
+ if (replyTo.empty()) throw qpid::Exception("Reply address not set!");
+ const string delims(":/?=");
+
+ string::size_type start = replyTo.find(':');//skip exchange type
+ string exchange = parse_next_word(replyTo, delims, start);
+ string routingKey = parse_next_word(replyTo, delims, start);
+ channel.publish(response, exchange, routingKey);
+}
+
+void Listener::received(Message& message)
+{
+ std::string type(message.getHeaders().getString("CONTROL_TYPE"));
+
+ if (type == "INVITE") {
+ std::string name(message.getHeaders().getString("TEST_NAME"));
+ if (name.empty() || invite(name)) {
+ sendSimpleResponse("ENLIST", message);
+ } else {
+ std::cout << "Can't take part in '" << name << "'" << std::endl;
+ }
+ } else if (type == "ASSIGN_ROLE") {
+ test->assign(message.getHeaders().getString("ROLE"),
message.getHeaders(), options);
+ sendSimpleResponse("ACCEPT_ROLE", message);
+ } else if (type == "START") {
+ reportTo = message.getReplyTo();
+ reportCorrelator = message.getCorrelationId();
+ runner = std::auto_ptr<Thread>(new Thread(this));
+ } else if (type == "STATUS_REQUEST") {
+ reportTo = message.getReplyTo();
+ reportCorrelator = message.getCorrelationId();
+ test->stop();
+ sendReport();
+ } else if (type == "TERMINATE") {
+ if (test != tests.end()) test->stop();
+ shutdown();
+ } else {
+ std::cerr <<"ERROR!: Received unknown control message: " << type <<
std::endl;
+ shutdown();
+ }
+}
+
+void Listener::shutdown()
+{
+ channel.close();
+}
+
+bool Listener::invite(const string& name)
+{
+ test = tests.find(name);
+ return test != tests.end();
+}
+
+void Listener::run()
+{
+ //NB: this method will be called in its own thread
+ //start test and when start returns...
+ test->start();
+ sendReport();
+}
+
+void Listener::sendReport()
+{
+ Message report;
+ report.getHeaders().setString("CONTROL_TYPE", "REPORT");
+ test->report(report);
+ report.setCorrelationId(reportCorrelator);
+ sendResponse(report, reportTo);
+}
+
+string parse_next_word(const string& input, const string& delims,
string::size_type& position)
+{
+ string::size_type start = input.find_first_not_of(delims, position);
+ if (start == string::npos) {
+ return "";
+ } else {
+ string::size_type end = input.find_first_of(delims, start);
+ if (end == string::npos) {
+ end = input.length();
+ }
+ position = end;
+ return input.substr(start, end - start);
+ }
+}
Propchange: incubator/qpid/branches/M2/cpp/tests/interop_runner.cpp
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/branches/M2/cpp/tests/interop_runner.cpp
------------------------------------------------------------------------------
svn:keywords = Rev Date
