sampleClient.cc

Go to the documentation of this file.

/*
 * Copyright (C) 2007-2015 Hypertable, Inc.
 *
 * This file is part of Hypertable.
 *
 * Hypertable is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License
 * as published by the Free Software Foundation; either version 3
 * of the License, or any later version.
 *
 * Hypertable is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
 * 02110-1301, USA.
 */

#include <Common/Compat.h>

#include "DispatchHandler.h"
#include "Comm.h"
#include "CommHeader.h"
#include "Event.h"

#include <Common/Init.h>
#include <Common/Error.h>
#include <Common/InetAddr.h>
#include <Common/Logger.h>
#include <Common/System.h>
#include <Common/Usage.h>
#include <Common/Serialization.h>

#include <boost/thread/thread.hpp>

#include <chrono>
#include <condition_variable>
#include <cstdio>
#include <fstream>
#include <iostream>
#include <mutex>
#include <queue>
#include <string>
#include <thread>

extern "C" {
#include <netdb.h>
#include <errno.h>
#include <pthread.h>
#include <stdint.h>
#include <string.h>
}

using namespace Hypertable;
using namespace Serialization;
using namespace std;

namespace {
  const char *DEFAULT_HOST = "localhost";
  const int DEFAULT_PORT = 11255;
  const int DEFAULT_TIMEOUT = 10000;
  const char *usage[] = {
    "usage: sampleClient [OPTIONS] <input-file>",
    "",
    "OPTIONS:",
    "  --host=<name>   Specifies the host to connect to (default = localhost)",
    "  --port=<n>      Specifies the port to connect to (default = 11255)",
    "  --recv-addr=<addr>  Let the server connect to us by listening for",
    "                  connection request on <addr> (host:port).  The address",
    "                  that the server is connecting from should be the same",
    "                  as in --host and --port or the defaults.",
    "                  (TCP only)",
    "  --timeout=<t>   Connection timeout in milliseconds (default=10000)",
    "  --verbose       Generate verbose output",
    "  --udp           Operate in UDP mode instead of TCP",
    "",
    "This is a sample program to test the AsyncComm library.  It establishes",
    "a connection with the sampleServer and sends each line of the input file",
    "to the server.  Each reply from the server is echoed to stdout.",
    (const char *)0
  };
  bool g_verbose = false;
}



class ResponseHandler : public DispatchHandler {

public:

  ResponseHandler() { }
  virtual ~ResponseHandler() { }

  virtual void handle(EventPtr &event_ptr) = 0;

  virtual bool get_response(EventPtr &event_ptr) = 0;

protected:
  std::queue<EventPtr> m_queue;
  std::mutex m_mutex;
  std::condition_variable m_cond;
};





class ResponseHandlerTCP : public ResponseHandler {

public:

  ResponseHandlerTCP() { }

  virtual void handle(EventPtr &event_ptr) {
    std::lock_guard<std::mutex> lock(m_mutex);
    if (event_ptr->type == Event::CONNECTION_ESTABLISHED) {
      if (g_verbose)
        HT_INFOF("Connection Established - %s", event_ptr->to_str().c_str());
      m_connected = true;
      m_cond.notify_one();
    }
    else if (event_ptr->type == Event::DISCONNECT) {
      if (event_ptr->error != 0) {
        HT_INFOF("Disconnect : %s", Error::get_text(event_ptr->error));
      }
      else {
        HT_INFO("Disconnect.");
      }
      m_connected = false;
      m_cond.notify_one();
    }
    else if (event_ptr->type == Event::ERROR) {
      HT_INFOF("Error : %s", Error::get_text(event_ptr->error));
      //exit(EXIT_FAILURE);
    }
    else if (event_ptr->type == Event::MESSAGE) {
      m_queue.push(event_ptr);
      m_cond.notify_one();
    }
  }

  bool wait_for_connection() {
    std::unique_lock<std::mutex> lock(m_mutex);
    if (m_connected)
      return true;
    m_cond.wait(lock);
    return m_connected;
  }

  virtual bool get_response(EventPtr &event_ptr) {
    std::unique_lock<std::mutex> lock(m_mutex);
    while (m_queue.empty()) {
      if (m_connected == false)
        return false;
      m_cond.wait(lock);
    }
    event_ptr = m_queue.front();
    m_queue.pop();
    return true;
  }

private:
  bool m_connected {};
};



class HandlerFactory : public ConnectionHandlerFactory {
public:
  HandlerFactory(DispatchHandlerPtr &dhp) {
    m_dispatch_handler_ptr = dhp;
  }
  virtual void get_instance(DispatchHandlerPtr &dhp) {
    dhp = m_dispatch_handler_ptr;
  }
private:
  DispatchHandlerPtr m_dispatch_handler_ptr;
};




class ResponseHandlerUDP : public ResponseHandler {

public:

  ResponseHandlerUDP() : ResponseHandler() { return; }

  virtual void handle(EventPtr &event_ptr) {
    std::lock_guard<std::mutex> lock(m_mutex);
    if (event_ptr->type == Event::MESSAGE) {
      m_queue.push(event_ptr);
      m_cond.notify_one();
    }
    else {
      HT_INFOF("%s", event_ptr->to_str().c_str());
      //exit(EXIT_FAILURE);
    }
  }

  virtual bool get_response(EventPtr &event_ptr) {
    std::unique_lock<std::mutex> lock(m_mutex);
    m_cond.wait(lock, [this](){ return !m_queue.empty(); });
    event_ptr = m_queue.front();
    m_queue.pop();
    return true;
  }
};





