2006-12-01 Jim Meyering <meyering@redhat.com>

This delta imposes two major changes on the C++ hierarchy:
  - adds autoconf, automake, libtool support
  - makes the hierarchy flatter and renames a few files (e.g., Queue.h,
  Queue.cpp) that appeared twice, once under client/ and again under broker/.

In the process, I've changed many #include directives, mostly
to remove a qpid/ or qpid/framing/ prefix from the file name argument.
Although most changes were to .cpp and .h files under qpid/cpp/, there
were also several to template files under qpid/gentools, and even one
to CppGenerator.java.

Nearly all files are moved to a new position in the hierarchy.
The new hierarchy looks like this:

  src               # this is the new home of qpidd.cpp
  tests             # all tests are here.  See Makefile.am.
  gen               # As before, all generated files go here.
  lib               # This is just a container for the 3 lib dirs:
  lib/client
  lib/broker
  lib/common
  lib/common/framing
  lib/common/sys
  lib/common/sys/posix
  lib/common/sys/apr
  build-aux
  m4


git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@481159 13f79535-47bb-0310-9956-ffa450edef68

5 files changed, 0 insertions, 797 deletions

diff --git a/cpp/test/client/Makefile.cppclient.examples b/cpp/test/client/Makefile.cppclient.examples
deleted file mode 100644
index a0ebb89ac5..0000000000
--- a/cpp/test/client/Makefile.cppclient.examples
+++ /dev/null
@@ -1,56 +0,0 @@
-#
-# This Makefile is intended to work in the RHEL3 release structure.
-# It is a simple example to get users up and running.
-#
-
-CXXFLAGS := -DNDEBUG -DUSE_APR -MMD -fpic
-
-#
-# Configure Boost.
-#
-BOOST_LOCATION := ../lib/boost-1.33.1
-BOOST_CFLAGS := -I$(BOOST_LOCATION)/include/boost-1_33_1
-
-CXXFLAGS := $(CXXFLAGS) $(BOOST_CFLAGS)
-
-#
-# Configure APR.
-#
-APR_LOCATION := ../lib/apr-1.2.7
-APR_CFLAGS := -I$(APR_LOCATION)/include/apr-1
-APR_LDFLAGS := $(shell $(APR_LOCATION)/bin/apr-1-config --libs) -L$(APR_LOCATION)/lib -lapr-1
-
-CXXFLAGS := $(CXXFLAGS) $(APR_CFLAGS)
-LDFLAGS := $(LDFLAGS) $(APR_LDFLAGS)
-
-#
-# Configure Qpid cpp client.
-#
-
-QPID_CLIENT_LDFLAGS := ../lib/libqpid_common.so.1.0 ../lib/libqpid_client.so.1.0
-QPID_CLIENT_CFLAGS := -I../include
-
-CXXFLAGS := $(CXXFLAGS) $(QPID_CLIENT_CFLAGS)
-LDFLAGS := $(LDFLAGS) $(QPID_CLIENT_LDFLAGS)
-
-CXX := g++
-
-#
-# Add rule to build examples.
-#
-.SUFFIX: .cpp
-%: %.cpp
-	$(CXX) $(CXXFLAGS) $(LDFLAGS) $< -o $@
-
-#
-# Define targets.
-#
-
-EXAMPLES := client_test topic_listener topic_publisher echo_service
-
-.PHONY:
-all: $(EXAMPLES)
-
-.PHONY:
-clean:
-	-rm $(EXAMPLES)
diff --git a/cpp/test/client/client_test.cpp b/cpp/test/client/client_test.cpp
deleted file mode 100644
index 0a3c300f4a..0000000000
--- a/cpp/test/client/client_test.cpp
+++ /dev/null
@@ -1,99 +0,0 @@
-/*
- *
- * 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 <iostream>
-
-#include <qpid/QpidError.h>
-#include <qpid/client/Channel.h>
-#include <qpid/client/Connection.h>
-#include <qpid/client/Message.h>
-#include <qpid/client/MessageListener.h>
-#include <qpid/sys/Monitor.h>
-#include <qpid/framing/FieldTable.h>
-
-using namespace qpid::client;
-using namespace qpid::sys;
-using std::string;
-
-class SimpleListener : public virtual MessageListener{
-    Monitor* monitor;
-
-public:
-    inline SimpleListener(Monitor* _monitor) : monitor(_monitor){}
-
-    inline virtual void received(Message& /*msg*/){
-	std::cout << "Received message " /**<< msg **/<< std::endl;
-	monitor->notify();
-    }
-};
-
-int main(int argc, char**)
-{
-    try{               
-	Connection con(argc > 1);
-	Channel channel;
-	Exchange exchange("MyExchange", Exchange::TOPIC_EXCHANGE);
-	Queue queue("MyQueue", true);
-	
-	string host("localhost");
-	
-	con.open(host);
-	std::cout << "Opened connection." << std::endl;
-	con.openChannel(&channel);
-	std::cout << "Opened channel." << std::endl;	
-	channel.declareExchange(exchange);
-	std::cout << "Declared exchange." << std::endl;
-	channel.declareQueue(queue);
-	std::cout << "Declared queue." << std::endl;
-	qpid::framing::FieldTable args;
-	channel.bind(exchange, queue, "MyTopic", args);
-	std::cout << "Bound queue to exchange." << std::endl;
-
-	//set up a message listener
-	Monitor monitor;
-	SimpleListener listener(&monitor);
-	string tag("MyTag");
-	channel.consume(queue, tag, &listener);
-	channel.start();
-	std::cout << "Registered consumer." << std::endl;
-
-	Message msg;
-	string data("MyMessage");
-	msg.setData(data);
-	channel.publish(msg, exchange, "MyTopic");
-	std::cout << "Published message." << std::endl;
-
-	{
-            Monitor::ScopedLock l(monitor);
-            monitor.wait();
-        }
-        
-	con.closeChannel(&channel);
-	std::cout << "Closed channel." << std::endl;
-	con.close();	
-	std::cout << "Closed connection." << std::endl;
-    }catch(qpid::QpidError error){
-	std::cout << "Error [" << error.code << "] " << error.msg << " ("
-                  << error.location.file << ":" << error.location.line
-                  << ")" << std::endl;
-	return 1;
-    }
-    return 0;
-}
diff --git a/cpp/test/client/echo_service.cpp b/cpp/test/client/echo_service.cpp
deleted file mode 100644
index 3df3da0b86..0000000000
--- a/cpp/test/client/echo_service.cpp
+++ /dev/null
@@ -1,198 +0,0 @@
-/*
- *
- * 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/QpidError.h>
-#include <qpid/client/Channel.h>
-#include <qpid/client/Connection.h>
-#include <qpid/client/Exchange.h>
-#include <qpid/client/MessageListener.h>
-#include <qpid/client/Queue.h>
-#include <qpid/sys/Time.h>
-#include <iostream>
-#include <sstream>
-
-using namespace qpid::client;
-using namespace qpid::sys;
-using std::string;
-
-class EchoServer : public MessageListener{    
-    Channel* const channel;
-public:
-    EchoServer(Channel* channel);
-    virtual void received(Message& msg);
-};
-
-class LoggingListener : public MessageListener{    
-public:
-    virtual void received(Message& msg);
-};
-
-class Args{
-    string host;
-    int port;
-    bool trace;
-    bool help;
-    bool client;
-public:
-    inline Args() : host("localhost"), port(5672), trace(false), help(false), client(false){}
-    void parse(int argc, char** argv);
-    void usage();
-
-    inline const string& getHost() const { return host;}
-    inline int getPort() const { return port; }
-    inline bool getTrace() const { return trace; }
-    inline bool getHelp() const { return help; }
-    inline bool getClient() const { return client; }
-};
-
-int main(int argc, char** argv){
-    const std::string echo_service("echo_service");
-    Args args;
-    args.parse(argc, argv);
-    if (args.getHelp()) {
-        args.usage();
-    } else if (args.getClient()) {
-        try {
-            //Create connection & open a channel
-            Connection connection(args.getTrace());
-            connection.open(args.getHost(), args.getPort());
-            Channel channel;
-            connection.openChannel(&channel);
-        
-            //Setup: declare the private 'response' queue and bind it
-            //to the direct exchange by its name which will be
-            //generated by the server
-            Queue response;
-            channel.declareQueue(response);
-            qpid::framing::FieldTable emptyArgs;
-            channel.bind(Exchange::DEFAULT_DIRECT_EXCHANGE, response, response.getName(), emptyArgs);
-
-            //Consume from the response queue, logging all echoed message to console:
-            LoggingListener listener;
-            std::string tag;
-            channel.consume(response, tag, &listener);
-
-            //Process incoming requests on a new thread
-            channel.start();
-
-            //get messages from console and send them:
-            std::string text;
-            std::cout << "Enter text to send:" << std::endl;
-            while (std::getline(std::cin, text)) {
-                std::cout << "Sending " << text << " to echo server." << std::endl;
-                Message msg;
-                msg.getHeaders().setString("RESPONSE_QUEUE", response.getName());
-                msg.setData(text);
-                channel.publish(msg, Exchange::DEFAULT_DIRECT_EXCHANGE, echo_service);
-                
-                std::cout << "Enter text to send:" << std::endl;
-            }
-            
-            connection.close();
-        } catch(qpid::QpidError error) {
-            std::cout << error.what() << std::endl;
-        }        
-    } else {
-        try {
-            //Create connection & open a channel
-            Connection connection(args.getTrace());
-            connection.open(args.getHost(), args.getPort());
-            Channel channel;
-            connection.openChannel(&channel);
-        
-            //Setup: declare the 'request' queue and bind it to the direct exchange with a 'well known' name
-            Queue request("request");
-            channel.declareQueue(request);
-            qpid::framing::FieldTable emptyArgs;
-            channel.bind(Exchange::DEFAULT_DIRECT_EXCHANGE, request, echo_service, emptyArgs);
-
-            //Consume from the request queue, echoing back all messages received to the client that sent them
-            EchoServer server(&channel);
-            std::string tag = "server_tag";
-            channel.consume(request, tag, &server);
-
-            //Process incoming requests on the main thread
-            channel.run();
-            
-            connection.close();
-        } catch(qpid::QpidError error) {
-            std::cout << error.what() << std::endl;
-        }
-    }
-}
-
-EchoServer::EchoServer(Channel* _channel) : channel(_channel){}
-
-void EchoServer::received(Message& message)
-{
-    //get name of response queues binding to the default direct exchange:
-    const std::string name = message.getHeaders().getString("RESPONSE_QUEUE");
-
-    if (name.empty()) {
-        std::cout << "Cannot echo " << message.getData() << ", no response queue specified." << std::endl;
-    } else {
-        //print message to console:
-        std::cout << "Echoing " << message.getData() << " back to " << name << std::endl;
-        
-        //'echo' the message back:
-        channel->publish(message, Exchange::DEFAULT_DIRECT_EXCHANGE, name);
-    }
-}
-
-void LoggingListener::received(Message& message)
-{
-    //print message to console:
-    std::cout << "Received echo: " << message.getData() << std::endl;
-}
-
-
-void Args::parse(int argc, char** argv){
-    for(int i = 1; i < argc; i++){
-        string name(argv[i]);
-        if("-help" == name){
-            help = true;
-            break;
-        }else if("-host" == name){
-            host = argv[++i];
-        }else if("-port" == name){
-            port = atoi(argv[++i]);
-        }else if("-trace" == name){
-            trace = true;
-        }else if("-client" == name){
-            client = true;
-        }else{
-            std::cout << "Warning: unrecognised option " << name << std::endl;
-        }
-    }
-}
-
-void Args::usage(){
-    std::cout << "Options:" << std::endl;
-    std::cout << "    -help" << std::endl;
-    std::cout << "            Prints this usage message" << std::endl;
-    std::cout << "    -host <host>" << std::endl;
-    std::cout << "            Specifies host to connect to (default is localhost)" << std::endl;
-    std::cout << "    -port <port>" << std::endl;
-    std::cout << "            Specifies port to conect to (default is 5762)" << std::endl;
-    std::cout << "    -trace" << std::endl;
-    std::cout << "            Indicates that the frames sent and received should be logged" << std::endl;
-    std::cout << "    -client" << std::endl;
-    std::cout << "            Run as a client (else will run as a server)" << std::endl;
-}
diff --git a/cpp/test/client/topic_listener.cpp b/cpp/test/client/topic_listener.cpp
deleted file mode 100644
index bd7cfdc62c..0000000000
--- a/cpp/test/client/topic_listener.cpp
+++ /dev/null
@@ -1,186 +0,0 @@
-/*
- *
- * 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/QpidError.h>
-#include <qpid/client/Channel.h>
-#include <qpid/client/Connection.h>
-#include <qpid/client/Exchange.h>
-#include <qpid/client/MessageListener.h>
-#include <qpid/client/Queue.h>
-#include <qpid/sys/Time.h>
-#include <iostream>
-#include <sstream>
-
-using namespace qpid::client;
-using namespace qpid::sys;
-using std::string;
-
-class Listener : public MessageListener{    
-    Channel* const channel;
-    const std::string responseQueue;
-    const bool transactional;
-    bool init;
-    int count;
-    Time start;
-    
-    void shutdown();
-    void report();
-public:
-    Listener(Channel* channel, const std::string& reponseQueue, bool tx);
-    virtual void received(Message& msg);
-};
-
-class Args{
-    string host;
-    int port;
-    int ackMode;
-    bool transactional;
-    int prefetch;
-    bool trace;
-    bool help;
-public:
-    inline Args() : host("localhost"), port(5672), ackMode(NO_ACK), transactional(false), prefetch(1000), trace(false), help(false){}
-    void parse(int argc, char** argv);
-    void usage();
-
-    inline const string& getHost() const { return host;}
-    inline int getPort() const { return port; }
-    inline int getAckMode(){ return ackMode; }
-    inline bool getTransactional() const { return transactional; }
-    inline int getPrefetch(){ return prefetch; }
-    inline bool getTrace() const { return trace; }
-    inline bool getHelp() const { return help; }
-};
-
-int main(int argc, char** argv){
-    Args args;
-    args.parse(argc, argv);
-    if(args.getHelp()){
-        args.usage();
-    }else{
-        try{
-            Connection connection(args.getTrace());
-            connection.open(args.getHost(), args.getPort());
-            Channel channel(args.getTransactional(), args.getPrefetch());
-            connection.openChannel(&channel);
-        
-            //declare exchange, queue and bind them:
-            Queue response("response");
-            channel.declareQueue(response);
-        
-            Queue control;
-            channel.declareQueue(control);
-            qpid::framing::FieldTable bindArgs;
-            channel.bind(Exchange::DEFAULT_TOPIC_EXCHANGE, control, "topic_control", bindArgs);
-            //set up listener
-            Listener listener(&channel, response.getName(), args.getTransactional());
-            std::string tag;
-            channel.consume(control, tag, &listener, args.getAckMode());
-            channel.run();
-            connection.close();
-        }catch(qpid::QpidError error){
-            std::cout << error.what() << std::endl;
-        }
-    }
-}
-
-Listener::Listener(Channel* _channel, const std::string& _responseq, bool tx) : 
-    channel(_channel), responseQueue(_responseq), transactional(tx), init(false), count(0){}
-
-void Listener::received(Message& message){
-    if(!init){        
-        start = now();
-        count = 0;
-        init = true;
-    }
-    std::string type(message.getHeaders().getString("TYPE"));
-
-    if(type == "TERMINATION_REQUEST"){
-        shutdown();
-    }else if(type == "REPORT_REQUEST"){        
-        //send a report:
-        report();
-        init = false;
-    }else if (++count % 100 == 0){        
-        std::cout <<"Received " << count << " messages." << std::endl;
-    }
-}
-
-void Listener::shutdown(){
-    channel->close();
-}
-
-void Listener::report(){
-    Time finish = now();
-    Time time = finish - start;
-    std::stringstream reportstr;
-    reportstr << "Received " << count << " messages in "
-              << time/TIME_MSEC << " ms.";
-    Message msg;
-    msg.setData(reportstr.str());
-    channel->publish(msg, string(), responseQueue);
-    if(transactional){
-        channel->commit();
-    }
-}
-
-
-void Args::parse(int argc, char** argv){
-    for(int i = 1; i < argc; i++){
-        string name(argv[i]);
-        if("-help" == name){
-            help = true;
-            break;
-        }else if("-host" == name){
-            host = argv[++i];
-        }else if("-port" == name){
-            port = atoi(argv[++i]);
-        }else if("-ack_mode" == name){
-            ackMode = atoi(argv[++i]);
-        }else if("-transactional" == name){
-            transactional = true;
-        }else if("-prefetch" == name){
-            prefetch = atoi(argv[++i]);
-        }else if("-trace" == name){
-            trace = true;
-        }else{
-            std::cout << "Warning: unrecognised option " << name << std::endl;
-        }
-    }
-}
-
-void Args::usage(){
-    std::cout << "Options:" << std::endl;
-    std::cout << "    -help" << std::endl;
-    std::cout << "            Prints this usage message" << std::endl;
-    std::cout << "    -host <host>" << std::endl;
-    std::cout << "            Specifies host to connect to (default is localhost)" << std::endl;
-    std::cout << "    -port <port>" << std::endl;
-    std::cout << "            Specifies port to conect to (default is 5762)" << std::endl;
-    std::cout << "    -ack_mode <mode>" << std::endl;
-    std::cout << "            Sets the acknowledgement mode" << std::endl;
-    std::cout << "            0=NO_ACK (default), 1=AUTO_ACK, 2=LAZY_ACK" << std::endl;
-    std::cout << "    -transactional" << std::endl;
-    std::cout << "            Indicates the client should use transactions" << std::endl;
-    std::cout << "    -prefetch <count>" << std::endl;
-    std::cout << "            Specifies the prefetch count (default is 1000)" << std::endl;
-    std::cout << "    -trace" << std::endl;
-    std::cout << "            Indicates that the frames sent and received should be logged" << std::endl;
-}
diff --git a/cpp/test/client/topic_publisher.cpp b/cpp/test/client/topic_publisher.cpp
deleted file mode 100644
index 97d589c1d1..0000000000
--- a/cpp/test/client/topic_publisher.cpp
+++ /dev/null
@@ -1,258 +0,0 @@
-/*
- *
- * 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/QpidError.h>
-#include <qpid/client/Channel.h>
-#include <qpid/client/Connection.h>
-#include <qpid/client/Exchange.h>
-#include <qpid/client/MessageListener.h>
-#include <qpid/client/Queue.h>
-#include <qpid/sys/Monitor.h>
-#include "unistd.h"
-#include <qpid/sys/Time.h>
-#include <cstdlib>
-#include <iostream>
-
-using namespace qpid::client;
-using namespace qpid::sys;
-using std::string;
-
-class Publisher : public MessageListener{    
-    Channel* const channel;
-    const std::string controlTopic;
-    const bool transactional;
-    Monitor monitor;
-    int count;
-    
-    void waitForCompletion(int msgs);
-    string generateData(int size);
-
-public:
-    Publisher(Channel* channel, const std::string& controlTopic, bool tx);
-    virtual void received(Message& msg);
-    int64_t publish(int msgs, int listeners, int size);
-    void terminate();
-};
-
-class Args{
-    string host;
-    int port;
-    int messages;
-    int subscribers;
-    int ackMode;
-    bool transactional;
-    int prefetch;
-    int batches;
-    int delay;
-    int size;
-    bool trace;
-    bool help;
-public:
-    inline Args() : host("localhost"), port(5672), messages(1000), subscribers(1), 
-                    ackMode(NO_ACK), transactional(false), prefetch(1000), batches(1), 
-                    delay(0), size(256), trace(false), help(false){}
-
-    void parse(int argc, char** argv);
-    void usage();
-
-    inline const string& getHost() const { return host;}
-    inline int getPort() const { return port; }
-    inline int getMessages() const { return messages; }
-    inline int getSubscribers() const { return subscribers; }
-    inline int getAckMode(){ return ackMode; }
-    inline bool getTransactional() const { return transactional; }
-    inline int getPrefetch(){ return prefetch; }
-    inline int getBatches(){ return batches; }
-    inline int getDelay(){ return delay; }
-    inline int getSize(){ return size; }
-    inline bool getTrace() const { return trace; }
-    inline bool getHelp() const { return help; }
-};
-
-int main(int argc, char** argv){
-    Args args;
-    args.parse(argc, argv);
-    if(args.getHelp()){
-        args.usage();
-    }else{
-        try{
-            Connection connection(args.getTrace());
-            connection.open(args.getHost(), args.getPort());
-            Channel channel(args.getTransactional(), args.getPrefetch());
-            connection.openChannel(&channel);
-
-            //declare queue (relying on default binding):
-            Queue response("response");
-            channel.declareQueue(response);
-
-            //set up listener
-            Publisher publisher(&channel, "topic_control", args.getTransactional());
-            std::string tag("mytag");
-            channel.consume(response, tag, &publisher, args.getAckMode());
-            channel.start();
-
-            int batchSize(args.getBatches());
-            int64_t max(0);
-            int64_t min(0);
-            int64_t sum(0);
-            for(int i = 0; i < batchSize; i++){
-                if(i > 0 && args.getDelay()) sleep(args.getDelay());
-                Time time = publisher.publish(
-                    args.getMessages(), args.getSubscribers(), args.getSize());
-                if(!max || time > max) max = time;
-                if(!min || time < min) min = time;
-                sum += time;
-                std::cout << "Completed " << (i+1) << " of " << batchSize
-                          << " in " << time/TIME_MSEC << "ms" << std::endl;
-            }
-            publisher.terminate();
-            int64_t avg = sum / batchSize;
-            if(batchSize > 1){
-                std::cout << batchSize << " batches completed. avg=" << avg << 
-                    ", max=" << max << ", min=" << min << std::endl;
-            }
-            channel.close();
-            connection.close();
-        }catch(qpid::QpidError error){
-            std::cout << error.what() << std::endl;
-        }
-    }
-}
-
-Publisher::Publisher(Channel* _channel, const std::string& _controlTopic, bool tx) : 
-    channel(_channel), controlTopic(_controlTopic), transactional(tx){}
-
-void Publisher::received(Message& ){
-    //count responses and when all are received end the current batch
-    Monitor::ScopedLock l(monitor);
-    if(--count == 0){
-        monitor.notify();
-    }
-}
-
-void Publisher::waitForCompletion(int msgs){
-    count = msgs;
-    monitor.wait();
-}
-
-int64_t Publisher::publish(int msgs, int listeners, int size){
-    Message msg;
-    msg.setData(generateData(size));
-    Time start = now();
-    {
-        Monitor::ScopedLock l(monitor);
-        for(int i = 0; i < msgs; i++){
-            channel->publish(msg, Exchange::DEFAULT_TOPIC_EXCHANGE, controlTopic);
-        }
-        //send report request
-        Message reportRequest;
-        reportRequest.getHeaders().setString("TYPE", "REPORT_REQUEST");
-        channel->publish(reportRequest, Exchange::DEFAULT_TOPIC_EXCHANGE, controlTopic);
-        if(transactional){
-            channel->commit();
-        }
-
-        waitForCompletion(listeners);
-    }
-
-    Time finish = now();
-    return finish - start; 
-}
-
-string Publisher::generateData(int size){
-    string data;
-    for(int i = 0; i < size; i++){
-        data += ('A' + (i / 26));
-    }
-    return data;
-}
-
-void Publisher::terminate(){
-    //send termination request
-    Message terminationRequest;
-    terminationRequest.getHeaders().setString("TYPE", "TERMINATION_REQUEST");
-    channel->publish(terminationRequest, Exchange::DEFAULT_TOPIC_EXCHANGE, controlTopic);
-    if(transactional){
-        channel->commit();
-    }
-}
-
-void Args::parse(int argc, char** argv){
-    for(int i = 1; i < argc; i++){
-        string name(argv[i]);
-        if("-help" == name){
-            help = true;
-            break;
-        }else if("-host" == name){
-            host = argv[++i];
-        }else if("-port" == name){
-            port = atoi(argv[++i]);
-        }else if("-messages" == name){
-            messages = atoi(argv[++i]);
-        }else if("-subscribers" == name){
-            subscribers = atoi(argv[++i]);
-        }else if("-ack_mode" == name){
-            ackMode = atoi(argv[++i]);
-        }else if("-transactional" == name){
-            transactional = true;
-        }else if("-prefetch" == name){
-            prefetch = atoi(argv[++i]);
-        }else if("-batches" == name){
-            batches = atoi(argv[++i]);
-        }else if("-delay" == name){
-            delay = atoi(argv[++i]);
-        }else if("-size" == name){
-            size = atoi(argv[++i]);
-        }else if("-trace" == name){
-            trace = true;
-        }else{
-            std::cout << "Warning: unrecognised option " << name << std::endl;
-        }
-    }
-}
-
-void Args::usage(){
-    std::cout << "Options:" << std::endl;
-    std::cout << "    -help" << std::endl;
-    std::cout << "            Prints this usage message" << std::endl;
-    std::cout << "    -host <host>" << std::endl;
-    std::cout << "            Specifies host to connect to (default is localhost)" << std::endl;
-    std::cout << "    -port <port>" << std::endl;
-    std::cout << "            Specifies port to conect to (default is 5762)" << std::endl;
-    std::cout << "    -messages <count>" << std::endl;
-    std::cout << "            Specifies how many messages to send" << std::endl;
-    std::cout << "    -subscribers <count>" << std::endl;
-    std::cout << "            Specifies how many subscribers to expect reports from" << std::endl;
-    std::cout << "    -ack_mode <mode>" << std::endl;
-    std::cout << "            Sets the acknowledgement mode" << std::endl;
-    std::cout << "            0=NO_ACK (default), 1=AUTO_ACK, 2=LAZY_ACK" << std::endl;
-    std::cout << "    -transactional" << std::endl;
-    std::cout << "            Indicates the client should use transactions" << std::endl;
-    std::cout << "    -prefetch <count>" << std::endl;
-    std::cout << "            Specifies the prefetch count (default is 1000)" << std::endl;
-    std::cout << "    -batches <count>" << std::endl;
-    std::cout << "            Specifies how many batches to run" << std::endl;
-    std::cout << "    -delay <seconds>" << std::endl;
-    std::cout << "            Causes a delay between each batch" << std::endl;
-    std::cout << "    -size <bytes>" << std::endl;
-    std::cout << "            Sets the size of the published messages (default is 256 bytes)" << std::endl;
-    std::cout << "    -trace" << std::endl;
-    std::cout << "            Indicates that the frames sent and received should be logged" << std::endl;
-}