refactor + some multi threading changes

Signed-off-by: Martin Litre <mnlitre@gmail.com>

Author: mlitre

modules/EnergyManagement/EEBUS/EEBUS.cpp

@@ -73,8 +73,10 @@ bool wait_for_channel_ready(const std::shared_ptr<grpc::Channel>& channel, std::
 
 } // namespace
 
-void EEBUS::set_use_case_event_reader(std::unique_ptr<UseCaseEventReader> reader) {
-    this->reader = std::move(reader);
+EEBUS::~EEBUS() {
+    if (this->config.manage_eebus_grpc_api_binary and this->eebus_grpc_api_thread_active.exchange(false)) {
+        this->eebus_grpc_api_thread.join();
+    }
 }
 
 void EEBUS::init() {
@@ -90,6 +92,7 @@ void EEBUS::init() {
     }
 
     if (this->config.manage_eebus_grpc_api_binary) {
+        this->eebus_grpc_api_thread_active.store(true);
         this->eebus_grpc_api_thread =
             std::thread(start_eebus_grpc_api, this->config_validator->get_eebus_grpc_api_binary_path(),
                         this->config.control_service_rpc_port, this->config_validator->get_certificate_path(),
@@ -119,7 +122,10 @@ void EEBUS::init() {
     this->cs_calls->call_start_setup();
     EVLOG_critical << "ML DEBUG PASSED CALL START SETUP";
     this->cs_calls->call_register_remote_ski(this->config.eebus_ems_ski);
-    std::string endpoint = this->cs_calls->call_add_use_case();
+    this->lpc_use_case = control_service::CreateUseCase(
+        control_service::UseCase_ActorType_Enum::UseCase_ActorType_Enum_ControllableSystem,
+        control_service::UseCase_NameType_Enum::UseCase_NameType_Enum_limitationOfPowerConsumption);
+    std::string endpoint = this->cs_calls->call_add_use_case(&this->lpc_use_case);
 
     std::shared_ptr<grpc::Channel> channel2 = grpc::CreateChannel(endpoint, grpc::InsecureChannelCredentials());
     this->failed = !wait_for_channel_ready(channel2, std::chrono::seconds(eebus_one_minute));
@@ -135,21 +141,21 @@ void EEBUS::init() {
     this->cs_lpc_calls->call_set_failsafe_consumption_active_power_limit();
     this->cs_lpc_calls->call_set_failsafe_duration_minimum();
 
-    this->cs_calls->subscribe_use_case_events(this, this->cs_lpc_stub);
+    this->cs_calls->subscribe_use_case_events(this, this->cs_lpc_stub, &this->lpc_use_case);
 }
 
 void EEBUS::ready() {
     invoke_ready(*p_main);
 
-    if (this->failed) {
+    if (this->failed and this->eebus_grpc_api_thread_active.exchange(false)) {
         EVLOG_error << "EEBUS module failed to initialize";
         this->eebus_grpc_api_thread.join();
         return;
     }
 
     this->cs_calls->call_start_service();
 
-    if (this->config.manage_eebus_grpc_api_binary) {
+    if (this->config.manage_eebus_grpc_api_binary and this->eebus_grpc_api_thread_active.exchange(false)) {
         this->eebus_grpc_api_thread.join();
     }
 }

modules/EnergyManagement/EEBUS/EEBUS.hpp

@@ -61,7 +61,7 @@ class EEBUS : public Everest::ModuleBase {
     const Conf& config;
 
     // ev@1fce4c5e-0ab8-41bb-90f7-14277703d2ac:v1
-    void set_use_case_event_reader(std::unique_ptr<UseCaseEventReader> reader);
+    ~EEBUS() override;
     // ev@1fce4c5e-0ab8-41bb-90f7-14277703d2ac:v1
 
 protected:
@@ -77,14 +77,15 @@ class EEBUS : public Everest::ModuleBase {
     // ev@211cfdbe-f69a-4cd6-a4ec-f8aaa3d1b6c8:v1
     std::shared_ptr<control_service::ControlService::Stub> control_service_stub;
     std::shared_ptr<cs_lpc::ControllableSystemLPCControl::Stub> cs_lpc_stub;
-    std::unique_ptr<UseCaseEventReader> reader;
     std::thread eebus_grpc_api_thread;
+    std::atomic<bool> eebus_grpc_api_thread_active;
 
     std::unique_ptr<grpc_calls::ControlServiceCalls> cs_calls;
     std::unique_ptr<grpc_calls::ControllableSystemLPCControlCalls> cs_lpc_calls;
     std::unique_ptr<ConfigValidator> config_validator;
 
     bool failed;
+    control_service::UseCase lpc_use_case;
     // ev@211cfdbe-f69a-4cd6-a4ec-f8aaa3d1b6c8:v1
 };
 

modules/EnergyManagement/EEBUS/grpc_calls/ControlServiceCalls.cpp

@@ -8,11 +8,8 @@ namespace module {
 namespace grpc_calls {
 
 ControlServiceCalls::ControlServiceCalls(const std::shared_ptr<control_service::ControlService::Stub>& stub) :
-    stub(stub) {
+    control_service_stub(stub) {
     this->entity_address = common_types::CreateEntityAddress(std::vector<uint32_t>{1});
-    this->use_case = control_service::CreateUseCase(
-        control_service::UseCase_ActorType_Enum::UseCase_ActorType_Enum_ControllableSystem,
-        control_service::UseCase_NameType_Enum::UseCase_NameType_Enum_limitationOfPowerConsumption);
 }
 
 void ControlServiceCalls::call_set_config(uint32_t port, std::string vendor_code, std::string device_brand,
@@ -25,47 +22,49 @@ void ControlServiceCalls::call_set_config(uint32_t port, std::string vendor_code
         control_service::DeviceType_Enum::DeviceType_Enum_CHARGING_STATION,
         std::vector<control_service::EntityType_Enum>{control_service::EntityType_Enum::EntityType_Enum_EVSE}, 4);
     control_service::EmptyResponse response;
-    control_service::CallSetConfig(this->stub, request, &response);
+    control_service::CallSetConfig(this->control_service_stub, request, &response);
 }
 
 void ControlServiceCalls::call_start_setup() {
     control_service::EmptyResponse response;
-    control_service::CallStartSetup(this->stub, control_service::EmptyRequest(), &response);
+    control_service::CallStartSetup(this->control_service_stub, control_service::EmptyRequest(), &response);
 }
 
 void ControlServiceCalls::call_register_remote_ski(const std::string& remote_ski) {
     control_service::RegisterRemoteSkiRequest request;
     request = control_service::CreateRegisterRemoteSkiRequest(remote_ski);
     control_service::EmptyResponse response;
-    control_service::CallRegisterRemoteSki(this->stub, request, &response);
+    control_service::CallRegisterRemoteSki(this->control_service_stub, request, &response);
 }
 
 void ControlServiceCalls::call_start_service() {
     control_service::EmptyResponse response;
-    control_service::CallStartService(this->stub, control_service::EmptyRequest(), &response);
+    control_service::CallStartService(this->control_service_stub, control_service::EmptyRequest(), &response);
 }
 
-std::string ControlServiceCalls::call_add_use_case() {
+std::string ControlServiceCalls::call_add_use_case(control_service::UseCase* use_case) {
     control_service::AddUseCaseRequest request =
-        control_service::CreateAddUseCaseRequest(&this->entity_address, &this->use_case);
+        control_service::CreateAddUseCaseRequest(&this->entity_address, use_case);
     control_service::AddUseCaseResponse response;
-    control_service::CallAddUseCase(this->stub, request, &response);
+    control_service::CallAddUseCase(this->control_service_stub, request, &response);
 
     std::ignore = request.release_entity_address();
     std::ignore = request.release_use_case();
 
     return response.endpoint();
 }
 
-void ControlServiceCalls::subscribe_use_case_events(
-    module::EEBUS* eebus_module, const std::shared_ptr<cs_lpc::ControllableSystemLPCControl::Stub>& cs_lpc_stub) {
-    control_service::SubscribeUseCaseEventsRequest request =
-        control_service::CreateSubscribeUseCaseEventsRequest(&this->entity_address, &this->use_case);
-    eebus_module->set_use_case_event_reader(
-        std::make_unique<UseCaseEventReader>(this->stub, cs_lpc_stub, request, eebus_module));
-
-    std::ignore = request.release_entity_address();
-    std::ignore = request.release_use_case();
+void ControlServiceCalls::subscribe_use_case_events(module::EEBUS* eebus_module, use_case_stubs stub,
+                                                    control_service::UseCase* use_case) {
+    if (std::holds_alternative<std::shared_ptr<cs_lpc::ControllableSystemLPCControl::Stub>>(stub)) {
+        control_service::SubscribeUseCaseEventsRequest request =
+            control_service::CreateSubscribeUseCaseEventsRequest(&this->entity_address, use_case);
+        this->lpc_uc_reader = std::make_unique<UseCaseEventReader>(
+            this->control_service_stub, std::get<std::shared_ptr<cs_lpc::ControllableSystemLPCControl::Stub>>(stub),
+            request, eebus_module);
+        std::ignore = request.release_entity_address();
+        std::ignore = request.release_use_case();
+    }
 }
 
 } // namespace grpc_calls

modules/EnergyManagement/EEBUS/include/grpc_calls/ControlServiceCalls.hpp

@@ -6,12 +6,16 @@
 #include <control_service/control_service.grpc-ext.pb.h>
 #include <usecases/cs/lpc/service.grpc-ext.pb.h>
 
+#include <UseCaseEventReader.hpp>
+
 namespace module {
 
 class EEBUS;
 
 namespace grpc_calls {
 
+using use_case_stubs = std::variant<std::shared_ptr<cs_lpc::ControllableSystemLPCControl::Stub>>;
+
 class ControlServiceCalls {
 public:
     explicit ControlServiceCalls(const std::shared_ptr<control_service::ControlService::Stub>& stub);
@@ -20,17 +24,17 @@ class ControlServiceCalls {
                          std::string serial_number);
     void call_start_setup();
     void call_register_remote_ski(const std::string& remote_ski);
-    std::string call_add_use_case();
+    std::string call_add_use_case(control_service::UseCase* use_case);
 
-    void subscribe_use_case_events(module::EEBUS* eebus_module,
-                                   const std::shared_ptr<cs_lpc::ControllableSystemLPCControl::Stub>& cs_lpc_stub);
+    void subscribe_use_case_events(module::EEBUS* eebus_module, use_case_stubs stub,
+                                   control_service::UseCase* use_case);
 
     void call_start_service();
 
 private:
-    const std::shared_ptr<control_service::ControlService::Stub> stub;
+    std::shared_ptr<control_service::ControlService::Stub> control_service_stub;
     common_types::EntityAddress entity_address;
-    control_service::UseCase use_case;
+    std::unique_ptr<UseCaseEventReader> lpc_uc_reader;
 };
 
 } // namespace grpc_calls