queued_server_write_reactor_impl.hpp

// queued_server_write_reactor_impl.hpp
// Created by Cascade AI / GPT4.1, 2025
// Copyright (C) 2025 Presage Security, Inc.
//
// SPDX-License-Identifier: LGPL-3.0-or-later
#pragma once
 
// === standard library includes (if any) ===
#include <memory>
// === third-party includes (if any) ===
#include <grpc/grpc.h>
#include <grpcpp/grpcpp.h>
#include <grpcpp/impl/codegen/server_callback.h>
// === local includes (if any) ===
#include "queued_server_write_reactor.hpp"
 
namespace presage::smartspectra::grpc_bindings {
 
template <typename TMessage>
QueuedServerWriteReactor<TMessage>::QueuedServerWriteReactor() = default;
 
template <typename TMessage>
QueuedServerWriteReactor<TMessage>::~QueuedServerWriteReactor() = default;
 
template <typename TMessage>
void QueuedServerWriteReactor<TMessage>::EnqueueWrite(const TMessage& message) {
    std::lock_guard<std::mutex> lock(write_mutex_);
    auto message_pointer = std::make_shared<TMessage>();
    message_pointer->CopyFrom(message);
    pending_writes.push(message_pointer);
    ProcessWriteQueue();
}
 
template <typename TMessage>
void QueuedServerWriteReactor<TMessage>::OnWriteDone(bool ok) {
    {
        std::lock_guard<std::mutex> lock(write_mutex_);
        write_in_progress = false;
        if (ok) {
            ProcessWriteQueue();
        }
    }
    if (!ok) {
        this->Finish(grpc::Status(grpc::StatusCode::UNKNOWN, "Unexpected Failure writing message."));
    }
}
 
template <typename TMessage>
void QueuedServerWriteReactor<TMessage>::ProcessWriteQueue() {
    if (!write_in_progress && !pending_writes.empty()) {
        auto message_ptr = pending_writes.front();
        pending_writes.pop();
        write_in_progress = true;
        this->StartWrite(message_ptr.get());
    }
}
 
} // namespace presage::smartspectra::grpc_bindings