text:
#include <algorithm>

#include <memo/cli/Network.hh>

#include <boost/algorithm/string/trim.hpp>
#include <boost/range/adaptor/transformed.hpp>
#include <boost/range/algorithm_ext/erase.hpp>

#include <elle/algorithm.hh>
#include <elle/log/FileLogger.hh>
#include <elle/make-vector.hh>
#include <elle/print.hh>
#include <elle/reactor/network/resolve.hh>

#include <memo/Network.hh> // Storages
#include <memo/cli/Memo.hh>
#include <memo/cli/utility.hh>
#include <memo/cli/xattrs.hh>
#include <memo/environ.hh>
#include <memo/grpc/grpc.hh>
#include <memo/grpc/memo_vs.hh>
#include <memo/log.hh>
#include <memo/model/MissingBlock.hh>
#include <memo/model/MonitoringServer.hh>
#include <memo/model/blocks/ACLBlock.hh>
#include <memo/model/blocks/ImmutableBlock.hh>
#include <memo/model/blocks/MutableBlock.hh>
#include <memo/model/doughnut/Doughnut.hh>
#include <memo/model/doughnut/NB.hh>
#include <memo/model/doughnut/consensus/Paxos.hh>
#include <memo/overlay/Kalimero.hh>
#include <memo/overlay/kelips/Kelips.hh>
#include <memo/overlay/kouncil/Configuration.hh>
#include <memo/silo/Silo.hh>
#include <memo/silo/Strip.hh>

#ifndef ELLE_WINDOWS
# include <elle/reactor/network/unix-domain-socket.hh>
#endif

ELLE_LOG_COMPONENT("cli.network");

using boost::adaptors::transformed;

namespace memo
{
  namespace cli
  {
    using Strings = Network::Strings;
    namespace bfs = boost::filesystem;
    namespace dnut = memo::model::doughnut;

    Network::Network(Memo& memo)
      : Object(memo)
      , create(*this,
               "Create a network",
               cli::name,
               cli::description = boost::none,
               cli::silo = Strings{},
               cli::port = boost::none,
               cli::replication_factor = 1,
               cli::eviction_delay = boost::none,
               cli::output = boost::none,
               cli::push_network = false,
               cli::push = false,
               cli::admin_r = Strings{},
               cli::admin_rw = Strings{},
               cli::peer = Strings{},
               // Consensus types.
               cli::paxos = false,
               cli::no_consensus = false,
               // Overlay types.
               cli::kelips = false,
               cli::kalimero = false,
               cli::kouncil = false,
               // Kelips options,
               cli::nodes = boost::none,
               cli::k = boost::none,
               cli::kelips_contact_timeout = elle::Duration{2min},
               cli::encrypt = boost::none,
               // Generic options
               cli::protocol = boost::none,
               cli::tcp_heartbeat = boost::none,
               cli::disable_encrypt_at_rest = false,
               cli::disable_encrypt_rpc = false,
               cli::disable_signature = false)
      , delete_(*this,
                "Delete a network locally",
                cli::name,
                cli::pull = false,
                cli::purge = false,
                cli::unlink = false)
      , export_(*this,
                "Export a network",
                cli::name,
                cli::output = boost::none)
      , fetch(*this,
              "Fetch a network from {hub}",
              cli::name = boost::none)
      , import(*this,
               "Fetch a network",
               cli::input = boost::none)
#ifndef ELLE_WINDOWS
      , inspect(*this,
                "Get information about a running network",
                cli::name,
                cli::output = boost::none,
                cli::status = false,
                cli::peers = false,
                cli::all = false,
                cli::redundancy = false)
#endif
      , link(*this,
             "Link this device to a network",
             cli::name,
             cli::silo = Strings{},
             cli::output = boost::none,
             cli::node_id = boost::none)
      , list(*this, "List networks")
      , list_services(*this,
                      "List network registered services",
                      cli::name,
                      cli::peer = Strings(),
                      cli::async = false,
                      cli::cache = false,
                      cli::cache_ram_size = boost::none,
                      cli::cache_ram_ttl = boost::none,
                      cli::cache_ram_invalidation = boost::none,
                      cli::cache_disk_size = boost::none,
                      cli::fetch_endpoints = false,
                      cli::fetch = false,
                      cli::push_endpoints = false,
                      cli::push = false,
                      cli::publish = false,
                      cli::endpoints_file = boost::none,
                      cli::port_file = boost::none,
                      cli::port = boost::none,
                      cli::peers_file = boost::none,
                      cli::listen = boost::none,
                      cli::fetch_endpoints_interval = boost::none,
                      cli::no_local_endpoints = false,
                      cli::no_public_endpoints = false,
                      cli::advertise_host = Strings())
      , list_silos(*this,
                   "List all silos contributed by this device to a network",
                   cli::name)
      , pull(*this,
             "Remove a network from {hub}",
             cli::name,
             cli::purge = false)
      , push(*this,
             "Push a network to {hub}",
             cli::name)
      , run(*this,
            "Run a network",
            cli::name,
            cli::input = boost::none,
#ifndef ELLE_WINDOWS
            cli::daemon = false,
            cli::monitoring = true,
#endif
            cli::peer = Strings{},
            cli::async = false,
            cli::cache = false,
            cli::cache_ram_size = boost::none,
            cli::cache_ram_ttl = boost::none,
            cli::cache_ram_invalidation = boost::none,
            cli::cache_disk_size = boost::none,
            cli::fetch_endpoints = false,
            cli::fetch = false,
            cli::push_endpoints = false,
            cli::push = false,
            cli::publish = false,
            cli::endpoints_file = boost::none,
            cli::port_file = boost::none,
            cli::port = boost::none,
            cli::peers_file = boost::none,
            cli::listen = boost::none,
            cli::fetch_endpoints_interval = boost::none,
            cli::no_local_endpoints = false,
            cli::no_public_endpoints = false,
            cli::advertise_host = Strings{},
            cli::grpc = boost::none,
            cli::grpc_port_file = boost::none,
            cli::paxos_rebalancing_auto_expand = boost::none,
            cli::paxos_rebalancing_inspect = boost::none,
            cli::resign_on_shutdown = boost::none)
      , stat(*this,
              "Fetch stats of a network on {hub}",
              cli::name)
      , unlink(*this,
               "Unlink this device from a network",
               cli::name)
      , update(*this,
               "Update a network",
               cli::name,
               cli::description = boost::none,
               cli::port = boost::none,
               cli::output = boost::none,
               cli::push_network = false,
               cli::push = false,
               cli::admin_r = Strings{},
               cli::admin_rw = Strings{},
               cli::admin_remove = Strings{},
               cli::mountpoint = boost::none,
               cli::peer = Strings{},
               cli::protocol = boost::none)
    {}

    /*---------------.
    | Mode: create.  |
    `---------------*/

    namespace
    {
      auto
      make_kelips_config(boost::optional<int> nodes,
                         boost::optional<int> k,
                         elle::Duration& timeout,
                         boost::optional<std::string> const& encrypt,
                         boost::optional<std::string> const& protocol)
      {
        auto res = std::make_unique<memo::overlay::kelips::Configuration>();
        res->k = [&]{
          if (k)
            return *k;
          else if (nodes)
            {
              if (*nodes < 10)
                return 1;
              else if (*nodes < 25)
                return *nodes / 5;
              else
                return int(sqrt(*nodes));
            }
          else
            return 1;
        }();
        res->contact_timeout = timeout;
        // encrypt support.
        {
          auto enc = encrypt.value_or("yes");
          if (enc == "no")
          {
            res->encrypt = false;
            res->accept_plain = true;
          }
          else if (enc == "lazy")
          {
            res->encrypt = true;
            res->accept_plain = true;
          }
          else if (enc == "yes")
          {
            res->encrypt = true;
            res->accept_plain = false;
          }
          else
            elle::err<CLIError>("'encrypt' must be 'no', 'lazy' or 'yes': %s",
                                enc);
        }
        if (protocol)
          res->rpc_protocol = protocol_get(protocol);
        return res;
      }

      std::unique_ptr<memo::silo::SiloConfig>
      make_silo_config(memo::Memo& memo,
                       Strings const& silos)
      {
        auto res = std::unique_ptr<memo::silo::SiloConfig>{};
        if (silos.empty())
          return {};
        else
        {
          auto backends
            = elle::make_vector(silos,
                                [&](auto const& s) { return memo.silo_get(s); });
          if (backends.size() == 1)
            return std::move(backends[0]);
          else
            return std::make_unique<memo::silo::StripSiloConfig>
              (std::move(backends));
        }
      }

      // Consensus
      auto
      make_consensus_config(bool paxos,
                            bool no_consensus,
                            int replication_factor,
                            boost::optional<std::string> const& eviction_delay)
        -> std::unique_ptr<dnut::consensus::Configuration>
      {
        if (replication_factor < 1)
          elle::err<CLIError>("replication factor must be greater than 0");
        if (!no_consensus)
          paxos = true;
        if (1 < no_consensus + paxos)
          elle::err<CLIError>("more than one consensus specified");
        if (paxos)
          return std::make_unique<
            dnut::consensus::Paxos::Configuration>(
              replication_factor,
              eviction_delay
              ? std::chrono::duration_from_string<std::chrono::seconds>(
                *eviction_delay)
              : std::chrono::seconds(10 * 60));
        else
        {
          if (replication_factor != 1)
            elle::err("without consensus, replication factor must be 1");
          return std::make_unique<
            dnut::consensus::Configuration>();
        }
      }

