text: /*
   * This file is open source software, licensed to you under the terms
   * of the Apache License, Version 2.0 (the "License").  See the NOTICE file
   * distributed with this work for additional information regarding copyright
   * ownership.  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.
   */
  /*
   * Copyright 2015 Cloudius Systems
   */
  #include <cmath>
  #include <seastar/core/reactor.hh>
  #include <seastar/core/app-template.hh>
  #include <seastar/rpc/rpc.hh>
  #include <seastar/core/sleep.hh>
  #include <seastar/rpc/lz4_compressor.hh>
  #include <seastar/util/log.hh>
  #include <seastar/core/loop.hh>
  
  using namespace seastar;
  
  struct serializer {
  };
  
  template <typename T, typename Output>
  inline
  void write_arithmetic_type(Output& out, T v) {
      static_assert(std::is_arithmetic<T>::value, "must be arithmetic type");
      return out.write(reinterpret_cast<const char*>(&v), sizeof(T));
  }
  
  template <typename T, typename Input>
  inline
  T read_arithmetic_type(Input& in) {
      static_assert(std::is_arithmetic<T>::value, "must be arithmetic type");
      T v;
      in.read(reinterpret_cast<char*>(&v), sizeof(T));
      return v;
  }
  
  template <typename Output>
  inline void write(serializer, Output& output, int32_t v) { return write_arithmetic_type(output, v); }
  template <typename Output>
  inline void write(serializer, Output& output, uint32_t v) { return write_arithmetic_type(output, v); }
  template <typename Output>
  inline void write(serializer, Output& output, int64_t v) { return write_arithmetic_type(output, v); }
  template <typename Output>
  inline void write(serializer, Output& output, uint64_t v) { return write_arithmetic_type(output, v); }
  template <typename Output>
  inline void write(serializer, Output& output, double v) { return write_arithmetic_type(output, v); }
  template <typename Input>
  inline int32_t read(serializer, Input& input, rpc::type<int32_t>) { return read_arithmetic_type<int32_t>(input); }
  template <typename Input>
  inline uint32_t read(serializer, Input& input, rpc::type<uint32_t>) { return read_arithmetic_type<uint32_t>(input); }
  template <typename Input>
  inline uint64_t read(serializer, Input& input, rpc::type<uint64_t>) { return read_arithmetic_type<uint64_t>(input); }
  template <typename Input>
  inline uint64_t read(serializer, Input& input, rpc::type<int64_t>) { return read_arithmetic_type<int64_t>(input); }
  template <typename Input>
  inline double read(serializer, Input& input, rpc::type<double>) { return read_arithmetic_type<double>(input); }
  
  template <typename Output>
  inline void write(serializer, Output& out, const sstring& v) {
      write_arithmetic_type(out, uint32_t(v.size()));
      out.write(v.c_str(), v.size());
  }
  
  template <typename Input>
  inline sstring read(serializer, Input& in, rpc::type<sstring>) {
      auto size = read_arithmetic_type<uint32_t>(in);
      sstring ret = uninitialized_string(size);
      in.read(ret.data(), size);
      return ret;
  }
  
  namespace bpo = boost::program_options;
  using namespace std::chrono_literals;
  
  class mycomp : public rpc::compressor::factory {
      const sstring _name = "LZ4";
  public:
      virtual const sstring& supported() const override {
          fmt::print("supported called\n");
          return _name;
      }
      virtual std::unique_ptr<rpc::compressor> negotiate(sstring feature, bool is_server) const override {
          fmt::print("negotiate called with {}\n", feature);
          return feature == _name ? std::make_unique<rpc::lz4_compressor>() : nullptr;
      }
  };
  
  int main(int ac, char** av) {
      app_template app;
      app.add_options()
                      ("port", bpo::value<uint16_t>()->default_value(10000), "RPC server port")
                      ("server", bpo::value<std::string>(), "Server address")
                      ("compress", bpo::value<bool>()->default_value(false), "Compress RPC traffic");
      std::cout << "start ";
      rpc::protocol<serializer> myrpc(serializer{});
      static std::unique_ptr<rpc::protocol<serializer>::server> server;
      static std::unique_ptr<rpc::protocol<serializer>::client> client;
      static double x = 30.0;
  
      static logger log("rpc_demo");
      myrpc.set_logger(&log);
  
      return app.run_deprecated(ac, av, [&] {
          auto&& config = app.configuration();
          uint16_t port = config["port"].as<uint16_t>();
          bool compress = config["compress"].as<bool>();
          static mycomp mc;
          auto test1 = myrpc.register_handler(1, [x = 0](int i) mutable { fmt::print("test1 count {:d} got {:d}\n", ++x, i); });
          auto test2 = myrpc.register_handler(2, [](int a, int b){ fmt::print("test2 got {:d} {:d}\n", a, b); return make_ready_future<int>(a+b); });
          auto test3 = myrpc.register_handler(3, [](double x){ fmt::print("test3 got {:f}\n", x); return std::make_unique<double>(sin(x)); });
          auto test4 = myrpc.register_handler(4, [](){ fmt::print("test4 throw!\n"); throw std::runtime_error("exception!"); });
          auto test5 = myrpc.register_handler(5, [](){ fmt::print("test5 no wait\n"); return rpc::no_wait; });
          auto test6 = myrpc.register_handler(6, [](const rpc::client_info& info, int x){ fmt::print("test6 client {}, {:d}\n", inet_ntoa(info.addr.as_posix_sockaddr_in().sin_addr), x); });
          auto test8 = myrpc.register_handler(8, [](){ fmt::print("test8 sleep for 2 sec\n"); return sleep(2s); });
          auto test13 = myrpc.register_handler(13, [](){ fmt::print("test13 sleep for 1 msec\n"); return sleep(1ms); });
          auto test_message_to_big = myrpc.register_handler(14, [](sstring payload){ fmt::print("test message to bit, should not get here"); });
  
          if (config.count("server")) {
              std::cout << "client" << std::endl;
              auto test7 = myrpc.make_client<long (long a, long b)>(7);
              auto test9 = myrpc.make_client<long (long a, long b)>(9); // do not send optional
              auto test9_1 = myrpc.make_client<long (long a, long b, int c)>(9); // send optional
              auto test9_2 = myrpc.make_client<long (long a, long b, int c, long d)>(9); // send more data than handler expects
              auto test10 = myrpc.make_client<long ()>(10); // receive less then replied
              auto test10_1 = myrpc.make_client<future<rpc::tuple<long, int>> ()>(10); // receive all
              auto test11 = myrpc.make_client<future<rpc::tuple<long, rpc::optional<int>>> ()>(11); // receive more then replied
              auto test12 = myrpc.make_client<void (int sleep_ms, sstring payload)>(12); // large payload vs. server limits
              auto test_nohandler = myrpc.make_client<void ()>(100000000); // non existing verb
              auto test_nohandler_nowait = myrpc.make_client<rpc::no_wait_type ()>(100000000); // non existing verb, no_wait call
              rpc::client_options co;
              if (compress) {
                  co.compressor_factory = &mc;
              }
  
              client = std::make_unique<rpc::protocol<serializer>::client>(myrpc, co, ipv4_addr{config["server"].as<std::string>()});
  
              auto f = test8(*client, 1500ms).then_wrapped([](future<> f) {
                  try {
                      f.get();
                      printf("test8 should not get here!\n");
                  } catch (rpc::timeout_error&) {
                      printf("test8 timeout!\n");
                  }
              });
              for (auto i = 0; i < 100; i++) {
                  fmt::print("iteration={:d}\n", i);
                  (void)test1(*client, 5).then([] (){ fmt::print("test1 ended\n");});
                  (void)test2(*client, 1, 2).then([] (int r) { fmt::print("test2 got {:d}\n", r); });
                  (void)test3(*client, x).then([](double x) { fmt::print("sin={:f}\n", x); });
                  (void)test4(*client).then_wrapped([](future<> f) {
                      try {
                          f.get();
                          fmt::print("test4 your should not see this!\n");
                      } catch (std::runtime_error& x){
                          fmt::print("test4 {}\n", x.what());
                      }
                  });
                  (void)test5(*client).then([] { fmt::print("test5 no wait ended\n"); });
                  (void)test6(*client, 1).then([] { fmt::print("test6 ended\n"); });
                  (void)test7(*client, 5, 6).then([] (long r) { fmt::print("test7 got {:d}\n", r); });
                  (void)test9(*client, 1, 2).then([] (long r) { fmt::print("test9 got {:d}\n", r); });
                  (void)test9_1(*client, 1, 2, 3).then([] (long r) { fmt::print("test9.1 got {:d}\n", r); });
                  (void)test9_2(*client, 1, 2, 3, 4).then([] (long r) { fmt::print("test9.2 got {:d}\n", r); });
                  (void)test10(*client).then([] (long r) { fmt::print("test10 got {:d}\n", r); });
                  (void)test10_1(*client).then([] (rpc::tuple<long, int> r) { fmt::print("test10_1 got {:d} and {:d}\n", std::get<0>(r), std::get<1>(r)); });
                  (void)test11(*client).then([] (rpc::tuple<long, rpc::optional<int> > r) { fmt::print("test11 got {:d} and {:d}\n", std::get<0>(r), bool(std::get<1>(r))); });
                  (void)test_nohandler(*client).then_wrapped([](future<> f) {
                      try {
                          f.get();
                          fmt::print("test_nohandler your should not see this!\n");
                      } catch (rpc::unknown_verb_error& x){
                          fmt::print("test_nohandle no such verb\n");
                      } catch (...) {
                          fmt::print("incorrect exception!\n");
                      }
                  });
                  (void)test_nohandler_nowait(*client);
                  auto c = make_lw_shared<rpc::cancellable>();
                  (void)test13(*client, *c).then_wrapped([](future<> f) {
                      try {
                          f.get();
                          fmt::print("test13 shold not get here\n");
                      } catch(rpc::canceled_error&) {
                          fmt::print("test13 canceled\n");
                      } catch(...) {
                          fmt::print("test13 wrong exception\n");
                      }
                  });
                  c->cancel();
                  (void)test13(*client, *c).then_wrapped([](future<> f) {
                      try {
                          f.get();
                          fmt::print("test13 shold not get here\n");
                      } catch(rpc::canceled_error&) {
                          fmt::print("test13 canceled\n");
                      } catch(...) {
                          fmt::print("test13 wrong exception\n");
                      }
                  });
                  (void)sleep(500us).then([c] { c->cancel(); });
                  (void)test_message_to_big(*client, uninitialized_string(10'000'001)).then_wrapped([](future<> f) {
                      try {
                          f.get();
                          fmt::print("test message to big shold not get here\n");
                      } catch(std::runtime_error& err) {
                          fmt::print("test message to big get error {}\n", err.what());
                      } catch(...) {
                          fmt::print("test message to big wrong exception\n");
                      }
                  });
              }
              // delay a little for a time-sensitive test
              (void)sleep(400ms).then([test12] () mutable {
                  // server is configured for 10MB max, throw 25MB worth of requests at it.
                  auto now = rpc::rpc_clock_type::now();
                  return parallel_for_each(boost::irange(0, 25), [test12, now] (int idx) mutable {
                      return test12(*client, 100, uninitialized_string(1'000'000)).then([idx, now] {
                          auto later = rpc::rpc_clock_type::now();
                          auto delta = std::chrono::duration_cast<std::chrono::milliseconds>(later - now);
                          fmt::print("idx {:d} completed after {:d} ms\n", idx, delta.count());
                      });
                  }).then([now] {
                      auto later = rpc::rpc_clock_type::now();
                      auto delta = std::chrono::duration_cast<std::chrono::milliseconds>(later - now);
                      fmt::print("test12 completed after {:d} ms (should be ~300)\n", delta.count());
                  });
              });
              (void)f.finally([] {
                  return sleep(1s).then([] {
                      return client->stop().then([] {
                          return engine().exit(0);
                      });
                  });
              });
          } else {
              std::cout << "server on port " << port << std::endl;
              myrpc.register_handler(7, [](long a, long b) mutable {
                  auto p = make_lw_shared<promise<>>();
                  auto t = make_lw_shared<timer<>>();
                  fmt::print("test7 got {:d} {:d}\n", a, b);
                  auto f = p->get_future().then([a, b, t] {
                      fmt::print("test7 calc res\n");
                      return a - b;
                  });
                  t->set_callback([p = std::move(p)] () mutable { p->set_value(); });
                  t->arm(1s);
                  return f;
              });
              myrpc.register_handler(9, [] (long a, long b, rpc::optional<int> c) {
                  long r = 2;
                  fmt::print("test9 got {:d} {:d} ", a, b);
                  if (c) {
                      fmt::print("{:d}", c.value());
                      r++;
                  }
                  fmt::print("\n");
                  return r;
              });
              myrpc.register_handler(10, [] {
                  fmt::print("test 10\n");
                  return make_ready_future<rpc::tuple<long, int>>(rpc::tuple<long, int>(1, 2));
              });
              myrpc.register_handler(11, [] {
                  fmt::print("test 11\n");
                  return 1ul;
              });
              myrpc.register_handler(12, [] (int sleep_ms, sstring payload) {
                  return sleep(std::chrono::milliseconds(sleep_ms)).then([] {
                      return make_ready_future<>();
                  });
              });
  
              rpc::resource_limits limits;
              limits.bloat_factor = 1;
              limits.basic_request_size = 0;
              limits.max_memory = 10'000'000;
              rpc::server_options so;
              if (compress) {
                  so.compressor_factory = &mc;
              }
              server = std::make_unique<rpc::protocol<serializer>::server>(myrpc, so, ipv4_addr{port}, limits);
          }
      });
  
  }
  