Rediska/rediska/worker/AsyncWorker.cpp

#include "rediska/worker/AsyncWorker.hpp"
#include <google/protobuf/empty.pb.h>
#include "v1/primitives/bool.grpc.pb.h"
#include "v1/primitives/bool.pb.h"
#include "v1/primitives/int.grpc.pb.h"
#include "v1/primitives/int.pb.h"
#include "v1/primitives/string.grpc.pb.h"
#include "v1/primitives/string.pb.h"
#include "v1/collections/list.grpc.pb.h"
#include "v1/collections/list.pb.h"

namespace rediska::worker {

AsyncWorker::AsyncWorker(size_t num_workers, size_t cache_capacity) 
    : num_workers_(num_workers) {
    cache_config_.maxCapacity = cache_capacity;
    cache_config_.ttl = 0;
    cache_config_.resetTTLOnAccess = true;
    InitializeCache();
}

AsyncWorker::~AsyncWorker() {
    Stop();
}

void AsyncWorker::Start() {
    if (running_) return;
    
    running_ = true;
    workers_.reserve(num_workers_);
    
    for (size_t i = 0; i < num_workers_; ++i) {
        workers_.emplace_back(&AsyncWorker::WorkerLoop, this);
    }
}

void AsyncWorker::Stop() {
    if (!running_) return;
    
    running_ = false;
    queue_cv_.notify_all();
    
    for (auto& worker : workers_) {
        if (worker.joinable()) {
            worker.join();
        }
    }
    
    workers_.clear();
}

void AsyncWorker::Enqueue(QueueMessage msg) {
    {
        std::lock_guard<std::mutex> lock(queue_mutex_);
        message_queue_.push(std::move(msg));
    }
    queue_cv_.notify_one();
}

void AsyncWorker::WorkerLoop() {
    while (running_) {
        QueueMessage msg;
        {
            std::unique_lock<std::mutex> lock(queue_mutex_);
            queue_cv_.wait(lock, [this] { return !message_queue_.empty() || !running_; });
            
            if (!running_) break;
            
            msg = std::move(message_queue_.front());
            message_queue_.pop();
        }
        
        ProcessMessage(msg);
    }
}

void AsyncWorker::ProcessMessage(QueueMessage& msg) {
    if (!msg.responder) return;
    
    try {
        switch (msg.operation) {
            case OperationId::SET: {
                if (msg.type == CacheValueId::ARRAY) {
                    v1::collections::list::ListSetResponse response;
                    msg.respond<v1::collections::list::ListSetResponse>(response);
                } else {
                    google::protobuf::Empty response;
                    msg.respond<google::protobuf::Empty>(response);
                }
                break;
            }
            case OperationId::GET: {
                cache_->get(std::string(msg.key));
                
                // заглушка, пока кэш не интегрирован полностью TODO
                if (msg.type == CacheValueId::BOOLEAN) {
                    v1::primitives::boolean::BoolGetResponse response;
                    response.set_value(false);
                    msg.respond<v1::primitives::boolean::BoolGetResponse>(response);
                } else if (msg.type == CacheValueId::INT) {
                    v1::primitives::integer::IntGetResponse response;
                    response.set_value(0);
                    msg.respond<v1::primitives::integer::IntGetResponse>(response);
                } else if (msg.type == CacheValueId::STRING) {
                    v1::primitives::str::StringGetResponse response;
                    response.set_value("cached_value");
                    msg.respond<v1::primitives::str::StringGetResponse>(response);
                } else if (msg.type == CacheValueId::ARRAY) {
                    v1::collections::list::ListGetResponse response;
                    msg.respond<v1::collections::list::ListGetResponse>(response);
                }
                break;
            }
            case OperationId::DELETE: {
                if (msg.type == CacheValueId::BOOLEAN) {
                    v1::primitives::boolean::BoolDeleteResponse response;
                    response.set_removed_value(false);
                    msg.respond<v1::primitives::boolean::BoolDeleteResponse>(response);
                } else if (msg.type == CacheValueId::INT) {
                    v1::primitives::integer::IntDeleteResponse response;
                    response.set_removed_value(0);
                    msg.respond<v1::primitives::integer::IntDeleteResponse>(response);
                } else if (msg.type == CacheValueId::STRING) {
                    v1::primitives::str::StringDeleteResponse response;
                    response.set_removed_value("");
                    msg.respond<v1::primitives::str::StringDeleteResponse>(response);
                } else if (msg.type == CacheValueId::ARRAY) {
                    google::protobuf::Empty response;
                    msg.respond<google::protobuf::Empty>(response);
                }
                break;
            }
            case OperationId::LIST_PUSH_BACK:
            case OperationId::LIST_POP_BACK:
            case OperationId::LIST_INSERT:
            case OperationId::LIST_ERASE: {
                google::protobuf::Empty response;
                msg.respond<google::protobuf::Empty>(response);
                break;
            }
            default:
                break;
        }
    } catch (const std::exception& e) {
        std::cout << "Error processing message: " << e.what() << std::endl;
    }
}

void AsyncWorker::InitializeCache() {
    auto callback = [this](auto result) {
        // TODO
    };
    
    cache_ = std::make_unique<cache::LRU>(cache_config_, callback);
}

} // namespace rediska::worker