      auto
      make_admin_keys(memo::Memo& memo,
                      Strings const& admin_r,
                      Strings const& admin_rw)
        -> dnut::AdminKeys
      {
        auto res = dnut::AdminKeys{};
        auto add =
          [&res] (elle::cryptography::rsa::PublicKey const& key,
                  bool read, bool write)
          {
            if (read && !write)
              push_back_if_missing(res.r, key);
            // write implies rw.
            if (write)
              push_back_if_missing(res.w, key);
          };
        for (auto const& a: admin_r)
          add(memo.user_get(a).public_key, true, false);
        for (auto const& a: admin_rw)
          add(memo.user_get(a).public_key, true, true);
        return res;
      }
    }

    void
    Network::mode_create(
      std::string const& network_name,
      boost::optional<std::string> const& description,
      Strings const& silos_names,
      boost::optional<int> port,
      int replication_factor,
      boost::optional<std::string> const& eviction_delay,
      boost::optional<std::string> const& output_name,
      bool push_network,
      bool push,
      Strings const& admin_r,
      Strings const& admin_rw,
      Strings const& peer,
      // Consensus types.
      bool paxos,
      bool no_consensus,
      // Overlay types.
      bool kelips,
      bool kalimero,
      bool kouncil,
      // Kelips options,
      boost::optional<int> nodes,
      boost::optional<int> k,
      elle::Duration kelips_contact_timeout,
      boost::optional<std::string> encrypt,
      boost::optional<std::string> protocol,
      elle::DurationOpt tcp_heartbeat,
      bool disable_encrypt_at_rest,
      bool disable_encrypt_rpc,
      bool disable_signature)
    {
      ELLE_TRACE_SCOPE("create");
      auto& cli = this->cli();
      auto& memo = cli.backend();
      auto owner = cli.as_user();
      auto overlay_config = [&]{
          auto res = std::unique_ptr<memo::overlay::Configuration>{};
          if (1 < kalimero + kelips + kouncil)
            elle::err<CLIError>("only one overlay type must be specified");
          if (kalimero)
            res = std::make_unique<memo::overlay::KalimeroConfiguration>();
          else if (kelips)
            res = make_kelips_config(nodes, k, kelips_contact_timeout,
                                     encrypt, protocol);
          else
            res = std::make_unique<memo::overlay::kouncil::Configuration>();
          if (protocol)
            res->rpc_protocol = protocol_get(protocol);
          return res;
        }();
      auto const encrypt_options = memo::model::doughnut::EncryptOptions{
        !disable_encrypt_at_rest,
        !disable_encrypt_rpc,
        !disable_signature};
      auto dht =
        std::make_unique<dnut::Configuration>(
          memo::model::Address::random(0),
          make_consensus_config(paxos, no_consensus, replication_factor,
                                eviction_delay),
          std::move(overlay_config),
          make_silo_config(memo, silos_names),
          owner.keypair(),
          std::make_shared<elle::cryptography::rsa::PublicKey>(
            owner.public_key),
          dnut::Passport(
            owner.public_key,
            memo.qualified_name(network_name, owner),
            elle::cryptography::rsa::KeyPair(owner.public_key,
                                                owner.private_key.get())),
          owner.name,
          std::move(port),
          memo::version(),
          make_admin_keys(memo, admin_r, admin_rw),
          parse_peers(peer),
          tcp_heartbeat,
          encrypt_options);
      {
        auto network = memo::Network(memo.qualified_name(network_name, owner),
                                        std::move(dht),
                                        description);
        auto const desc = [&] {
            if (output_name)
            {
              auto output = cli.get_output(output_name);
              elle::serialization::json::serialize(network, *output, false);
              return std::make_unique<memo::NetworkDescriptor>(std::move(network));
            }
            else
            {
              memo.network_save(owner, network);
              auto res = std::make_unique<memo::NetworkDescriptor>(std::move(network));
              memo.network_save(*res);
              return res;
            }
          }();
        if (push || push_network)
          memo.hub_push("network", desc->name, *desc, owner);
      }
    }

    /*---------------.
    | Mode: delete.  |
    `---------------*/

    void
    Network::mode_delete(std::string const& network_name,
                         bool pull,
                         bool purge,
                         bool unlink)
    {
      ELLE_TRACE_SCOPE("delete");
      auto& cli = this->cli();
      auto& memo = cli.backend();
      auto owner = cli.as_user();

      auto network = memo.network_get(network_name, owner, false);
      auto linked_users = memo.network_linked_users(network.name);
      if (linked_users.size() && !unlink)
      {
        auto user_names = elle::make_vector(linked_users,
                                            [](auto const& u) { return u.name; });
        elle::err("Network is still linked with this device by %s. "
                  "Please unlink it first or add the --unlink flag",
                  user_names);
      }
      if (purge)
      {
        auto key_value_stores = memo.key_value_stores_for_network(network.name);
        for (auto const& k: key_value_stores)
          memo.key_value_store_delete(k);
        for (auto const& user: memo.user_passports_for_network(network.name))
          memo.passport_delete(network.name, user);
      }
      if (pull)
        memo.hub_delete("network", network.name, owner, true, purge);
      memo.network_delete(network.name, owner, unlink);
    }

    /*---------------.
    | Mode: export.  |
    `---------------*/

    void
    Network::mode_export(std::string const& network_name,
                         boost::optional<std::string> const& output_name)
    {
      ELLE_TRACE_SCOPE("export");
      auto& cli = this->cli();
      auto& memo = cli.backend();
      auto owner = cli.as_user();
      auto name = memo.qualified_name(network_name, owner);
      auto desc = memo.network_descriptor_get(network_name, owner);
      auto output = cli.get_output(output_name);
      name = desc.name;
      elle::serialization::json::serialize(desc, *output, false);
      cli.report_exported(*output, "network", desc.name);
    }

    /*--------------.
    | Mode: fetch.  |
    `--------------*/

    void
    Network::mode_fetch(boost::optional<std::string> const& network_name)
    {
      ELLE_TRACE_SCOPE("export");
      auto& cli = this->cli();
      auto& memo = cli.backend();
      auto owner = cli.as_user();

      auto save = [&memo] (memo::NetworkDescriptor const desc_) {
        // Save or update network descriptor.
        memo.network_save(desc_, true);
        for (auto const& u: memo.network_linked_users(desc_.name))
        {
          // Copy network descriptor.
          auto desc = desc_;
          auto network = memo.network_get(desc.name, u, false);
          if (network.model)
          {
            auto* d = dynamic_cast<dnut::Configuration*>(
              network.model.get()
            );
            auto updated_network = memo::Network(
              desc.name,
              std::make_unique<dnut::Configuration>(
                d->id,
                std::move(desc.consensus),
                std::move(desc.overlay),
                // XXX[Silo]: dnut::Configuration::storage ?
                std::move(d->storage),
                u.keypair(),
                std::make_shared<elle::cryptography::rsa::PublicKey>(desc.owner),
                d->passport,
                u.name,
                d->port,
                desc.version,
                desc.admin_keys,
                desc.peers,
                elle::DurationOpt{desc.tcp_heartbeat},
                desc.encrypt_options),
              desc.description);
            // Update linked network for user.
            memo.network_save(u, updated_network, true);
          }
        }
      };
      if (network_name)
      {
        auto const name = memo.qualified_name(*network_name, owner);
        auto const network = memo.hub_fetch<memo::NetworkDescriptor>(
          "network", name);
        save(network);
      }
      else
      {
        // Fetch all networks for owner.
        using Networks
          = std::unordered_map<std::string, std::vector<memo::NetworkDescriptor>>;
        auto const res =
          memo.hub_fetch<Networks>(
            elle::print("users/%s/networks", owner.name),
            "networks for user",
            owner.name,
            owner);
        for (auto const& n: res.at("networks"))
          save(n);
      }
    }

    /*---------------.
    | Mode: import.  |
    `---------------*/
    void
    Network::mode_import(boost::optional<std::string> const& input_name)
    {
      ELLE_TRACE_SCOPE("import");
      auto& cli = this->cli();
      auto& memo = cli.backend();
      auto input = this->cli().get_input(input_name);
      auto desc =
        elle::serialization::json::deserialize<memo::NetworkDescriptor>
        (*input, false);
      memo.network_save(desc);
      cli.report_imported("network", desc.name);
    }

#ifndef ELLE_WINDOWS
    /*----------------.
    | Mode: inspect.  |
    `----------------*/

