014-lifetime-observer.cpp 4.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132
  1. //
  2. // Copyright (c) 2019-2021 Ivan Baidakou (basiliscos) (the dot dmol at gmail dot com)
  3. //
  4. // Distributed under the MIT Software License
  5. //
  6. #include "rotor.hpp"
  7. #include "actor_test.h"
  8. #include "supervisor_test.h"
  9. #include "access.h"
  10. namespace r = rotor;
  11. namespace rt = r::test;
  12. struct observer_config_t : r::actor_config_t {
  13. r::address_ptr_t observable;
  14. using r::actor_config_t::actor_config_t;
  15. };
  16. template <typename Actor> struct observer_config_builder_t : r::actor_config_builder_t<Actor> {
  17. using builder_t = typename Actor::template config_builder_t<Actor>;
  18. using parent_t = r::actor_config_builder_t<Actor>;
  19. using parent_t::parent_t;
  20. builder_t &&observable(const r::address_ptr_t &addr) {
  21. parent_t::config.observable = addr;
  22. return std::move(*static_cast<builder_t *>(this));
  23. }
  24. bool validate() noexcept override { return parent_t::config.observable && parent_t::validate(); }
  25. };
  26. struct observer_t : public r::actor_base_t {
  27. using config_t = observer_config_t;
  28. template <typename Actor> using config_builder_t = observer_config_builder_t<Actor>;
  29. explicit observer_t(config_t &cfg) : r::actor_base_t(cfg), observable{cfg.observable} {}
  30. void configure(r::plugin::plugin_base_t &plugin) noexcept override {
  31. using namespace r::plugin;
  32. plugin.with_casted<starter_plugin_t>(
  33. [this](auto &p) {
  34. p.subscribe_actor(&observer_t::on_sample_initialize, observable);
  35. p.subscribe_actor(&observer_t::on_sample_start, observable);
  36. p.subscribe_actor(&observer_t::on_sample_shutdown, observable);
  37. },
  38. config_phase_t::PREINIT);
  39. }
  40. void on_sample_initialize(r::message::init_request_t &) noexcept { event += 1; }
  41. void on_sample_start(r::message_t<r::payload::start_actor_t> &) noexcept { event += 2; }
  42. void on_sample_shutdown(r::message::shutdown_request_t &) noexcept { event += 4; }
  43. std::uint32_t event = 0;
  44. r::address_ptr_t observable;
  45. };
  46. TEST_CASE("lifetime observer, same locality", "[actor]") {
  47. r::system_context_t system_context;
  48. auto sup = system_context.create_supervisor<rt::supervisor_test_t>().timeout(rt::default_timeout).finish();
  49. auto sample_actor = sup->create_actor<rt::actor_test_t>().timeout(rt::default_timeout).finish();
  50. auto &address = sample_actor->get_address();
  51. auto observer = sup->create_actor<observer_t>().observable(address).timeout(rt::default_timeout).finish();
  52. sup->do_process();
  53. REQUIRE(observer->event == 3);
  54. sample_actor->do_shutdown();
  55. sup->do_process();
  56. REQUIRE(observer->event == 7);
  57. sup->do_shutdown();
  58. sup->do_process();
  59. REQUIRE(sup->get_state() == r::state_t::SHUT_DOWN);
  60. REQUIRE(sup->get_leader_queue().size() == 0);
  61. REQUIRE(sup->get_points().size() == 0);
  62. CHECK(rt::empty(sup->get_subscription()));
  63. }
  64. TEST_CASE("lifetime observer, different localities", "[actor]") {
  65. r::system_context_t system_context;
  66. const char locality1[] = "l1";
  67. const char locality2[] = "l2";
  68. auto sup1 = system_context.create_supervisor<rt::supervisor_test_t>()
  69. .locality(locality1)
  70. .timeout(rt::default_timeout)
  71. .finish();
  72. auto sup2 = sup1->create_actor<rt::supervisor_test_t>().locality(locality2).timeout(rt::default_timeout).finish();
  73. auto sample_actor = sup2->create_actor<rt::actor_test_t>().timeout(rt::default_timeout).finish();
  74. auto &address = sample_actor->get_address();
  75. auto observer = sup1->create_actor<observer_t>().observable(address).timeout(rt::default_timeout).finish();
  76. sup1->do_process();
  77. CHECK(observer->event == 0);
  78. CHECK(observer->access<rt::to::state>() == r::state_t::INITIALIZING);
  79. sup1->do_process();
  80. sup2->do_process();
  81. CHECK(sample_actor->get_state() == r::state_t::OPERATIONAL);
  82. CHECK(observer->access<rt::to::state>() == r::state_t::INITIALIZING);
  83. sup1->do_process();
  84. sup2->do_process();
  85. CHECK(observer->access<rt::to::state>() == r::state_t::OPERATIONAL);
  86. CHECK(observer->event == 3);
  87. sample_actor->do_shutdown();
  88. sup2->do_process();
  89. sup1->do_process();
  90. sup2->do_process();
  91. sup1->do_process();
  92. REQUIRE(observer->event == 7);
  93. sup1->do_shutdown();
  94. sup1->do_process();
  95. std::cout << "sup1->do_process()\n";
  96. sup2->do_process();
  97. sup1->do_process();
  98. sup2->do_process();
  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. CHECK(rt::empty(sup1->get_subscription()));
  103. }