pong-registry.cpp 4.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125
  1. //
  2. // Copyright (c) 2019-2022 Ivan Baidakou (basiliscos) (the dot dmol at gmail dot com)
  3. //
  4. // Distributed under the MIT Software License
  5. //
  6. #include <rotor/ev.hpp>
  7. #include <rotor/registry.h>
  8. #include <iostream>
  9. #include <iomanip>
  10. #include <chrono>
  11. #include <cstdlib>
  12. #include <boost/asio/detail/winsock_init.hpp> // for calling WSAStartup on Windows
  13. /* NB: there is some race-condition, as the ponger might be register self in the registry
  14. * later then pinger asks for ponger address; then pinger receives discovery failure
  15. * reply and shuts self down.
  16. *
  17. * The correct behavior will be introduced later in rotor with erlang-like actor restart
  18. * policy, i.e.: pinger should be downed, and it's supervisor should restart if (may be
  19. * a few times, may be with some delay in between).
  20. */
  21. namespace payload {
  22. struct pong_t {};
  23. struct ping_t {
  24. using response_t = pong_t;
  25. };
  26. } // namespace payload
  27. namespace message {
  28. using ping_t = rotor::request_traits_t<payload::ping_t>::request::message_t;
  29. using pong_t = rotor::request_traits_t<payload::ping_t>::response::message_t;
  30. } // namespace message
  31. static const char ponger_name[] = "service:ponger";
  32. static const auto timeout = boost::posix_time::milliseconds{10};
  33. struct pinger_t : public rotor::actor_base_t {
  34. using rotor::actor_base_t::actor_base_t;
  35. void configure(rotor::plugin::plugin_base_t &plugin) noexcept override {
  36. rotor::actor_base_t::configure(plugin);
  37. plugin.with_casted<rotor::plugin::starter_plugin_t>([](auto &p) { p.subscribe_actor(&pinger_t::on_pong); });
  38. plugin.with_casted<rotor::plugin::registry_plugin_t>([&](auto &p) {
  39. p.discover_name(ponger_name, ponger_addr).link(true).callback([](auto phase, auto &ec) {
  40. auto p = (phase == rotor::plugin::registry_plugin_t::phase_t::linking) ? "link" : "discovery";
  41. std::cout << "executing " << p << " in accordance with ponger : " << (!ec ? "yes" : "no") << "\n";
  42. });
  43. });
  44. }
  45. void on_start() noexcept override {
  46. rotor::actor_base_t::on_start();
  47. std::cout << "lets send ping\n";
  48. request<payload::ping_t>(ponger_addr).send(timeout);
  49. }
  50. void on_pong(message::pong_t &) noexcept {
  51. std::cout << "pong received, going to shutdown\n";
  52. do_shutdown();
  53. }
  54. std::uint32_t attempts = 3;
  55. rotor::address_ptr_t ponger_addr;
  56. };
  57. struct ponger_t : public rotor::actor_base_t {
  58. using rotor::actor_base_t::actor_base_t;
  59. void set_registry_addr(const rotor::address_ptr_t &addr) { registry_addr = addr; }
  60. void configure(rotor::plugin::plugin_base_t &plugin) noexcept override {
  61. rotor::actor_base_t::configure(plugin);
  62. plugin.with_casted<rotor::plugin::starter_plugin_t>([](auto &p) { p.subscribe_actor(&ponger_t::on_ping); });
  63. plugin.with_casted<rotor::plugin::registry_plugin_t>([&](auto &p) { p.register_name(ponger_name, address); });
  64. }
  65. void on_ping(message::ping_t &req) noexcept {
  66. std::cout << "ponger received ping request\n";
  67. reply_to(req);
  68. }
  69. rotor::address_ptr_t registry_addr;
  70. };
  71. int main() {
  72. try {
  73. auto *loop = ev_loop_new(0);
  74. auto system_context = rotor::ev::system_context_ptr_t{new rotor::ev::system_context_ev_t()};
  75. auto sup = system_context->create_supervisor<rotor::ev::supervisor_ev_t>()
  76. .loop(loop)
  77. .loop_ownership(true) /* let supervisor takes ownership on the loop */
  78. .timeout(timeout)
  79. .create_registry(true)
  80. .finish();
  81. auto ponger = sup->create_actor<ponger_t>().timeout(timeout).finish();
  82. auto pinger = sup->create_actor<pinger_t>().timeout(timeout).autoshutdown_supervisor().finish();
  83. sup->start();
  84. ev_run(loop);
  85. } catch (const std::exception &ex) {
  86. std::cout << "exception : " << ex.what();
  87. }
  88. std::cout << "exiting...\n";
  89. return 0;
  90. }
  91. /*
  92. sample output:
  93. discovered & linked with ponger : yes
  94. lets send ping
  95. ponger received ping request
  96. pong received, going to shutdown
  97. exiting...
  98. */