    void
    Network::mode_inspect(std::string const& network_name,
                          boost::optional<std::string> const& output_name,
                          bool status,
                          bool peers,
                          bool all,
                          bool redundancy)
    {
      ELLE_TRACE_SCOPE("inspect");
      auto& cli = this->cli();
      auto& memo = cli.backend();
      auto owner = cli.as_user();
      auto network = memo.network_get(network_name, owner, false);
      auto s_path = memo._network_monitoring_socket_path(network.name, owner);
      if (!bfs::exists(s_path))
        elle::err("network not running or monitoring disabled");
      elle::reactor::network::UnixDomainSocket socket(s_path);
      using Monitoring = memo::model::MonitoringServer;
      using Query = memo::model::MonitoringServer::MonitorQuery::Query;
      auto do_query = [&] (Query query_val)
        {
          auto query = Monitoring::MonitorQuery(query_val);
          elle::serialization::json::serialize(query, socket, false, false);
          auto json = elle::json::read(socket);
          return Monitoring::MonitorResponse(std::move(json));
        };
      auto print_response = [&] (Monitoring::MonitorResponse const& response)
        {
          if (cli.script())
            elle::json::write(*cli.get_output(output_name), response.as_object());
          else
          {
            if (response.error)
              std::cout << "Error: " << response.error.get() << std::endl;
            if (response.result)
              std::cout << elle::json::pretty_print(response.result.get())
                        << std::endl;
            else
              std::cout << "Running" << std::endl;
          }
        };

      if (status)
        print_response(do_query(Query::Status));
      else if (all)
        print_response(do_query(Query::Stats));
      else if (redundancy)
      {
        auto res = do_query(Query::Stats);
        if (res.result)
        {
          auto redundancy = res.result.get()["redundancy"];
          if (cli.script())
            elle::json::write(*cli.get_output(output_name), redundancy);
          else
            std::cout << elle::json::pretty_print(redundancy) << std::endl;
        }
      }
      else if (peers)
      {
        auto res = do_query(Query::Stats);
        if (res.result)
        {
          auto peer_list = res.result.get()["peers"];
          if (cli.script())
            elle::json::write(*cli.get_output(output_name), peer_list);
          else
          {
            if (peer_list.empty())
              std::cout << "No peers" << std::endl;
            else
              for (auto json: peer_list)
              {
                elle::print(std::cout, "{}", std::string(json["id"]));
                json.erase("id");
                if (json.size())
                  elle::print(std::cout, ": {}\n", json);
              }
          }
        }
      }
      else
        elle::err<CLIError>("specify either \"--status\", \"--peers\","
                            " \"--redundancy\", or \"--all\"");
    }
#endif

    /*-------------.
    | Mode: link.  |
    `-------------*/

    void
    Network::mode_link(std::string const& network_name,
                       Strings const& silos_names,
                       boost::optional<std::string> const& output_name,
                       boost::optional<std::string> const& node_id)
    {
      ELLE_TRACE_SCOPE("link");
      auto& cli = this->cli();
      auto& memo = cli.backend();
      auto owner = cli.as_user();

      {
        auto network = memo.network_get(network_name, owner, false);
        if (network.model)
          elle::err("%s is already linked with %s", network.name, owner.name);
      }

      auto silo = make_silo_config(memo, silos_names);
      auto desc = memo.network_descriptor_get(network_name, owner);
      auto const passport = [&] () -> memo::Memo::Passport
        {
          if (owner.public_key == desc.owner)
            return {owner.public_key, desc.name,
                    elle::cryptography::rsa::KeyPair(owner.public_key,
                                                        owner.private_key.get())};
          try
          {
            return memo.passport_get(desc.name, owner.name);
          }
          catch (memo::MissingLocalResource const&)
          {
            elle::err("missing passport (%s: %s), "
                      "use `memo passport` to fetch or import",
                      desc.name, owner.name);
          }
        }();
      if (!passport.verify(passport.certifier() ? *passport.certifier() : desc.owner))
        elle::err("passport signature is invalid");
      if (silo && !passport.allow_storage())
        elle::err("passport does not allow storage");
      auto const id = [&]
        {
          if (node_id)
            {
              std::stringstream ss(elle::print("\"%s\"", *node_id));
              namespace json = elle::serialization::json;
              return json::deserialize<memo::model::Address>(ss, false);
            }
          else
            return memo::model::Address::random(0); // FIXME
        }();
      auto network = memo::Network(
        desc.name,
        std::make_unique<dnut::Configuration>(
          id,
          std::move(desc.consensus),
          std::move(desc.overlay),
          std::move(silo),
          owner.keypair(),
          std::make_shared<elle::cryptography::rsa::PublicKey>(desc.owner),
          std::move(passport),
          owner.name,
          boost::optional<int>(),
          desc.version,
          desc.admin_keys,
          desc.peers,
          desc.tcp_heartbeat,
          desc.encrypt_options),
        desc.description);
      if (output_name)
        memo.save(*cli.get_output(output_name), network, false);
      else
      {
        memo.network_save(owner, network, true);
        cli.report_action("linked", "device to network", desc.name);
      }
    }

    /*-------------.
    | Mode: list.  |
    `-------------*/

    void
    Network::mode_list()
    {
      ELLE_TRACE_SCOPE("list");

#ifndef MEMO_PRODUCTION_BUILD
      {
        auto crash = memo::getenv("CRASH", ""s);
        if (crash == "assert")
          ELLE_ASSERT_EQ(42, 51);
        else if (!crash.empty())
          *(volatile int*)nullptr = 0;
      }
#endif

      auto& cli = this->cli();
      auto& memo = cli.backend();
      auto owner = cli.as_user();
      if (cli.script())
      {
        auto const l = elle::json::Json(
          memo.networks_get(owner) |
          transformed(
            [&] (auto const& network) {
              auto res = elle::json::Json
                {
                  {"name", static_cast<std::string>(network.name)},
                  {"linked", bool(network.model) && network.user_linked(owner)},
                };
              if (network.description)
                res["description"] = network.description.get();
              return res;
            }));
        elle::json::write(std::cout, l);
      }
      else
        for (auto const& network: memo.networks_get(owner))
        {
          std::cout << network.name;
          if (network.description)
            std::cout << " \"" << network.description.get() << "\"";
          if (network.model && network.user_linked(owner))
            std::cout << ": linked";
          else
            std::cout << ": not linked";
          std::cout << std::endl;
        }
    }

    /*----------------------.
    | Mode: list_services.  |
    `----------------------*/

    namespace
    {
      /// Action to run by network_run.
      using Action =
        std::function<void (memo::User& owner,
                            memo::Network& network,
                            dnut::Doughnut& dht,
                            bool push)>;
      void
      network_run(Memo& cli,
                  std::string const& network_name,
#ifndef ELLE_WINDOWS
                  bool daemon = false,
                  bool monitoring = true,
#endif
                  Strings peer = {},
                  bool async = false,
                  bool cache = false,
                  boost::optional<int> cache_ram_size = {},
                  boost::optional<int> cache_ram_ttl = {},
                  boost::optional<int> cache_ram_invalidation = {},
                  boost::optional<uint64_t> cache_disk_size = {},
                  bool fetch = false,
                  bool push = false,
                  bool publish = false,
                  boost::optional<std::string> const& endpoints_file = {},
                  boost::optional<std::string> const& port_file = {},
                  boost::optional<int> port = {},
                  boost::optional<std::string> const& peers_file = {},
                  boost::optional<std::string> listen = {},
                  boost::optional<int> fetch_endpoints_interval = {},
                  bool no_local_endpoints = false,
                  bool no_public_endpoints = false,
                  Strings advertise_host = {},
                  boost::optional<std::string> grpc = {},
                  boost::optional<std::string> grpc_port_file = {},
                  boost::optional<bool> paxos_rebalancing_auto_expand = {},
                  boost::optional<bool> paxos_rebalancing_inspect = {},
                  boost::optional<bool> resign_on_shutdown = {},
                  Action const& action = {})
      {
        auto& memo = cli.backend();
        auto owner = cli.as_user();
        auto network = memo.network_get(network_name, owner);
        // Use the qualified name, in case the user is running the
        // network several times concurrently under different ids.
        // So it can be `bob/infinit/company`, or `bob/bob/bobnet`.
        main_log_family(elle::print("%s/%s", owner.name, network.name));
        if (paxos_rebalancing_auto_expand || paxos_rebalancing_inspect)
        {
          auto paxos = dynamic_cast<
            dnut::consensus::Paxos::Configuration*>(
              network.dht()->consensus.get());
          if (!paxos)
            elle::err<CLIError>("paxos options on non-paxos consensus");
          if (paxos_rebalancing_auto_expand)
            paxos->rebalance_auto_expand(*paxos_rebalancing_auto_expand);
          if (paxos_rebalancing_inspect)
            paxos->rebalance_inspect(*paxos_rebalancing_inspect);
        }
        network.ensure_allowed(owner, "run");
        cache |= (cache_ram_size || cache_ram_ttl
                  || cache_ram_invalidation || cache_disk_size);
        auto const listen_address
          = listen
          ? boost::asio::ip::address::from_string(*listen)
          : boost::optional<boost::asio::ip::address>{};
        auto dht = network.run(
          owner,
          false,
          cache,
          cache_ram_size, cache_ram_ttl, cache_ram_invalidation,
          // memo._network_cache_dir(network_name, owner),
          async, cache_disk_size, cli.compatibility_version(), port,
          listen_address, resign_on_shutdown
#ifndef ELLE_WINDOWS
          , monitoring
#endif
        );
        hook_stats_signals(*dht);
        elle::reactor::Thread::unique_ptr grpc_thread;
        int grpc_port = -1;
        if (grpc)
        {
          grpc_thread.reset(
            new elle::reactor::Thread(
              "memo_vs_grpc_server",
              [dht=dht.get(), grpc, &grpc_port]
              {
                memo::grpc::serve_memo_vs(*dht, *grpc, &grpc_port);
              }));
          if (grpc_port_file)
          {
            while (grpc_port == -1)
              elle::reactor::sleep(50ms);
            port_to_file(grpc_port, *grpc_port_file);
          }
        }
        if (peers_file)
        {
          auto more_peers = hook_peer_discovery(*dht, *peers_file);
          ELLE_TRACE("Peer list file got %s peers", more_peers.size());
          if (!more_peers.empty())
            dht->overlay()->discover(more_peers);
        }
        if (!peer.empty())
          dht->overlay()->discover(parse_peers(peer));
        // Only push if we have are contributing storage.
        bool push_p = (push || publish) && dht->local() && dht->local()->storage();
        if (!dht->local() && push_p)
          elle::err("network %s is client only since no storage is attached", name);
        if (dht->local())
        {
          if (port_file)
            port_to_file(dht->local()->server_endpoint().port(), *port_file);
          if (endpoints_file)
            endpoints_to_file(dht->local()->server_endpoints(), *endpoints_file);
        }
#ifndef ELLE_WINDOWS
        auto daemon_handle = daemon ? daemon_hold(0, 1) : daemon_invalid;
#endif
        auto run = [&, push_p]
          {
            elle::reactor::Thread::unique_ptr poll_thread;
            if (fetch || publish)
            {
              memo::model::NodeLocations eps;
              network.hub_fetch_endpoints(eps);
              dht->overlay()->discover(eps);
              if (fetch_endpoints_interval && *fetch_endpoints_interval > 0)
                poll_thread =
                  network.make_poll_hub_thread(*dht, eps,
                                                  *fetch_endpoints_interval);
            }
#ifndef ELLE_WINDOWS
            if (daemon)
            {
              ELLE_TRACE("releasing daemon");
              daemon_release(daemon_handle);
            }
#endif
            action(owner, network, *dht, push_p);
          };
        if (push_p)
          elle::With<InterfacePublisher>(
            memo,
            network, owner, dht->id(),
            dht->local()->server_endpoint().port(),
            advertise_host,
            no_local_endpoints,
            no_public_endpoints) << [&]
            {
              run();
            };
        else
          run();
      }
    }