int main(int argc, char **argv) {
  Comm *comm;
  int rval;
  const char *host = DEFAULT_HOST;
  struct sockaddr_in addr;
  uint16_t port = DEFAULT_PORT;
  time_t timeout = DEFAULT_TIMEOUT;
  const char *in_file = 0;
  int error;
  EventPtr event_ptr;
  DispatchHandlerPtr dhp;
  ResponseHandler *resp_handler;
  bool udp_mode = false;
  string line;
  int outstanding = 0;
  int max_outstanding = 50;
  const char *str;
  CommAddress udp_send_addr;
  sockaddr_in inet_addr;

  Config::init(0, 0);

  memset(&inet_addr, 0, sizeof(inet_addr));

  if (argc == 1)
    Usage::dump_and_exit(usage);

  ReactorFactory::initialize(1);

  for (int i=1; i<argc; i++) {
    if (!strncmp(argv[i], "--host=", 7))
      host = &argv[i][7];
    else if (!strncmp(argv[i], "--port=", 7)) {
      rval = atoi(&argv[i][7]);
      if (rval <= 1024 || rval > 65535) {
        cerr << "Invalid port.  Must be in the range of 1024-65535." << endl;
        exit(EXIT_FAILURE);
      }
      port = rval;
    }
    else if (!strncmp(argv[i], "--timeout=", 10))
      timeout = (time_t)atoi(&argv[i][10]);
    else if (!strcmp(argv[i], "--udp"))
      udp_mode = true;
    else if (!strncmp(argv[i], "--recv-addr=", 12)) {
      if (!InetAddr::initialize(&inet_addr, &argv[i][12]))
        HT_ABORT;
    }
    else if (!strcmp(argv[i], "--verbose")) {
      g_verbose = true;
    }
    else if (in_file == 0)
      in_file = argv[i];
    else
      Usage::dump_and_exit(usage);
  }

  if (in_file == 0)
    Usage::dump_and_exit(usage);

  if (!InetAddr::initialize(&addr, host, port))
    exit(EXIT_FAILURE);

  comm = Comm::instance();

  ifstream myfile(in_file);

  if (!myfile.is_open()) {
    HT_ERRORF("Unable to open file '%s' : %s", in_file, strerror(errno));
    return 0;
  }

  if (udp_mode) {
    assert(inet_addr.sin_port == 0);
    dhp = make_shared<ResponseHandlerUDP>();
    resp_handler = static_cast<ResponseHandler *>(dhp.get());
    port++;
    InetAddr::initialize(&inet_addr, INADDR_ANY, port);
    udp_send_addr.set_inet(inet_addr);
    comm->create_datagram_receive_socket(udp_send_addr, 0, dhp);
  }
  else {
    dhp = make_shared<ResponseHandlerTCP>();
    resp_handler = static_cast<ResponseHandler *>(dhp.get());

    if (inet_addr.sin_port == 0) {
      if ((error = comm->connect(addr, dhp)) != Error::OK) {
        HT_ERRORF("Comm::connect error - %s", Error::get_text(error));
        exit(EXIT_FAILURE);
      }
    }
    else {
      ConnectionHandlerFactoryPtr handler_factory = make_shared<HandlerFactory>(dhp);
      comm->listen(inet_addr, handler_factory, dhp);
    }
    if (!((ResponseHandlerTCP *)resp_handler)->wait_for_connection())
      exit(EXIT_FAILURE);

  }

  CommHeader header;
  const uint8_t *decode_ptr;
  size_t decode_remain;

  while (!myfile.eof()) {
    getline (myfile,line);
    if (line.length() > 0) {
      CommBufPtr cbp(new CommBuf(header, encoded_length_str16(line)));
      cbp->append_str16(line);
      int retries = 0;

      if (udp_mode) {
        if ((error = comm->send_datagram(addr, udp_send_addr, cbp)) != Error::OK) {
          HT_ERRORF("Problem sending datagram - %s", Error::get_text(error));
          return 1;
        }
      }
      else {
        while ((error = comm->send_request(addr, timeout, cbp, resp_handler))
                != Error::OK) {
          if (error == Error::COMM_NOT_CONNECTED) {
            if (retries == 5) {
              HT_ERROR("Connection timeout.");
              return 1;
            }
            this_thread::sleep_for(chrono::milliseconds(1000));
            retries++;
          }
          else {
            HT_ERRORF("CommEngine::send_message returned '%s'",
                      Error::get_text(error));
            return 1;
          }
        }
      }
      outstanding++;

      if (outstanding  > max_outstanding) {
        if (!resp_handler->get_response(event_ptr))
          break;
        try {
          decode_ptr = event_ptr->payload;
          decode_remain = event_ptr->payload_len;
          str = decode_str16(&decode_ptr, &decode_remain);
          if (*str != 0)
            cout << "ECHO: " << str << endl;
          else
            cout << "[NULL]" << endl;
        }
        catch (Exception &e) {
          cout <<"Error: "<< e << endl;
        }
        outstanding--;
      }
    }
  }

  while (outstanding > 0 && resp_handler->get_response(event_ptr)) {
    try {
      decode_ptr = event_ptr->payload;
      decode_remain = event_ptr->payload_len;
      str = decode_str16(&decode_ptr, &decode_remain);
      if (str != 0)
        cout << "ECHO: " << str << endl;
      else
        cout << "[NULL]" << endl;
    }
    catch (Exception &e) {
      cout <<"Error: "<< e << endl;
    }
    //cout << "out = " << outstanding << endl;
    outstanding--;
  }


  myfile.close();

  ReactorFactory::destroy();

  return 0;
}