/*
 *
 * 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.
 *
 */

/**
 * This class provides an example of using AMQP for a request-response
 * style system. 'Requests' are messages sent to a well known
 * destination. A 'service' process consumes these message and
 * responds by echoing the message back to the sender on a
 * sender-specified private queue.
 */

#include <QpidError.h>
#include <ClientChannel.h>
#include <Connection.h>
#include <ClientExchange.h>
#include <MessageListener.h>
#include <ClientQueue.h>
#include <sys/Time.h>
#include <iostream>
#include <sstream>

using namespace qpid::client;
using namespace qpid::sys;
using std::string;


/**
 * A message listener implementation representing the 'service', this
 * will 'echo' any requests received.
 */
class EchoServer : public MessageListener{    
    Channel* const channel;
public:
    EchoServer(Channel* channel);
    virtual void received(Message& msg);
};

/** 
 * A message listener implementation that merely prints received
 * messages to the console. Used to report on 'echo' responses.
 */
class LoggingListener : public MessageListener{    
public:
    virtual void received(Message& msg);
};

/**
 * A utility class that manages the command line options needed to run
 * the example confirgurably.
 */
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; }
};

/**
 * The main test path. There are two basic modes: 'client' and
 * 'service'. First one or more services are started, then one or more
 * clients are started and messages can be sent.
 */
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()) {
        //we have been started in 'client' mode, i.e. we will send an
        //echo requests and print responses received.
        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::STANDARD_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::STANDARD_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 {
        // we are in 'service' mode, i.e. we will consume messages
        // from the request queue and echo each request back to the
        // senders own private response queue.
        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::STANDARD_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::STANDARD_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;
}