    void
    Network::mode_list_services(std::string const& network_name,
                                Strings peer,
                                bool async,
                                bool cache,
                                boost::optional<int> cache_ram_size,
                                boost::optional<int> cache_ram_ttl,
                                boost::optional<int> cache_ram_invalidation,
                                boost::optional<uint64_t> cache_disk_size,
                                bool fetch_endpoints,
                                bool fetch,
                                bool push_endpoints,
                                bool push,
                                bool publish,
                                boost::optional<std::string> const& endpoints_file,
                                boost::optional<std::string> const& port_file,
                                boost::optional<int> port,
                                boost::optional<std::string> const& peers_file,
                                boost::optional<std::string> listen,
                                boost::optional<int> fetch_endpoints_interval,
                                bool no_local_endpoints,
                                bool no_public_endpoints,
                                Strings advertise_host)
    {
      ELLE_TRACE_SCOPE("list_services");
      auto& cli = this->cli();
      network_run
        (
         cli,
         network_name,
#ifndef ELLE_WINDOWS
         // Passing explicitly the default values.  This is not nice.
         false,
         true,
#endif
         peer,
         async,
         cache,
         cache_ram_size,
         cache_ram_ttl,
         cache_ram_invalidation,
         cache_disk_size,
         fetch || fetch_endpoints,
         push || push_endpoints,
         publish,
         endpoints_file,
         port_file,
         port,
         peers_file,
         listen,
         fetch_endpoints_interval,
         no_local_endpoints,
         no_public_endpoints,
         advertise_host,
         {}, // grpc
         {}, // grpc_port_file
         {}, // paxos_rebalancing_auto_expand
         {}, // paxos_rebalancing_inspect
         {}, // resign_on_shutdown
         [&] (memo::User& owner,
              memo::Network& network,
              dnut::Doughnut& dht,
              bool /*push*/)
         {
           auto services = dht.services();
           if (cli.script())
           {
             auto res = std::unordered_map<std::string, Strings>{};
             for (auto const& type: services)
               res.emplace(type.first,
                           elle::make_vector(type.second,
                                             [](auto const& service)
                                             {
                                               return service.first;
                                             }));
             elle::serialization::json::serialize(res, std::cout, false);
           }
           else
           {
             for (auto const& type: services)
             {
               std::cout << type.first << ":" << std::endl;
               for (auto const& service: type.second)
                 std::cout << "  " << service.first << std::endl;
             }
           }
         });
    }

    /*-------------------.
    | Mode: list_silos.  |
    `-------------------*/

    void
    Network::mode_list_silos(std::string const& network_name)
    {
      ELLE_TRACE_SCOPE("list_silos");
      auto& cli = this->cli();
      auto& memo = cli.backend();
      auto owner = cli.as_user();

      auto network = memo.network_get(network_name, owner, true);
      // XXX[Silo]: Network::modem::storage
      if (network.model->storage)
      {
        if (auto strip = dynamic_cast<memo::silo::StripSiloConfig*>(
            network.model->storage.get()))
          for (auto const& s: strip->storage)
            std::cout << s->name << "\n";
        else
          std::cout << network.model->storage->name;
        std::cout << std::endl;
      }
    }

    /*-------------.
    | Mode: pull.  |
    `-------------*/

    void
    Network::mode_pull(std::string const& network_name,
                       bool purge)
    {
      ELLE_TRACE_SCOPE("pull");
      auto& cli = this->cli();
      auto& memo = cli.backend();
      auto owner = cli.as_user();
      auto name = memo.qualified_name(network_name, owner);
      memo.hub_delete("network", name, owner, false, purge);
    }

    /*-------------.
    | Mode: push.  |
    `-------------*/

    void
    Network::mode_push(std::string const& network_name)
    {
      ELLE_TRACE_SCOPE("push");
      auto& cli = this->cli();
      auto& memo = cli.backend();
      auto owner = cli.as_user();

      auto network = memo.network_get(network_name, owner);
      {
        auto& dht = *network.dht();
        auto owner_uid = memo::User::uid(*dht.owner);
        auto desc = memo::NetworkDescriptor(std::move(network));
        memo.hub_push("network", desc.name, desc, owner, false, true);
      }
    }

    /*------------.
    | Mode: run.  |
    `------------*/

    namespace
    {
      /// An auxiliary struct to make reports.
      ///
      /// Respond
      ///   ("success", true)
      ///   ("value", block);
      ///
      /// serializes to std::cout.
      struct Respond
      {
        template <typename T>
        Respond(std::string const& k, T&& v)
          : response(std::cout, false, true)
        {
          response.serialize(k, std::forward<T>(v));
        }

        template <typename T>
        Respond&
        operator()(std::string const& k, T&& v)
        {
          response.serialize(k, std::forward<T>(v));
          return *this;
        }
        elle::serialization::json::SerializerOut response;
      };
    }

