tec_actor_worker.hpp

Go to the documentation of this file.

// Time-stamp: <Last changed 2026-02-20 15:52:23 by magnolia>
/*----------------------------------------------------------------------
------------------------------------------------------------------------
Copyright (c) 2020-2026 The Emacs Cat (https://github.com/olddeuteronomy/tec).
 
   Licensed under the Apache License, Version 2.0 (the "License");
   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.
------------------------------------------------------------------------
----------------------------------------------------------------------*/
#pragma once
 
#include <cassert>
#include <memory>
#include <mutex>
 
#include "tec/tec_def.hpp" // IWYU pragma: keep
#include "tec/tec_signal.hpp"
#include "tec/tec_status.hpp"
#include "tec/tec_trace.hpp"
#include "tec/tec_worker.hpp"
#include "tec/tec_actor.hpp"
 
 
namespace tec {
 
template <typename TParams, typename TActor>
class ActorWorker : public Worker<TParams> {
  public:
      using Params = TParams;
 
  protected:
      std::unique_ptr<TActor> actor_;
 
  private:
      std::thread actor_thread_;
 
      Signal sig_started_;
 
      Signal sig_stopped_;
 
      Status status_started_;
 
      Status status_stopped_;
 
      std::mutex mtx_request_;
 
public:
    ActorWorker(const Params& params, std::unique_ptr<TActor> actor)
        : Worker<Params>(params)
        , actor_{std::move(actor)}
    {
        static_assert(
            std::is_base_of_v<Actor, TActor>,
            "ActorWorker::TActor must derive from tec::Actor");
        //
        // Register synchronous request handler.
        //
        this->template register_callback<ActorWorker, Payload*>(
            this, &ActorWorker::on_request);
    }
 
    ActorWorker(const ActorWorker&) = delete;
 
    ActorWorker(ActorWorker&&) = delete;
 
    ~ActorWorker() override {
        if (actor_thread_.joinable()) {
            actor_thread_.join();
        }
    }
 
protected:
    Status on_init() override {
        TEC_ENTER("ActorWorker::on_init");
        if (actor_thread_.joinable()) {
            return Status{"Actor is already running", Error::Kind::RuntimeErr};
        }
        //
        // Launch Actor in dedicated thread.
        //
        actor_thread_ = std::thread([this] {
            actor_->start(&sig_started_, &status_started_);
        });
        //
        // Block until startup completes.
        //
        sig_started_.wait();
        TEC_TRACE("Actor thread {} started with status: {}", actor_thread_.get_id(), status_started_);
        return status_started_;
    }
 
    Status on_exit() override {
        TEC_ENTER("ActorWorker::on_exit");
        if (!actor_thread_.joinable()) {
            return {}; // Already stopped
        }
        //
        // Launch Actor's shutdown in a separate thread.
        //
        std::thread shutdown_thread([this] {
            actor_->shutdown(&sig_stopped_);
        });
        TEC_TRACE("Actor thread {} is stopping...", actor_thread_.get_id());
        sig_stopped_.wait();
        TEC_TRACE("Actor thread {} stopped.", actor_thread_.get_id());
 
        shutdown_thread.join();
        actor_thread_.join();
        return {};
    }
 
    virtual void on_request(const Message& msg) {
        typename Worker<TParams>::Lock lk{mtx_request_};
        TEC_ENTER("ActorWorker::on_request");
        TEC_TRACE("Payload received: {}", msg.type().name());
        //
        auto payload = std::any_cast<Payload*>(msg);
        Signal::OnExit on_exit(payload->ready);
        *(payload->status) = actor_->process_request(
            std::move(payload->request), std::move(payload->reply));
    }
 
public:
 
    template <typename WorkerDerived, typename ActorDerived>
    struct Builder {
        std::unique_ptr<Daemon>
        operator()(const typename WorkerDerived::Params& params) const {
            static_assert(
                std::is_base_of_v<ActorWorker, WorkerDerived>,
                "WorkerDerived must derive from tec::ActorWorker");
            static_assert(
                std::is_base_of_v<Actor, ActorDerived>,
                "ActorDerived must derive from tec::Actor");
 
            return std::make_unique<WorkerDerived>(
                params,
                std::move(std::make_unique<ActorDerived>(params)));
        }
    };
 
}; // class ActorWorker
 
} // namespace tec