021-ping_pong-external.cpp 6.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191
  1. //
  2. // Copyright (c) 2019-2020 Ivan Baidakou (basiliscos) (the dot dmol at gmail dot com)
  3. //
  4. // Distributed under the MIT Software License
  5. //
  6. #include "catch.hpp"
  7. #include "rotor.hpp"
  8. #include "supervisor_test.h"
  9. #include "access.h"
  10. #include <iostream>
  11. namespace r = rotor;
  12. namespace rt = r::test;
  13. struct ping_t {};
  14. struct pong_t {};
  15. struct pinger_t : public r::actor_base_t {
  16. std::uint32_t ping_sent = 0;
  17. std::uint32_t pong_received = 0;
  18. using r::actor_base_t::actor_base_t;
  19. void set_ponger_addr(const r::address_ptr_t &addr) { ponger_addr = addr; }
  20. void configure(r::plugin::plugin_base_t &plugin) noexcept override {
  21. plugin.with_casted<r::plugin::starter_plugin_t>([](auto &p) { p.subscribe_actor(&pinger_t::on_pong); });
  22. }
  23. void on_pong(r::message_t<pong_t> &) noexcept { ++pong_received; }
  24. void do_send_ping() {
  25. ++ping_sent;
  26. send<ping_t>(ponger_addr);
  27. }
  28. r::address_ptr_t ponger_addr;
  29. };
  30. struct pinger_autostart_t : public pinger_t {
  31. using pinger_t::pinger_t;
  32. void configure(r::plugin::plugin_base_t &plugin) noexcept override {
  33. pinger_t::configure(plugin);
  34. plugin.with_casted<r::plugin::starter_plugin_t>(
  35. [this](auto &p) { p.subscribe_actor(&pinger_autostart_t::on_ponger_start, ponger_addr); });
  36. }
  37. void on_ponger_start(r::message_t<r::payload::start_actor_t> &) noexcept { do_send_ping(); }
  38. };
  39. struct ponger_t : public r::actor_base_t {
  40. std::uint32_t ping_received = 0;
  41. std::uint32_t pong_sent = 0;
  42. using r::actor_base_t::actor_base_t;
  43. void set_pinger_addr(const r::address_ptr_t &addr) { pinger_addr = addr; }
  44. void configure(r::plugin::plugin_base_t &plugin) noexcept override {
  45. plugin.with_casted<r::plugin::starter_plugin_t>([](auto &p) { p.subscribe_actor(&ponger_t::on_ping); });
  46. }
  47. void on_ping(r::message_t<ping_t> &) noexcept {
  48. ++ping_received;
  49. send<pong_t>(pinger_addr);
  50. ++pong_sent;
  51. }
  52. private:
  53. r::address_ptr_t pinger_addr;
  54. };
  55. TEST_CASE("pinger & ponger on different supervisors, manually controlled", "[supervisor]") {
  56. r::system_context_t system_context;
  57. const char locality1[] = "l1";
  58. const char locality2[] = "l2";
  59. auto sup1 = system_context.create_supervisor<rt::supervisor_test_t>()
  60. .locality(locality1)
  61. .timeout(rt::default_timeout)
  62. .finish();
  63. auto sup2 = sup1->create_actor<rt::supervisor_test_t>().locality(locality2).timeout(rt::default_timeout).finish();
  64. auto pinger = sup1->create_actor<pinger_t>().timeout(rt::default_timeout).finish();
  65. auto ponger = sup2->create_actor<ponger_t>().timeout(rt::default_timeout).finish();
  66. pinger->set_ponger_addr(ponger->get_address());
  67. ponger->set_pinger_addr(pinger->get_address());
  68. sup1->do_process();
  69. sup2->do_process();
  70. REQUIRE(pinger->ping_sent == 0);
  71. REQUIRE(pinger->pong_received == 0);
  72. REQUIRE(ponger->ping_received == 0);
  73. REQUIRE(ponger->pong_sent == 0);
  74. pinger->do_send_ping();
  75. sup1->do_process();
  76. REQUIRE(pinger->ping_sent == 1);
  77. REQUIRE(pinger->pong_received == 0);
  78. REQUIRE(ponger->ping_received == 0);
  79. REQUIRE(ponger->pong_sent == 0);
  80. sup2->do_process();
  81. REQUIRE(pinger->ping_sent == 1);
  82. REQUIRE(pinger->pong_received == 0);
  83. REQUIRE(ponger->ping_received == 1);
  84. REQUIRE(ponger->pong_sent == 1);
  85. sup1->do_process();
  86. REQUIRE(pinger->ping_sent == 1);
  87. REQUIRE(pinger->pong_received == 1);
  88. REQUIRE(ponger->ping_received == 1);
  89. REQUIRE(ponger->pong_sent == 1);
  90. sup1->do_shutdown();
  91. while (!sup1->get_leader_queue().empty() || !sup2->get_leader_queue().empty()) {
  92. sup1->do_process();
  93. sup2->do_process();
  94. }
  95. REQUIRE(sup2->get_state() == r::state_t::SHUT_DOWN);
  96. REQUIRE(sup2->get_leader_queue().size() == 0);
  97. REQUIRE(sup2->get_points().size() == 0);
  98. REQUIRE(rt::empty(sup2->get_subscription()));
  99. REQUIRE(sup1->get_state() == r::state_t::SHUT_DOWN);
  100. REQUIRE(sup1->get_leader_queue().size() == 0);
  101. REQUIRE(sup1->get_points().size() == 0);
  102. REQUIRE(rt::empty(sup1->get_subscription()));
  103. }
  104. TEST_CASE("pinger & ponger on different supervisors, self controlled", "[supervisor]") {
  105. r::system_context_t system_context;
  106. const char locality1[] = "l1";
  107. const char locality2[] = "l2";
  108. auto sup1 = system_context.create_supervisor<rt::supervisor_test_t>()
  109. .locality(locality1)
  110. .timeout(rt::default_timeout)
  111. .finish();
  112. auto sup2 = sup1->create_actor<rt::supervisor_test_t>().locality(locality2).timeout(rt::default_timeout).finish();
  113. auto pinger = sup1->create_actor<pinger_autostart_t>().timeout(rt::default_timeout).finish();
  114. auto ponger = sup2->create_actor<ponger_t>().timeout(rt::default_timeout).finish();
  115. pinger->set_ponger_addr(ponger->get_address());
  116. ponger->set_pinger_addr(pinger->get_address());
  117. sup1->do_process();
  118. sup2->do_process();
  119. REQUIRE(pinger->ping_sent == 0);
  120. REQUIRE(pinger->pong_received == 0);
  121. REQUIRE(ponger->ping_received == 0);
  122. REQUIRE(ponger->pong_sent == 0);
  123. sup1->do_process();
  124. REQUIRE(pinger->ping_sent == 1);
  125. REQUIRE(pinger->pong_received == 0);
  126. REQUIRE(ponger->ping_received == 0);
  127. REQUIRE(ponger->pong_sent == 0);
  128. sup2->do_process();
  129. REQUIRE(pinger->ping_sent == 1);
  130. REQUIRE(pinger->pong_received == 0);
  131. REQUIRE(ponger->ping_received == 1);
  132. REQUIRE(ponger->pong_sent == 1);
  133. sup1->do_process();
  134. REQUIRE(pinger->ping_sent == 1);
  135. REQUIRE(pinger->pong_received == 1);
  136. REQUIRE(ponger->ping_received == 1);
  137. REQUIRE(ponger->pong_sent == 1);
  138. sup1->do_shutdown();
  139. while (!sup1->get_leader_queue().empty() || !sup2->get_leader_queue().empty()) {
  140. sup1->do_process();
  141. sup2->do_process();
  142. }
  143. CHECK(sup2->get_state() == r::state_t::SHUT_DOWN);
  144. CHECK(sup2->get_leader_queue().size() == 0);
  145. CHECK(sup2->get_points().size() == 0);
  146. REQUIRE(rt::empty(sup2->get_subscription()));
  147. CHECK(sup1->get_state() == r::state_t::SHUT_DOWN);
  148. CHECK(sup1->get_leader_queue().size() == 0);
  149. CHECK(sup1->get_points().size() == 0);
  150. REQUIRE(rt::empty(sup1->get_subscription()));
  151. }