    void
    Network::mode_run(std::string const& network_name,
                      boost::optional<std::string> const& commands,
#ifndef ELLE_WINDOWS
                      bool daemon,
                      bool monitoring,
#endif
                      Strings peer,
                      bool async,
                      bool cache,
                      boost::optional<int> cache_ram_size,
                      boost::optional<int> cache_ram_ttl,
                      boost::optional<int> cache_ram_invalidation,
                      boost::optional<uint64_t> cache_disk_size,
                      bool fetch_endpoints,
                      bool fetch,
                      bool push_endpoints,
                      bool push,
                      bool publish,
                      boost::optional<std::string> const& endpoints_file,
                      boost::optional<std::string> const& port_file,
                      boost::optional<int> port,
                      boost::optional<std::string> const& peers_file,
                      boost::optional<std::string> listen,
                      boost::optional<int> fetch_endpoints_interval,
                      bool no_local_endpoints,
                      bool no_public_endpoints,
                      Strings advertise_host,
                      boost::optional<std::string> grpc,
                      boost::optional<std::string> const& grpc_port_file,
                      boost::optional<bool> paxos_rebalancing_auto_expand,
                      boost::optional<bool> paxos_rebalancing_inspect,
                      boost::optional<bool> resign_on_shutdown)
    {
      ELLE_TRACE_SCOPE("run");
      auto& cli = this->cli();
      network_run
        (
         cli,
         network_name,
#ifndef ELLE_WINDOWS
         daemon,
         monitoring,
#endif
         peer,
         async,
         cache,
         cache_ram_size,
         cache_ram_ttl,
         cache_ram_invalidation,
         cache_disk_size,
         fetch || fetch_endpoints,
         push || push_endpoints,
         publish,
         endpoints_file,
         port_file,
         port,
         peers_file,
         listen,
         fetch_endpoints_interval,
         no_local_endpoints,
         no_public_endpoints,
         advertise_host,
         grpc,
         grpc_port_file,
         paxos_rebalancing_auto_expand,
         paxos_rebalancing_inspect,
         resign_on_shutdown,
         [&] (memo::User& owner,
              memo::Network& network,
              dnut::Doughnut& dht,
              bool push)
         {
           auto stat_thread = elle::reactor::Thread::unique_ptr{};
           if (push)
             stat_thread = network.make_stat_update_thread(cli.backend(), owner, dht);
           cli.report_action("running", "network", network.name);
           if (cli.script())
           {
             auto input = commands_input(commands);
             while (true)
             {
               try
               {
                 auto const json = elle::json::read(*input);
                 auto command
                   = elle::serialization::json::SerializerIn(json, false);
                 command.set_context<dnut::Doughnut*>(&dht);
                 auto const op = command.deserialize<std::string>("operation");
                 if (op == "fetch")
                 {
                   auto const address =
                     command.deserialize<memo::model::Address>("address");
                   auto const block = dht.fetch(address);
                   ELLE_ASSERT(block);
                   Respond
                     ("success", true)
                     ("value", block);
                 }
                 else if (op == "create_mutable")
                 {
                   auto block = dht.make_mutable_block();
                   auto addr = block->address();
                   dht.insert(std::move(block));
                   Respond
                     ("success", true)
                     ("address", addr);
                 }
                 else if (op == "insert" || op == "update")
                 {
                   auto block = command.deserialize<
                     std::unique_ptr<memo::model::blocks::Block>>("value");
                   if (!block)
                     elle::err("missing field: value");
                   if (op == "insert")
                     dht.insert(std::move(block));
                   else
                     dht.update(std::move(block));
                   Respond("success", true);
                 }
                 else if (op == "write_immutable")
                 {
                   auto block = dht.make_block<memo::model::blocks::ImmutableBlock>(
                     elle::Buffer(command.deserialize<std::string>("data")));
                   auto const addr = block->address();
                   dht.insert(std::move(block));
                   Respond
                     ("success", true)
                     ("address", addr);
                 }
                 else if (op == "read")
                 {
                   auto const block = dht.fetch(
                       command.deserialize<memo::model::Address>("address"));
                   int const version = [&]{
                     if (auto mb =
                         dynamic_cast<memo::model::blocks::MutableBlock*>(block.get()))
                       return mb->version();
                     else
                       return -1;
                   }();
                   Respond
                      ("success", true)
                      ("data", block->data().string())
                      ("version", version);
                 }
                 else if (op == "update_mutable")
                 {
                   auto const addr = command.deserialize<memo::model::Address>("address");
                   auto const version = command.deserialize<int>("version");
                   auto const data = command.deserialize<std::string>("data");
                   auto block = dht.fetch(addr);
                   auto& mb = dynamic_cast<memo::model::blocks::MutableBlock&>(*block);
                   if (mb.version() >= version)
                     elle::err("Current version is %s", mb.version());
                   mb.data(elle::Buffer(data));
                   dht.update(std::move(block));
                   Respond("success", true);
                 }
                 else if (op == "resolve_named")
                 {
                   auto const name = command.deserialize<std::string>("name");
                   bool const create = command.deserialize<bool>("create_if_missing");
                   auto const addr = dnut::NB::address(*dht.owner(),
                     name, dht.version());
                   auto res = [&] {
                     try
                     {
                       auto const block = dht.fetch(addr);
                       auto& nb = dynamic_cast<dnut::NB&>(*block);
                       return memo::model::Address::from_string(nb.data().string());
                     }
                     catch (memo::model::MissingBlock const& mb)
                     {
                       if (!create)
                         elle::err("NB %s does not exist", name);
                       auto ab = dht.make_block<memo::model::blocks::ACLBlock>();
                       auto const addr = ab->address();
                       dht.insert(std::move(ab));
                       auto nb = dnut::NB(dht, name, elle::print("%s", addr));
                       dht.seal_and_insert(nb);
                       return addr;
                     }
                   }();
                   Respond
                     ("success", true)
                     ("address", res);
                 }
                 else if (op == "remove")
                 {
                   auto const addr = command.deserialize<memo::model::Address>("address");
                   dht.remove(addr);
                   Respond("success", true);
                 }
                 else
                   elle::err("invalid operation: %s", op);
               }
               catch (elle::Error const& e)
               {
                 if (input->eof())
                   return;
                 Respond
                   ("success", false)
                   ("message", e.what());
               }
             }
           }
           else
             elle::reactor::sleep();
         });
    }

    /*-------------.
    | Mode: stat.  |
    `-------------*/

    void
    Network::mode_stat(std::string const& network_name)
    {
      ELLE_TRACE_SCOPE("stat");
      auto& cli = this->cli();
      auto& memo = cli.backend();
      auto owner = cli.as_user();
      auto name = memo.qualified_name(network_name, owner);
      auto res =
        memo.hub_fetch<memo::Storages>(
          elle::print("networks/%s/stat", name),
          "stat",
          "stat",
          boost::none,
          memo::Headers());

      // FIXME: write Storages::operator(std::ostream&)
      elle::print(std::cout, "{\"usage\": %s, \"capacity\": %s}",
                  res.usage, res.capacity);
    }

    /*---------------.
    | Mode: unlink.  |
    `---------------*/

    void
    Network::mode_unlink(std::string const& network_name)
    {
      ELLE_TRACE_SCOPE("unlink");
      auto& cli = this->cli();
      auto& memo = cli.backend();
      auto owner = cli.as_user();
      auto network = memo.network_get(network_name, owner, true);
      if (!network.model)
        elle::err("%s is not linked", network.name);
      boost::optional<std::string> node_id;
      if (auto model = dynamic_cast<dnut::Configuration*>(network.model.get()))
      {
        std::stringstream ss;
        {
          elle::serialization::json::SerializerOut s(ss);
          s.serialize_forward(model->id);
        }
        auto tmp = ss.str();
        boost::trim(tmp);
        boost::erase_all(tmp, "\"");
        node_id = tmp;
      }
      memo.network_unlink(network.name, owner);
      if (node_id)
        cli.report("if you relink this network, use \"--node-id %s\"", node_id);
    }

    /*---------------.
    | Mode: update.  |
    `---------------*/

    namespace
    {
      std::pair<elle::cryptography::rsa::PublicKey, bool>
      user_key(memo::Memo& memo,
               std::string name,
               boost::optional<std::string> const& mountpoint)
      {
        bool is_group = false;
        if (!name.empty() && name[0] == '@')
        {
          is_group = true;
          name = name.substr(1);
        }
        if (!name.empty() && name[0] == '{')
        {
          elle::Buffer buf(name);
          elle::IOStream is(buf.istreambuf());
          auto key = elle::serialization::json::deserialize
            <elle::cryptography::rsa::PublicKey>(is);
          return std::make_pair(key, is_group);
        }
        if (!is_group)
          return std::make_pair(memo.user_get(name).public_key, false);
        if (!mountpoint)
          elle::err("A mountpoint is required to fetch groups.");
        char buf[32768];
        int res = get_xattr(*mountpoint,
                            "infinit.group.control_key." + name,
                            buf, 16384, true);
        if (res <= 0)
          elle::err("Unable to fetch group %s", name);
        auto b = elle::Buffer(buf, res);
        elle::IOStream is(b.istreambuf());
        auto key = elle::serialization::json::deserialize
          <elle::cryptography::rsa::PublicKey>(is);
        return std::make_pair(key, is_group);
      }
    }

