/*
 * 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 <thrift/async/TAsyncBufferProcessor.h>
#include <thrift/async/TAsyncProtocolProcessor.h>
#include <thrift/async/TEvhttpServer.h>
#include <thrift/concurrency/ThreadFactory.h>
#include <thrift/concurrency/ThreadManager.h>
#include <thrift/processor/TMultiplexedProcessor.h>
#include <thrift/protocol/TBinaryProtocol.h>
#include <thrift/protocol/TCompactProtocol.h>
#include <thrift/protocol/THeaderProtocol.h>
#include <thrift/protocol/TJSONProtocol.h>
#include <thrift/server/TNonblockingServer.h>
#include <thrift/server/TSimpleServer.h>
#include <thrift/server/TThreadPoolServer.h>
#include <thrift/server/TThreadedServer.h>
#include <thrift/transport/PlatformSocket.h>
#include <thrift/transport/THttpServer.h>
#include <thrift/transport/THttpTransport.h>
#include <thrift/transport/TNonblockingSSLServerSocket.h>
#include <thrift/transport/TNonblockingServerSocket.h>
#include <thrift/transport/TSSLServerSocket.h>
#include <thrift/transport/TSSLSocket.h>
#include <thrift/transport/TServerSocket.h>
#include <thrift/transport/TTransportUtils.h>
#include <thrift/transport/TWebSocketServer.h>
#include <thrift/transport/TZlibTransport.h>

#include "SecondService.h"
#include "ThriftTest.h"

#ifdef HAVE_STDINT_H
#include <stdint.h>
#endif
#ifdef HAVE_INTTYPES_H
#include <inttypes.h>
#endif
#ifdef HAVE_SIGNAL_H
#include <signal.h>
#endif
#ifdef HAVE_SYS_SOCKET_H
#include <sys/socket.h>
#endif
#ifdef HAVE_SYS_UN_H
#include <sys/un.h>
#endif

#include <iostream>
#include <memory>
#include <sstream>
#include <stdexcept>

#include <boost/algorithm/string.hpp>
#include <boost/filesystem.hpp>
#include <boost/program_options.hpp>

#if _WIN32
#include <thrift/windows/TWinsockSingleton.h>
#endif

using namespace std;

using namespace apache::thrift;
using namespace apache::thrift::async;
using namespace apache::thrift::concurrency;
using namespace apache::thrift::protocol;
using namespace apache::thrift::transport;
using namespace apache::thrift::server;

using namespace thrift::test;

// to handle a controlled shutdown, signal handling is mandatory
#ifdef HAVE_SIGNAL_H
apache::thrift::concurrency::Monitor gMonitor;
void signal_handler(int signum)
{
  if (signum == SIGINT) {
    gMonitor.notifyAll();
  }
}
#endif

class TestHandler : public ThriftTestIf {
public:
  TestHandler() = default;

  void testVoid() override { printf("testVoid()\n"); }

  void testString(string& out, const string& thing) override {
    printf("testString(\"%s\")\n", thing.c_str());
    out = thing;
  }

  bool testBool(const bool thing) override {
    printf("testBool(%s)\n", thing ? "true" : "false");
    return thing;
  }

  int8_t testByte(const int8_t thing) override {
    printf("testByte(%d)\n", (int)thing);
    return thing;
  }

  int32_t testI32(const int32_t thing) override {
    printf("testI32(%d)\n", thing);
    return thing;
  }

  int64_t testI64(const int64_t thing) override {
    printf("testI64(%" PRId64 ")\n", thing);
    return thing;
  }

  double testDouble(const double thing) override {
    printf("testDouble(%f)\n", thing);
    return thing;
  }

  void testBinary(std::string& _return, const std::string& thing) override {
    std::ostringstream hexstr;
    hexstr << std::hex << thing;
    printf("testBinary(%lu: %s)\n", static_cast<unsigned long>(thing.size()), hexstr.str().c_str());
    _return = thing;
  }

  void testUuid(apache::thrift::TUuid& _return, const apache::thrift::TUuid& thing) override {
    printf("testUuid(\"{%s}\")\n", to_string(thing).c_str());
    _return = thing;
  }

  void testStruct(Xtruct& out, const Xtruct& thing) override {
    printf("testStruct({\"%s\", %d, %d, %" PRId64 "})\n",
           thing.string_thing.c_str(),
           (int)thing.byte_thing,
           thing.i32_thing,
           thing.i64_thing);
    out = thing;
  }

  void testNest(Xtruct2& out, const Xtruct2& nest) override {
    const Xtruct& thing = nest.struct_thing;
    printf("testNest({%d, {\"%s\", %d, %d, %" PRId64 "}, %d})\n",
           (int)nest.byte_thing,
           thing.string_thing.c_str(),
           (int)thing.byte_thing,
           thing.i32_thing,
           thing.i64_thing,
           nest.i32_thing);
    out = nest;
  }

  void testMap(map<int32_t, int32_t>& out, const map<int32_t, int32_t>& thing) override {
    printf("testMap({");
    map<int32_t, int32_t>::const_iterator m_iter;
    bool first = true;
    for (m_iter = thing.begin(); m_iter != thing.end(); ++m_iter) {
      if (first) {
        first = false;
      } else {
        printf(", ");
      }
      printf("%d => %d", m_iter->first, m_iter->second);
    }
    printf("})\n");
    out = thing;
  }

  void testStringMap(map<std::string, std::string>& out,
                     const map<std::string, std::string>& thing) override {
    printf("testMap({");
    map<std::string, std::string>::const_iterator m_iter;
    bool first = true;
    for (m_iter = thing.begin(); m_iter != thing.end(); ++m_iter) {
      if (first) {
        first = false;
      } else {
        printf(", ");
      }
      printf("%s => %s", (m_iter->first).c_str(), (m_iter->second).c_str());
    }
    printf("})\n");
    out = thing;
  }

  void testSet(set<int32_t>& out, const set<int32_t>& thing) override {
    printf("testSet({");
    set<int32_t>::const_iterator s_iter;
    bool first = true;
    for (s_iter = thing.begin(); s_iter != thing.end(); ++s_iter) {
      if (first) {
        first = false;
      } else {
        printf(", ");
      }
      printf("%d", *s_iter);
    }
    printf("})\n");
    out = thing;
  }

  void testList(vector<int32_t>& out, const vector<int32_t>& thing) override {
    printf("testList({");
    vector<int32_t>::const_iterator l_iter;
    bool first = true;
    for (l_iter = thing.begin(); l_iter != thing.end(); ++l_iter) {
      if (first) {
        first = false;
      } else {
        printf(", ");
      }
      printf("%d", *l_iter);
    }
    printf("})\n");
    out = thing;
  }

  Numberz::type testEnum(const Numberz::type thing) override {
    printf("testEnum(%d)\n", thing);
    return thing;
  }

  UserId testTypedef(const UserId thing) override {
    printf("testTypedef(%" PRId64 ")\n", thing);
    return thing;
  }

  void testMapMap(map<int32_t, map<int32_t, int32_t> >& mapmap, const int32_t hello) override {
    printf("testMapMap(%d)\n", hello);

    map<int32_t, int32_t> pos;
    map<int32_t, int32_t> neg;
    for (int i = 1; i < 5; i++) {
      pos.insert(make_pair(i, i));
      neg.insert(make_pair(-i, -i));
    }

    mapmap.insert(make_pair(4, pos));
    mapmap.insert(make_pair(-4, neg));
  }

  void testInsanity(map<UserId, map<Numberz::type, Insanity> >& insane, const Insanity& argument) override {
    printf("testInsanity()\n");

    Insanity looney;
    map<Numberz::type, Insanity> first_map;
    map<Numberz::type, Insanity> second_map;

    first_map.insert(make_pair(Numberz::TWO, argument));
    first_map.insert(make_pair(Numberz::THREE, argument));

    second_map.insert(make_pair(Numberz::SIX, looney));

    insane.insert(make_pair(1, first_map));
    insane.insert(make_pair(2, second_map));

    printf("return");
    printf(" = {");
    map<UserId, map<Numberz::type, Insanity> >::const_iterator i_iter;
    for (i_iter = insane.begin(); i_iter != insane.end(); ++i_iter) {
      printf("%" PRId64 " => {", i_iter->first);
      map<Numberz::type, Insanity>::const_iterator i2_iter;
      for (i2_iter = i_iter->second.begin(); i2_iter != i_iter->second.end(); ++i2_iter) {
        printf("%d => {", i2_iter->first);
        map<Numberz::type, UserId> userMap = i2_iter->second.userMap;
        map<Numberz::type, UserId>::const_iterator um;
        printf("{");
        for (um = userMap.begin(); um != userMap.end(); ++um) {
          printf("%d => %" PRId64 ", ", um->first, um->second);
        }
        printf("}, ");

        vector<Xtruct> xtructs = i2_iter->second.xtructs;
        vector<Xtruct>::const_iterator x;
        printf("{");
        for (x = xtructs.begin(); x != xtructs.end(); ++x) {
          printf("{\"%s\", %d, %d, %" PRId64 "}, ",
                 x->string_thing.c_str(),
                 (int)x->byte_thing,
                 x->i32_thing,
                 x->i64_thing);
        }
        printf("}");

        printf("}, ");
      }
      printf("}, ");
    }
    printf("}\n");
  }

  void testMulti(Xtruct& hello,
                 const int8_t arg0,
                 const int32_t arg1,
                 const int64_t arg2,
                 const std::map<int16_t, std::string>& arg3,
                 const Numberz::type arg4,
                 const UserId arg5) override {
    (void)arg3;
    (void)arg4;
    (void)arg5;

    printf("testMulti()\n");

    hello.string_thing = "Hello2";
    hello.byte_thing = arg0;
    hello.i32_thing = arg1;
    hello.i64_thing = (int64_t)arg2;
  }

  void testException(const std::string& arg) override {
    printf("testException(%s)\n", arg.c_str());
    if (arg.compare("Xception") == 0) {
      Xception e;
      e.errorCode = 1001;
      e.message = arg;
      throw e;
    } else if (arg.compare("TException") == 0) {
      apache::thrift::TException e;
      throw e;
    } else {
      Xtruct result;
      result.string_thing = arg;
      return;
    }
  }

  void testMultiException(Xtruct& result,
                          const std::string& arg0,
                          const std::string& arg1) override {

    printf("testMultiException(%s, %s)\n", arg0.c_str(), arg1.c_str());

    if (arg0.compare("Xception") == 0) {
      Xception e;
      e.errorCode = 1001;
      e.message = "This is an Xception";
      throw e;
    } else if (arg0.compare("Xception2") == 0) {
      Xception2 e;
      e.errorCode = 2002;
      e.struct_thing.string_thing = "This is an Xception2";
      throw e;
    } else {
      result.string_thing = arg1;
      return;
    }
  }

  void testOneway(const int32_t aNum) override {
    printf("testOneway(%d): call received\n", aNum);
  }
};

class SecondHandler : public SecondServiceIf
{
  public:
    void secondtestString(std::string& result, const std::string& thing) override
    { result = "testString(\"" + thing + "\")"; }
};

class TestProcessorEventHandler : public TProcessorEventHandler {
  void* getContext(const char* fn_name, void* serverContext) override {
    (void)serverContext;
    return new std::string(fn_name);
  }
  void freeContext(void* ctx, const char* fn_name) override {
    (void)fn_name;
    delete static_cast<std::string*>(ctx);
  }
  void preRead(void* ctx, const char* fn_name) override { communicate("preRead", ctx, fn_name); }
  void postRead(void* ctx, const char* fn_name, uint32_t bytes) override {
    (void)bytes;
    communicate("postRead", ctx, fn_name);
  }
  void preWrite(void* ctx, const char* fn_name) override { communicate("preWrite", ctx, fn_name); }
  void postWrite(void* ctx, const char* fn_name, uint32_t bytes) override {
    (void)bytes;
    communicate("postWrite", ctx, fn_name);
  }
  void asyncComplete(void* ctx, const char* fn_name) override {
    communicate("asyncComplete", ctx, fn_name);
  }
  void handlerError(void* ctx, const char* fn_name) override {
    communicate("handlerError", ctx, fn_name);
  }

  void communicate(const char* event, void* ctx, const char* fn_name) {
    std::cout << event << ": " << *static_cast<std::string*>(ctx) << " = " << fn_name << '\n';
  }
};

class TestHandlerAsync : public ThriftTestCobSvIf {
public:
  TestHandlerAsync(std::shared_ptr<TestHandler>& handler) : _delegate(handler) {}
  ~TestHandlerAsync() override = default;

  void testVoid(std::function<void()> cob) override {
    _delegate->testVoid();
    cob();
  }

  void testString(std::function<void(std::string const& _return)> cob,
                          const std::string& thing) override {
    std::string res;
    _delegate->testString(res, thing);
    cob(res);
  }

  void testBool(std::function<void(bool const& _return)> cob, const bool thing) override {
    bool res = _delegate->testBool(thing);
    cob(res);
  }

  void testByte(std::function<void(int8_t const& _return)> cob, const int8_t thing) override {
    int8_t res = _delegate->testByte(thing);
    cob(res);
  }

  void testI32(std::function<void(int32_t const& _return)> cob, const int32_t thing) override {
    int32_t res = _delegate->testI32(thing);
    cob(res);
  }

  void testI64(std::function<void(int64_t const& _return)> cob, const int64_t thing) override {
    int64_t res = _delegate->testI64(thing);
    cob(res);
  }

  void testDouble(std::function<void(double const& _return)> cob, const double thing) override {
    double res = _delegate->testDouble(thing);
    cob(res);
  }

  void testBinary(std::function<void(std::string const& _return)> cob,
                          const std::string& thing) override {
    std::string res;
    _delegate->testBinary(res, thing);
    cob(res);
  }

  void testUuid(::std::function<void(apache::thrift::TUuid const& _return)> cob, const apache::thrift::TUuid& thing) override {
    TUuid res;
    _delegate->testUuid(res, thing);
    cob(res);
  }

  void testStruct(std::function<void(Xtruct const& _return)> cob, const Xtruct& thing) override {
    Xtruct res;
    _delegate->testStruct(res, thing);
    cob(res);
  }

  void testNest(std::function<void(Xtruct2 const& _return)> cob, const Xtruct2& thing) override {
    Xtruct2 res;
    _delegate->testNest(res, thing);
    cob(res);
  }

  void testMap(std::function<void(std::map<int32_t, int32_t> const& _return)> cob,
                       const std::map<int32_t, int32_t>& thing) override {
    std::map<int32_t, int32_t> res;
    _delegate->testMap(res, thing);
    cob(res);
  }

  void testStringMap(
      std::function<void(std::map<std::string, std::string> const& _return)> cob,
      const std::map<std::string, std::string>& thing) override {
    std::map<std::string, std::string> res;
    _delegate->testStringMap(res, thing);
    cob(res);
  }

  void testSet(std::function<void(std::set<int32_t> const& _return)> cob,
                       const std::set<int32_t>& thing) override {
    std::set<int32_t> res;
    _delegate->testSet(res, thing);
    cob(res);
  }

  void testList(std::function<void(std::vector<int32_t> const& _return)> cob,
                        const std::vector<int32_t>& thing) override {
    std::vector<int32_t> res;
    _delegate->testList(res, thing);
    cob(res);
  }

  void testEnum(std::function<void(Numberz::type const& _return)> cob,
                        const Numberz::type thing) override {
    Numberz::type res = _delegate->testEnum(thing);
    cob(res);
  }

  void testTypedef(std::function<void(UserId const& _return)> cob, const UserId thing) override {
    UserId res = _delegate->testTypedef(thing);
    cob(res);
  }

  void testMapMap(
      std::function<void(std::map<int32_t, std::map<int32_t, int32_t> > const& _return)> cob,
      const int32_t hello) override {
    std::map<int32_t, std::map<int32_t, int32_t> > res;
    _delegate->testMapMap(res, hello);
    cob(res);
  }

  void testInsanity(
      std::function<void(std::map<UserId, std::map<Numberz::type, Insanity> > const& _return)> cob,
      const Insanity& argument) override {
    std::map<UserId, std::map<Numberz::type, Insanity> > res;
    _delegate->testInsanity(res, argument);
    cob(res);
  }

  void testMulti(std::function<void(Xtruct const& _return)> cob,
                         const int8_t arg0,
                         const int32_t arg1,
                         const int64_t arg2,
                         const std::map<int16_t, std::string>& arg3,
                         const Numberz::type arg4,
                         const UserId arg5) override {
    Xtruct res;
    _delegate->testMulti(res, arg0, arg1, arg2, arg3, arg4, arg5);
    cob(res);
  }

  void testException(
      std::function<void()> cob,
      std::function<void(::apache::thrift::TDelayedException* _throw)> exn_cob,
      const std::string& arg) override {
    try {
      _delegate->testException(arg);
    } catch (const apache::thrift::TException& e) {
      exn_cob(apache::thrift::TDelayedException::delayException(e));
      return;
    }
    cob();
  }

  void testMultiException(
      std::function<void(Xtruct const& _return)> cob,
      std::function<void(::apache::thrift::TDelayedException* _throw)> exn_cob,
      const std::string& arg0,
      const std::string& arg1) override {
    Xtruct res;
    try {
      _delegate->testMultiException(res, arg0, arg1);
    } catch (const apache::thrift::TException& e) {
      exn_cob(apache::thrift::TDelayedException::delayException(e));
      return;
    }
    cob(res);
  }

  void testOneway(std::function<void()> cob, const int32_t secondsToSleep) override {
    _delegate->testOneway(secondsToSleep);
    cob();
  }

protected:
  std::shared_ptr<TestHandler> _delegate;
};

struct DomainSocketFd {
  THRIFT_SOCKET socket_fd;
  std::string path;
  DomainSocketFd(const std::string& path) : path(path) {
#ifdef HAVE_SYS_UN_H
    unlink(path.c_str());
    socket_fd = socket(AF_UNIX, SOCK_STREAM, IPPROTO_IP);
    if (socket_fd == -1) {
      std::ostringstream os;
      os << "Cannot create domain socket: " << strerror(errno);
      throw std::runtime_error(os.str());
    }
    if (path.size() > sizeof(sockaddr_un::sun_path) - 1)
      throw std::runtime_error("Path size on domain socket too big");
    struct sockaddr_un sa;
    memset(&sa, 0, sizeof(sa));
    sa.sun_family = AF_UNIX;
    strcpy(sa.sun_path, path.c_str());
    int rv = ::bind(socket_fd, (struct sockaddr*)&sa, sizeof(sa));
    if (rv == -1) {
      std::ostringstream os;
      os << "Cannot bind domain socket: " << strerror(errno);
      throw std::runtime_error(os.str());
    }

    rv = ::listen(socket_fd, 16);
    if (rv == -1) {
      std::ostringstream os;
      os << "Cannot listen on domain socket: " << strerror(errno);
      throw std::runtime_error(os.str());
    }
#else
    throw std::runtime_error("Cannot create a domain socket without AF_UNIX");
#endif
  }
  ~DomainSocketFd() {
    ::THRIFT_CLOSESOCKET(socket_fd);
    unlink(path.c_str());
  }
};

namespace po = boost::program_options;

int main(int argc, char** argv) {

#if _WIN32
  transport::TWinsockSingleton::create();
#endif
  int port = 9090;
  bool ssl = false;
  bool zlib = false;
  string transport_type = "buffered";
  string protocol_type = "binary";
  string server_type = "simple";
  string domain_socket = "";
  bool abstract_namespace = false;
  bool emulate_socketactivation = false;
  std::unique_ptr<DomainSocketFd> domain_socket_fd;
  size_t workers = 4;
  int string_limit = 0;
  int container_limit = 0;

  po::options_description desc("Allowed options");
  desc.add_options()
    ("help,h", "produce help message")
    ("port", po::value<int>(&port)->default_value(port), "Port number to listen")
    ("domain-socket", po::value<string>(&domain_socket) ->default_value(domain_socket), "Unix Domain Socket (e.g. /tmp/ThriftTest.thrift)")
    ("abstract-namespace", "Create the domain socket in the Abstract Namespace (no connection with filesystem pathnames)")
    ("emulate-socketactivation","Open the socket from the tester program and pass the library an already open fd")
    ("server-type", po::value<string>(&server_type)->default_value(server_type), "type of server, \"simple\", \"thread-pool\", \"threaded\", or \"nonblocking\"")
    ("transport", po::value<string>(&transport_type)->default_value(transport_type), "transport: buffered, framed, http, websocket, zlib")
    ("protocol", po::value<string>(&protocol_type)->default_value(protocol_type), "protocol: binary, compact, header, json, multi, multic, multih, multij")
    ("ssl", "Encrypted Transport using SSL")
    ("zlib", "Wrapped Transport using Zlib")
    ("processor-events", "processor-events")
    ("workers,n", po::value<size_t>(&workers)->default_value(workers), "Number of thread pools workers. Only valid for thread-pool server type")
    ("string-limit", po::value<int>(&string_limit))
    ("container-limit", po::value<int>(&container_limit));

  po::variables_map vm;
  po::store(po::parse_command_line(argc, argv, desc), vm);
  po::notify(vm);

  if (vm.count("help")) {
    cout << desc << "\n";
    return 1;
  }

  try {
    if (!server_type.empty()) {
      if (server_type == "simple") {
      } else if (server_type == "thread-pool") {
      } else if (server_type == "threaded") {
      } else if (server_type == "nonblocking") {
      } else {
        throw invalid_argument("Unknown server type " + server_type);
      }
    }

    if (!protocol_type.empty()) {
      if (protocol_type == "binary") {
      } else if (protocol_type == "compact") {
      } else if (protocol_type == "json") {
      } else if (protocol_type == "header") {
      } else if (protocol_type == "multi") {  // multiplexed binary
      } else if (protocol_type == "multic") { // multiplexed compact
      } else if (protocol_type == "multih") { // multiplexed header
      } else if (protocol_type == "multij") { // multiplexed json
      } else {
        throw invalid_argument("Unknown protocol type " + protocol_type);
      }
    }

    if (!transport_type.empty()) {
      if (transport_type == "buffered") {
      } else if (transport_type == "framed") {
      } else if (transport_type == "http") {
      } else if (transport_type == "websocket") {
      } else if (transport_type == "zlib") {
        // crosstester will pass zlib as a flag and a transport right now...
      } else {
        throw invalid_argument("Unknown transport type " + transport_type);
      }
    }

  } catch (std::exception& e) {
    cerr << e.what() << '\n';
    cout << desc << "\n";
    return 1;
  }

  if (vm.count("ssl")) {
    ssl = true;
  }

  if (vm.count("zlib")) {
    zlib = true;
  }

#if defined(HAVE_SIGNAL_H) && defined(SIGPIPE)
  if (ssl) {
    signal(SIGPIPE, SIG_IGN); // for OpenSSL, otherwise we end abruptly
  }
#endif

  if (vm.count("abstract-namespace")) {
    abstract_namespace = true;
  }
  if (vm.count("emulate-socketactivation")) {
    emulate_socketactivation = true;
  }

  // Dispatcher
  std::shared_ptr<TProtocolFactory> protocolFactory;
  if (protocol_type == "json" || protocol_type == "multij") {
    std::shared_ptr<TProtocolFactory> jsonProtocolFactory(new TJSONProtocolFactory());
    protocolFactory = jsonProtocolFactory;
  } else if (protocol_type == "compact" || protocol_type == "multic") {
    auto *compactProtocolFactory = new TCompactProtocolFactoryT<TBufferBase>();
    compactProtocolFactory->setContainerSizeLimit(container_limit);
    compactProtocolFactory->setStringSizeLimit(string_limit);
    protocolFactory.reset(compactProtocolFactory);
  } else if (protocol_type == "header" || protocol_type == "multih") {
    std::shared_ptr<TProtocolFactory> headerProtocolFactory(new THeaderProtocolFactory());
    protocolFactory = headerProtocolFactory;
  } else {
    auto* binaryProtocolFactory = new TBinaryProtocolFactoryT<TBufferBase>();
    binaryProtocolFactory->setContainerSizeLimit(container_limit);
    binaryProtocolFactory->setStringSizeLimit(string_limit);
    protocolFactory.reset(binaryProtocolFactory);
  }

  // Processors
  std::shared_ptr<TestHandler> testHandler(new TestHandler());
  std::shared_ptr<TProcessor> testProcessor(new ThriftTestProcessor(testHandler));

  if (vm.count("processor-events")) {
    testProcessor->setEventHandler(
        std::shared_ptr<TProcessorEventHandler>(new TestProcessorEventHandler()));
  }

  // Transport
  std::shared_ptr<TSSLSocketFactory> sslSocketFactory;
  std::shared_ptr<TServerSocket> serverSocket;

  if (ssl) {
    boost::filesystem::path testDir = boost::filesystem::system_complete(argv[0]).parent_path();
    while (!boost::filesystem::exists(testDir / "keys" / "server.crt")) {
      boost::filesystem::path parent = testDir.parent_path();
      if (parent == testDir) {
        cerr << "Unable to locate test/keys from " << argv[0] << '\n';
        return 1;
      }
      testDir = parent;
    }
    string certPath = (testDir / "keys" / "server.crt").string();
    string keyPath = (testDir / "keys" / "server.key").string();

    sslSocketFactory = std::shared_ptr<TSSLSocketFactory>(new TSSLSocketFactory());
    sslSocketFactory->loadCertificate(certPath.c_str());
    sslSocketFactory->loadPrivateKey(keyPath.c_str());
    sslSocketFactory->ciphers("ALL:!ADH:!LOW:!EXP:!MD5:@STRENGTH");
    if (server_type != "nonblocking") {
      serverSocket = std::shared_ptr<TServerSocket>(new TSSLServerSocket(port, sslSocketFactory));
    }
  } else {
    if (domain_socket != "") {
      if (abstract_namespace) {
        std::string abstract_socket("\0", 1);
        abstract_socket += domain_socket;
        serverSocket = std::shared_ptr<TServerSocket>(new TServerSocket(abstract_socket));
      } else {
        if (emulate_socketactivation) {
          unlink(domain_socket.c_str());
          // open and bind the socket
          domain_socket_fd.reset(new DomainSocketFd(domain_socket));
          serverSocket = std::shared_ptr<TServerSocket>(
              new TServerSocket(domain_socket_fd->socket_fd, SocketType::UNIX));
        } else {
          unlink(domain_socket.c_str());
          serverSocket = std::shared_ptr<TServerSocket>(new TServerSocket(domain_socket));
        }
      }
      port = 0;
    } else {
      serverSocket = std::shared_ptr<TServerSocket>(new TServerSocket(port));
    }
  }

  // Factory
  std::shared_ptr<TTransportFactory> transportFactory;

  if (transport_type == "http" && server_type != "nonblocking") {
    transportFactory = std::make_shared<THttpServerTransportFactory>();
  } else if (transport_type == "websocket" && server_type != "nonblocking") {
    if (protocol_type == "json" || protocol_type == "multij") {
      transportFactory = std::make_shared<TTextWebSocketServerTransportFactory>();
    } else {
      transportFactory = std::make_shared<TBinaryWebSocketServerTransportFactory>();
    }
  } else if (transport_type == "framed") {
    transportFactory = std::make_shared<TFramedTransportFactory>();
  } else {
    transportFactory = std::make_shared<TBufferedTransportFactory>();
  }

  if (zlib) {
    // currently TZlibTransportFactory is the only factory than can wrap another:
    transportFactory = std::make_shared<TZlibTransportFactory>(transportFactory);
  }

  // Server Info
  cout << "Starting \"" << server_type << "\" server (" << transport_type << "/" << protocol_type
       << ") listen on: ";
  if (abstract_namespace) {
    cout << '@';
  }
  cout << domain_socket;
  if (port != 0) {
    cout << port;
  }
  cout << '\n';

  // Multiplexed Processor if needed
  if (boost::starts_with(protocol_type, "multi")) {
    std::shared_ptr<SecondHandler> secondHandler(new SecondHandler());
    std::shared_ptr<SecondServiceProcessor> secondProcessor(new SecondServiceProcessor(secondHandler));

    std::shared_ptr<TMultiplexedProcessor> multiplexedProcessor(new TMultiplexedProcessor());
    multiplexedProcessor->registerDefault(testProcessor); // non-multi clients go to the default processor (multi:binary, multic:compact, ...)
    multiplexedProcessor->registerProcessor("ThriftTest", testProcessor);
    multiplexedProcessor->registerProcessor("SecondService", secondProcessor);
    testProcessor = std::dynamic_pointer_cast<TProcessor>(multiplexedProcessor);
  }

  // Server
  std::shared_ptr<apache::thrift::server::TServer> server;

  if (server_type == "simple") {
    server.reset(new TSimpleServer(testProcessor, serverSocket, transportFactory, protocolFactory));
  } else if (server_type == "thread-pool") {

    std::shared_ptr<ThreadFactory> threadFactory
        = std::shared_ptr<ThreadFactory>(new ThreadFactory());

    std::shared_ptr<ThreadManager> threadManager = ThreadManager::newSimpleThreadManager(workers);
    threadManager->threadFactory(threadFactory);
    threadManager->start();

    server.reset(new TThreadPoolServer(testProcessor,
                                       serverSocket,
                                       transportFactory,
                                       protocolFactory,
                                       threadManager));
  } else if (server_type == "threaded") {
    server.reset(
        new TThreadedServer(testProcessor, serverSocket, transportFactory, protocolFactory));
  } else if (server_type == "nonblocking") {
    if (transport_type == "http") {
      std::shared_ptr<TestHandlerAsync> testHandlerAsync(new TestHandlerAsync(testHandler));
      std::shared_ptr<TAsyncProcessor> testProcessorAsync(
          new ThriftTestAsyncProcessor(testHandlerAsync));
      std::shared_ptr<TAsyncBufferProcessor> testBufferProcessor(
          new TAsyncProtocolProcessor(testProcessorAsync, protocolFactory));

      // not loading nonblockingServer into "server" because
      // TEvhttpServer doesn't inherit from TServer, and doesn't
      // provide a stop method.
      TEvhttpServer nonblockingServer(testBufferProcessor, port);
      nonblockingServer.serve();
    } else if (transport_type == "framed") {
      std::shared_ptr<transport::TNonblockingServerTransport> nbSocket;
      nbSocket.reset(
          ssl ? new transport::TNonblockingSSLServerSocket(port, sslSocketFactory)
              : new transport::TNonblockingServerSocket(port));
      server.reset(new TNonblockingServer(testProcessor, protocolFactory, nbSocket));
    } else {
      cerr << "server-type nonblocking requires transport of http or framed" << '\n';
      exit(1);
    }
  }

  if (server.get() != nullptr) {
    if (protocol_type == "header") {
      // Tell the server to use the same protocol for input / output
      // if using header
      server->setOutputProtocolFactory(std::shared_ptr<TProtocolFactory>());
    }

    apache::thrift::concurrency::ThreadFactory factory;
    factory.setDetached(false);
    std::shared_ptr<apache::thrift::concurrency::Runnable> serverThreadRunner(server);
    std::shared_ptr<apache::thrift::concurrency::Thread> thread
        = factory.newThread(serverThreadRunner);

#ifdef HAVE_SIGNAL_H
    signal(SIGINT, signal_handler);
#endif

    thread->start();
    gMonitor.waitForever();         // wait for a shutdown signal

#ifdef HAVE_SIGNAL_H
    signal(SIGINT, SIG_DFL);
#endif

    server->stop();
    thread->join();
    server.reset();
  }

  cout << "done." << '\n';
  return 0;
}