    void
    Network::mode_update(
      std::string const& network_name,
      boost::optional<std::string> const& description,
      boost::optional<int> port,
      boost::optional<std::string> const& output_name,
      bool push_network,
      bool push,
      Strings const& admin_r,
      Strings const& admin_rw,
      Strings const& admin_remove,
      boost::optional<std::string> const& mountpoint,
      Strings const& peer,
      boost::optional<std::string> const& protocol)
    {
      ELLE_TRACE_SCOPE("create");
      auto& cli = this->cli();
      auto& memo = cli.backend();
      auto owner = cli.as_user();
      auto network = memo.network_get(network_name, owner);
      if (description)
        network.description = description;
      network.ensure_allowed(owner, "update");
      auto& dht = *network.dht();
      if (port)
        dht.port = *port;
      if (cli.compatibility_version())
        dht.version = *cli.compatibility_version();
      bool changed_admins = false;
      auto check_group_mount = [&] (bool group)
        {
          if (group && !mountpoint)
            elle::err<CLIError>("must specify mountpoint of volume on "
                                "network \"%s\" to edit group admins",
                                network.name);
        };
      auto add_admin = [&keys = dht.admin_keys, &changed_admins]
        (elle::cryptography::rsa::PublicKey const& key,
         bool group, bool read, bool write)
        {
          if (read && !write)
            push_back_if_missing(group ? keys.group_r : keys.r, key);
          if (write) // Implies RW.
            push_back_if_missing(group ? keys.group_w : keys.w, key);
          changed_admins = true;
        };
      for (auto const& u: admin_r)
      {
        auto const r = user_key(memo, u, mountpoint);
        check_group_mount(r.second);
        add_admin(r.first, r.second, true, false);
      }
      for (auto const& u: admin_rw)
      {
        auto const r = user_key(memo, u, mountpoint);
        check_group_mount(r.second);
        add_admin(r.first, r.second, true, true);
      }
      for (auto const& u: admin_remove)
      {
        auto const r = user_key(memo, u, mountpoint);
        check_group_mount(r.second);
        auto del = [&r](auto& cont)
          {
            boost::remove_erase(cont, r.first);
          };
        del(dht.admin_keys.r);
        del(dht.admin_keys.w);
        del(dht.admin_keys.group_r);
        del(dht.admin_keys.group_w);
        changed_admins = true;
      }
      if (!peer.empty())
        dht.peers = parse_peers(peer);
      if (protocol)
        dht.overlay->rpc_protocol = protocol_get(protocol);
      if (output_name)
      {
        auto output = cli.get_output(output_name);
        elle::serialization::json::serialize(network, *output, false);
      }
      else
        memo.network_save(owner, network, true);
      auto desc = std::make_unique<memo::NetworkDescriptor>(std::move(network));
      if (push || push_network)
        memo.hub_push("network", desc->name, *desc, owner, false, true);
      if (changed_admins && !output_name)
        std::cout
          << "INFO: Changes to network admins do not affect existing data:\n"
          << "INFO: Admin access will be updated on the next write to each\n"
          << "INFO: file or folder.\n";
    }
  }
}

max_stars_repo_name: infinit/memo

max_issues_repo_name: infinit/memo

max_forks_repo_name: infinit/memo


text:
<%
def includes(n)
    return <<-EOS
        #include <boost/hana/ext/boost/tuple.hpp>
        #include <boost/tuple/tuple.hpp>
    EOS
end

def list(xs_t, xs)
  return "boost::make_tuple(#{xs.join(', ')})"
end
%>

ext: hpp

max_stars_repo_name: rbock/hana

max_issues_repo_name: rbock/hana

max_forks_repo_name: rbock/hana


text:
//
// Copyright 2022 Autodesk
//
// Licensed 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 "wrapSparseValueWriter.h"

#include <boost/python/class.hpp>

using namespace boost::python;

PXR_NAMESPACE_USING_DIRECTIVE;

void wrapSparseValueWriter()
{
    class_<MayaUsdLibSparseValueWriter>("SparseValueWriter", no_init)
        .def("SetAttribute", &MayaUsdLibSparseValueWriter::SetAttribute);
}

ext: cpp

max_stars_repo_name: sun-frog/maya-usd

max_issues_repo_name: sun-frog/maya-usd

max_forks_repo_name: sun-frog/maya-usd


text:
//
// server_base.hpp
// web_server
// created by changkun at changkun.de
// https://github.com/changkun/modern-cpp-tutorial/
//

#ifndef SERVER_BASE_HPP
#define SERVER_BASE_HPP

#include <boost/asio.hpp>

#include <regex>
#include <unordered_map>
#include <thread>

namespace Web {
    struct Request {
        // request method, POST, GET; path; HTTP version
        std::string method, path, http_version;
        // use smart pointer for reference counting of content
        std::shared_ptr<std::istream> content;
        // hash container, key-value dict
        std::unordered_map<std::string, std::string> header;
        // use regular expression for path match
        std::smatch path_match;
    };

    // use typedef simplify resource type
    typedef std::map<std::string, std::unordered_map<std::string,
    std::function<void(std::ostream&, Request&)>>> resource_type;

    // socket_type is HTTP or HTTPS
    template <typename socket_type>
    class ServerBase {
    public:
        resource_type resource;
        resource_type default_resource;

        // construct server, initalize port, default: 1 thread
        ServerBase(unsigned short port, size_t num_threads = 1) :
            endpoint(boost::asio::ip::tcp::v4(), port),
            acceptor(m_io_service, endpoint),
            num_threads(num_threads) {}

        void start() {
            // default resource in the end of vector, as response method
            for(auto it = resource.begin(); it != resource.end(); it++) {
                all_resources.push_back(it);
            }
            for(auto it = default_resource.begin(); it != default_resource.end(); it++) {
                all_resources.push_back(it);
            }

            // socket connection
            accept();

            // if num_threads>1, then m_io_service.run()
            // it will start (num_threads-1) threads as thread pool
            for(size_t c = 1;c < num_threads; c++) {
                threads.emplace_back([this](){
                    m_io_service.run();
                });
            }

            // main thread
            m_io_service.run();

            // wait for other threads finish
            for(auto& t: threads)
                t.join();
        }
    protected:
        // io_service is a dispatcher in asio library, all asynchronous io events are dispatched by it
        // in another word, constructor of IO object need a io_service object as parameter
        boost::asio::io_service m_io_service;
        // IP address, port and protocol version as a endpoint, and generated on serverside
        // tcp::acceptor object, wait for connection
        boost::asio::ip::tcp::endpoint endpoint;
        // thus, a acceptor object requires io_service and endpoint as parameters
        boost::asio::ip::tcp::acceptor acceptor;

        // server side threads
        size_t num_threads;
        std::vector<std::thread> threads;

        // all resource will be append to the end of vector, and created in start()
        std::vector<resource_type::iterator> all_resources;

        // requires to implement this method for different type of server
        virtual void accept() {}

        void process_request_and_respond(std::shared_ptr<socket_type> socket) const {
            // created cache for async_read_untile()
            // shared_ptr will use for passing object to anonymous function
            // the type will be deduce as std::shared_ptr<boost::asio::streambuf>
            auto read_buffer = std::make_shared<boost::asio::streambuf>();

            boost::asio::async_read_until(*socket, *read_buffer, "\r\n\r\n",
            [this, socket, read_buffer](const boost::system::error_code& ec, size_t bytes_transferred) {
                if(!ec) {
                    // Note: read_buffer->size() always equal to bytes_transferred, the document of Boost indicates:
                    // after async_read_until operation, streambuf contains some extra data out of delimiter
                    // thus, the best way is to read and parse the content from the left of read_buffer, then append the content of async_read
                    size_t total = read_buffer->size();

                    // convert istream to string-lines
                    std::istream stream(read_buffer.get());

                    // deduce the type of std::shared_ptr<Request>
                    auto request = std::make_shared<Request>();
                    *request = parse_request(stream);

                    size_t num_additional_bytes = total-bytes_transferred;

                    // if satisfy then also read
                    if(request->header.count("Content-Length")>0) {
                        boost::asio::async_read(*socket, *read_buffer,
                        boost::asio::transfer_exactly(stoull(request->header["Content-Length"]) - num_additional_bytes),
                        [this, socket, read_buffer, request](const boost::system::error_code& ec, size_t bytes_transferred) {
                            if(!ec) {
                                // pointer as istream object stored in read_buffer
                                request->content = std::shared_ptr<std::istream>(new std::istream(read_buffer.get()));
                                respond(socket, request);
                            }
                        });
                    } else {
                        respond(socket, request);
                    }
                }
            });
        }

        Request parse_request(std::istream& stream) const {
            Request request;

            std::regex e("^([^ ]*) ([^ ]*) HTTP/([^ ]*)$");

            std::smatch sub_match;

            // read method, path and http version from the frist line
            std::string line;
            getline(stream, line);
            line.pop_back();
            if(std::regex_match(line, sub_match, e)) {
                request.method       = sub_match[1];
                request.path         = sub_match[2];
                request.http_version = sub_match[3];

                bool matched;
                e="^([^:]*): ?(.*)$";
                // parse head information
                do {
                    getline(stream, line);
                    line.pop_back();
                    matched=std::regex_match(line, sub_match, e);
                    if(matched) {
                        request.header[sub_match[1]] = sub_match[2];
                    }

                } while(matched==true);
            }
            return request;
        }

        void respond(std::shared_ptr<socket_type> socket, std::shared_ptr<Request> request) const {
            // response after search requested path and method
            for(auto res_it: all_resources) {
                std::regex e(res_it->first);
                std::smatch sm_res;
                if(std::regex_match(request->path, sm_res, e)) {
                    if(res_it->second.count(request->method)>0) {
                        request->path_match = move(sm_res);

                        // will be deduce to std::shared_ptr<boost::asio::streambuf>
                        auto write_buffer = std::make_shared<boost::asio::streambuf>();
                        std::ostream response(write_buffer.get());
                        res_it->second[request->method](response, *request);

                        // capture write_buffer in lambda, make sure it can be destroyed after async_write
                        boost::asio::async_write(*socket, *write_buffer,
                        [this, socket, request, write_buffer](const boost::system::error_code& ec, size_t bytes_transferred) {
                            // HTTP 1.1 connection
                            if(!ec && stof(request->http_version)>1.05)
                                process_request_and_respond(socket);
                        });
                        return;
                    }
                }
            }
        }

    };

    template<typename socket_type>
    class Server : public ServerBase<socket_type> {};
}
#endif /* SERVER_BASE_HPP */

max_stars_repo_name: Weissle/modern-cpp-tutorial

max_issues_repo_name: RhmpzYYX/modern-cpp-tutorial

max_forks_repo_name: RhmpzYYX/modern-cpp-tutorial


text:
/*
 *  Created on: Aug, 08, 2014
 *      Author: Thomas Faeulhammer
 */
#include <iostream>
#include <fstream>
#include <boost/program_options.hpp>
#include <pcl/io/pcd_io.h>
#include <pcl/visualization/pcl_visualizer.h>
#include <v4r/recognition/source.h>
#include <v4r/common/pcl_utils.h>
#include <v4r/common/miscellaneous.h>
#include <v4r/common/zbuffering.h>
#include <v4r/common/pcl_visualization_utils.h>
#include <v4r/io/filesystem.h>
#include <v4r/io/eigen.h>
#include <time.h>
#include <stdlib.h>

namespace po = boost::program_options;

//-g /media/Data/datasets/TUW/annotations -o /tmp/bla -s /media/Data/datasets/TUW/test_set_static/ -m /media/Data/datasets/TUW/models -t 0.01 -v
//-g /media/Data/datasets/willow/annotations -o /tmp/bla -s /media/Data/datasets/willow_dataset_new/willow_large_dataset -m /media/Data/datasets/willow_dataset/models -t 0.01 -v

namespace v4r
{

template<typename PointT>
class PcdGtAnnotator
{
public:
    class View{
    public:
        typename pcl::PointCloud<PointT>::Ptr cloud_;
        std::vector< std::string > model_id_;
        std::vector<Eigen::Matrix4f, Eigen::aligned_allocator<Eigen::Matrix4f> > transform_to_scene_;
        View()
        {
            cloud_.reset(new pcl::PointCloud<PointT>());
        }
    };

private:
    typename pcl::PointCloud<PointT>::Ptr reconstructed_scene_;
    typename v4r::Source<PointT>::Ptr source_;
    pcl::visualization::PCLVisualizer::Ptr vis_;
    std::vector<int> viewports_;

    std::vector< std::vector<bool> > visible_model_points_;
    std::vector< std::string > model_id_;
    std::vector<Eigen::Matrix4f, Eigen::aligned_allocator<Eigen::Matrix4f> > transform_to_scene_;
    std::vector<std::vector<bool> > pixel_annotated_obj_in_first_view_;

    std::vector<std::vector< std::vector<bool> > > is_pixel_annotated_for_model_in_view_; /// @brief outer loop: model id, middle loop: view id, inner loop: pixel id
    std::vector<View> views_;

public:
    std::string gt_dir_;
    std::string models_dir_;
    std::string out_dir_;
    float threshold_;
    float f_;
    bool first_view_only_;  // has been used for RA-L 16 paper, where we were interested in the visible object in the first viewpoint

    size_t unique_id_;

    PcdGtAnnotator()
    {
        f_ = 525.f;
        source_.reset(new v4r::Source<PointT>);
        models_dir_ = "";
        gt_dir_ = "";
        threshold_ = 0.01f;
        first_view_only_ = false;
        reconstructed_scene_.reset(new pcl::PointCloud<PointT>);
        out_dir_ = "/tmp/annotated_pcds";

        unique_id_ = 0;
    }

    void init_source()
    {
        source_.reset( new v4r::Source<PointT> (models_dir_) );
    }

    void annotate(const std::string &scenes_dir, const std::string & scene_id = "");
    void visualize();
    void clear();
    void save_to_disk(const std::string &path);
    static void printUsage(int argc, char ** argv);
};

template<typename PointT>
void PcdGtAnnotator<PointT>::annotate (const std::string &scenes_dir, const std::string &scene_id)
{
    std::string scene_full_path = scenes_dir + "/" + scene_id;

    std::vector < std::string > scene_view = v4r::io::getFilesInDirectory( scene_full_path, ".*.pcd", true);
    std::sort(scene_view.begin(), scene_view.end());

    if ( !scene_view.empty() )
    {
        std::cout << "Number of viewpoints in directory is:" << scene_view.size () << std::endl;

        std::string annotations_dir = gt_dir_ + "/" + scene_id;
        std::vector < std::string > gt_file = v4r::io::getFilesInDirectory( annotations_dir, ".*.txt", true);

        if( gt_file.empty())
        {
            std::cerr << "Could not find any annotations in " << annotations_dir << ". " << std::endl;
            return;
        }

        for (size_t s_id = 0; s_id < scene_view.size (); s_id++)
        {
            if (first_view_only_ && s_id > 0)
                    continue;

            std::vector<std::string> strs;
            boost::split (strs, scene_view[s_id], boost::is_any_of ("."));
            std::string scene_file_wo_ext = strs[0] + "_";
            std::string gt_occ_check = scene_file_wo_ext + "occlusion_";

            std::string scene_full_file_path = scene_full_path + "/" + scene_view[s_id];
            pcl::io::loadPCDFile(scene_full_file_path, *reconstructed_scene_);
            View view;
            pcl::copyPointCloud(*reconstructed_scene_, *view.cloud_);
            view.cloud_->sensor_origin_ = reconstructed_scene_->sensor_origin_;
            view.cloud_->sensor_orientation_ = reconstructed_scene_->sensor_orientation_;

            for(size_t gt_id=0; gt_id < gt_file.size(); gt_id++)
            {
                const std::string gt_fn = gt_file[gt_id];
                if( gt_fn.compare(0, scene_file_wo_ext.size(), scene_file_wo_ext) == 0 &&
                        gt_fn.compare(0, gt_occ_check.size(), gt_occ_check))
                {
                    std::cout << gt_fn << std::endl;
                    std::string model_instance = gt_fn.substr(scene_file_wo_ext.size());
                    std::vector<std::string> str_split;
                    boost::split (str_split, model_instance, boost::is_any_of ("."));
                    model_instance = str_split[0];

                    size_t found = model_instance.find_last_of("_");
                    std::string times_text ("times.txt");
                    if ( model_instance.compare(times_text) == 0 )
                    {
                        std::cout << "skipping this one" << std::endl;
                        continue;
                    }
                    std::string model_name = model_instance.substr(0,found);
                    std::cout << "Model: " << model_name << std::endl;

                    bool found_model;
                    typename Model<PointT>::ConstPtr pModel = source_->getModelById("", model_name, found_model);

                    if (!found_model)
                        std::cerr << "Did not find " << model_name << ". There is something wrong! " << std::endl;

                    std::string gt_full_file_path = annotations_dir + "/" + gt_fn;
                    Eigen::Matrix4f transform = v4r::io::readMatrixFromFile(gt_full_file_path);

                    typename pcl::PointCloud<PointT>::ConstPtr model_cloud = pModel->getAssembled(3);
                    typename pcl::PointCloud<PointT>::Ptr model_aligned(new pcl::PointCloud<PointT>());
                    pcl::transformPointCloud(*model_cloud, *model_aligned, transform);

                    bool model_exists = false;
                    size_t current_model_id = 0;
                    for (size_t m_id=0; m_id < model_id_.size(); m_id++)
                    {
                        if ( model_id_[m_id].compare( model_instance ) == 0)
                        {
                            model_exists = true;
                            current_model_id = m_id;
                            transform_to_scene_[m_id] = transform;
                            break;
                        }
                    }
                    if ( !model_exists )
                    {
                        std::vector<bool> visible_model_pts_tmp (model_aligned->points.size(), false);
                        visible_model_points_.push_back(visible_model_pts_tmp);
                        model_id_.push_back( model_instance );
                        transform_to_scene_.push_back(transform);
                        std::vector<bool> obj_mask (reconstructed_scene_->points.size(), false);
                        pixel_annotated_obj_in_first_view_.push_back ( obj_mask );
                        current_model_id = model_id_.size()-1;
                    }

                    const float cx = (static_cast<float> (reconstructed_scene_->width) / 2.f - 0.5f);
                    const float cy = (static_cast<float> (reconstructed_scene_->height) / 2.f - 0.5f);

                    std::vector<bool> obj_mask_in_view (reconstructed_scene_->points.size(), false);

                    for(size_t m_pt_id=0; m_pt_id < model_aligned->points.size(); m_pt_id++)
                    {
                        PointT mp = model_aligned->points[m_pt_id];
                        int u = static_cast<int> (f_ * mp.x / mp.z + cx);
                        int v = static_cast<int> (f_ * mp.y / mp.z + cy);

                        if(u<0 || u >= (int)reconstructed_scene_->width || v<0 || v >= (int)reconstructed_scene_->height) // model point outside field of view
                            continue;

                        int px_id = v*reconstructed_scene_->width + u;

                        PointT sp = reconstructed_scene_->at(u,v);

                        if ( !pcl::isFinite(sp) )   // Model point is not visible from the view point (shiny spot, noise,...)
                            continue;

                        if( std::abs(mp.z - sp.z) < threshold_ )
                        {
                            visible_model_points_[current_model_id][m_pt_id] = true;

                            obj_mask_in_view[px_id] = true;

                            if( s_id == 0 )
                                pixel_annotated_obj_in_first_view_[current_model_id][px_id] = true;
                        }
                    }

                    std::stringstream out_fn; out_fn << out_dir_ << "/" << setfill('0') << setw(10) << unique_id_ << ".txt";
                    v4r::io::createDirForFileIfNotExist(out_fn.str());
                    std::ofstream obj_indices_per_view_f ( out_fn.str().c_str() );
                    for(size_t i=0; i< obj_mask_in_view.size(); i++)
                    {
                        if(obj_mask_in_view[i])
                            obj_indices_per_view_f << i << std::endl;
                    }
                    obj_indices_per_view_f.close();

                    std::ofstream label_file;
                    label_file.open( out_dir_ + "/label.anno", std::ofstream::out | std::ofstream::app);
                    label_file << scenes_dir << " " << scene_id << " " << scene_view[s_id] << " " << setfill('0') << setw(10) << unique_id_ << " " << model_name << std::endl;
                    label_file.close();
                    unique_id_ ++;
                }
            }
            std::cout << std::endl;

            view.transform_to_scene_ = transform_to_scene_;
            views_.push_back(view);
        }
    }
    else
    {
        PCL_ERROR("You should pass a directory\n");
        return;
    }
}

template<typename PointT>
void PcdGtAnnotator<PointT>::visualize()
{
    if(!vis_)
        vis_.reset ( new pcl::visualization::PCLVisualizer ("ground truth model", true));

    std::vector<std::string> subwindow_titles;
    subwindow_titles.push_back( "scene" );

    for (size_t m_id=0; m_id < model_id_.size(); m_id++)
        subwindow_titles.push_back ( model_id_[m_id] );

    std::vector<int> viewports = v4r::pcl_visualizer::visualization_framework(*vis_, 1, model_id_.size() + 1 , subwindow_titles);

    size_t vp_id = 0;

    Eigen::Vector4f zero_origin;
    zero_origin[0] = zero_origin[1] = zero_origin[2] = zero_origin[3] = 0.f;

    reconstructed_scene_->sensor_origin_ = zero_origin;   // for correct visualization
    reconstructed_scene_->sensor_orientation_ = Eigen::Quaternionf::Identity();
    vis_->addPointCloud(reconstructed_scene_, "scene", viewports[vp_id++]);
    for (size_t m_id=0; m_id < model_id_.size(); m_id++)
    {
        size_t found = model_id_[m_id].find_last_of("_");
        std::string model_name = model_id_[m_id].substr(0,found);

        bool found_model;
        typename Model<PointT>::ConstPtr pModel = source_->getModelById("", model_name, found_model);

        if (!found_model)
            std::cerr << "Did not find " << model_name << ". There is something wrong! " << std::endl;

        typename pcl::PointCloud<PointT>::Ptr visible_model ( new pcl::PointCloud<PointT>() );
        typename pcl::PointCloud<PointT>::Ptr visible_model_aligned ( new pcl::PointCloud<PointT>() );
        typename pcl::PointCloud<PointT>::ConstPtr model_cloud = pModel->getAssembled( 3 );
        size_t num_vis_pts = 0;
        for (size_t v_id=0; v_id<visible_model_points_[m_id].size(); v_id++)
        {
            num_vis_pts++;
        }
        std::cout << num_vis_pts << " visible points of total " << visible_model_points_[m_id].size() << std::endl;
        pcl::copyPointCloud( *model_cloud, visible_model_points_[ m_id ], *visible_model);
        pcl::transformPointCloud( *visible_model, *visible_model_aligned, transform_to_scene_ [m_id]);
        vis_->addPointCloud( visible_model_aligned, model_id_[ m_id ], viewports[vp_id++]);
    }
    vis_->spin();
}

template<typename PointT>
void PcdGtAnnotator<PointT>::clear ()
{
    visible_model_points_.clear();
    model_id_.clear();
    transform_to_scene_.clear();
    views_.clear();
    reconstructed_scene_.reset(new pcl::PointCloud<PointT>);
    pixel_annotated_obj_in_first_view_.clear();
}

template<typename PointT>
void PcdGtAnnotator<PointT>::save_to_disk(const std::string &path)
{
    bf::path path_bf = path;

    if(!bf::exists(path_bf))
        bf::create_directories(path);

//    pcl::visualization::PCLVisualizer vis("obj_mask");
//    vis.addPointCloud(views_[0].cloud_, "cloud");

    for (size_t m_id=0; m_id < model_id_.size(); m_id++)
    {
        size_t found = model_id_[m_id].find_last_of("_");
        std::string model_name = model_id_[m_id].substr(0,found);

        bool found_model;
        typename Model<PointT>::ConstPtr pModel = source_->getModelById("", model_name, found_model);

        if (!found_model)
            std::cerr << "Did not find " << model_name << ". There is something wrong! " << std::endl;

        typename pcl::PointCloud<PointT>::Ptr visible_model ( new pcl::PointCloud<PointT>() );
        typename pcl::PointCloud<PointT>::Ptr visible_model_aligned ( new pcl::PointCloud<PointT>() );
        typename pcl::PointCloud<PointT>::ConstPtr model_cloud = pModel->getAssembled( 3 );
        size_t num_vis_pts = 0;
        for (size_t v_id=0; v_id<visible_model_points_[m_id].size(); v_id++)
        {
            num_vis_pts++;
        }
        std::cout << num_vis_pts << " visible points of total " << visible_model_points_[m_id].size() << std::endl;
        pcl::copyPointCloud( *model_cloud, visible_model_points_[ m_id ], *visible_model);
        pcl::transformPointCloud( *visible_model, *visible_model_aligned, transform_to_scene_ [m_id]);
        visible_model_aligned->sensor_orientation_ = views_.back().cloud_->sensor_orientation_;
        visible_model_aligned->sensor_origin_ = views_.back().cloud_->sensor_origin_;

        if(visible_model_aligned->points.empty())
            continue;

        pcl::io::savePCDFileBinary( path + "/" + model_id_[m_id] + ".pcd", *visible_model_aligned);

        if( pixel_annotated_obj_in_first_view_.size() > m_id )
        {
            std::ofstream obj_indices_f ( (path + "/" + model_id_[m_id] + "_mask.txt").c_str() );
            for(size_t i=0; i< pixel_annotated_obj_in_first_view_[m_id].size(); i++)
            {
                if(pixel_annotated_obj_in_first_view_[m_id][i])
                    obj_indices_f << i << std::endl;
            }
            obj_indices_f.close();
        }
    }
//    vis.spin();
}
}

int
main (int argc, char ** argv)
{
    srand (time(NULL));
    typedef pcl::PointXYZRGB PointT;
    v4r::PcdGtAnnotator<PointT> annotator;
    std::string scene_dir;
    bool visualize = false;

    po::options_description desc("Pixel-wise annotation of point clouds using recognition results represented as ground-truth object pose\n======================================\n **Allowed options");
    desc.add_options()
            ("help,h", "produce help message")
            ("scenes_dir,s", po::value<std::string>(&scene_dir)->required(), "directory containing the scene .pcd files")
            ("models_dir,m", po::value<std::string>(&annotator.models_dir_)->required(), "directory containing the model .pcd files")
            ("gt_dir,g", po::value<std::string>(&annotator.gt_dir_)->required(), "directory containing recognition results")
            ("output_dir,o", po::value<std::string>(&annotator.out_dir_)->default_value(annotator.out_dir_), "output directory")
            ("threshold,t", po::value<float>(&annotator.threshold_)->default_value(annotator.threshold_), "Threshold in m for a point to be counted as inlier")
            ("focal_length,f", po::value<float>(&annotator.f_)->default_value(annotator.f_), "Threshold in m for a point to be counted as inlier")
            ("visualize,v", po::bool_switch(&visualize), "turn visualization on")
            ("first_view_only", po::bool_switch(&annotator.first_view_only_), "if true, outputs the visible object model in the first view only")
    ;

    po::variables_map vm;
    po::store(po::parse_command_line(argc, argv, desc), vm);
    if (vm.count("help"))
    {
        std::cout << desc << std::endl;
        return false;
    }

    try { po::notify(vm); }
    catch(std::exception& e) { std::cerr << "Error: " << e.what() << std::endl << std::endl << desc << std::endl; return false;  }

    std::vector< std::string> sub_folder_names = v4r::io::getFoldersInDirectory( scene_dir );
    if( sub_folder_names.empty() )
    {
        std::cerr << "No subfolders in directory " << scene_dir << ". " << std::endl;
        sub_folder_names.push_back("");
    }

    std::sort(sub_folder_names.begin(), sub_folder_names.end());
    annotator.init_source();
    for (size_t sub_folder_id=0; sub_folder_id < sub_folder_names.size(); sub_folder_id++)
    {
        annotator.annotate (scene_dir, sub_folder_names[ sub_folder_id ]);

        if(visualize)
            annotator.visualize();

        annotator.save_to_disk(annotator.out_dir_ + "/" + sub_folder_names[ sub_folder_id ]);
        annotator.clear();
    }
}