UnixSession.hpp

// SPDX-FileCopyrightText: 2025 Vincent Leroy
// SPDX-License-Identifier: MIT
//
// This file is part of dfx.
//
// Licensed under the MIT License. See the LICENSE file in the project root
// for full license information.
 
#pragma once
 
// Standard includes
#include <atomic>
#include <chrono>
#include <memory>
 
// Third-party includes
#include <nlohmann/json.hpp>
 
// Project includes
#include <dfx-fdwatch/PollerFd.hpp>
#include <dfx-utilities/CopyMoveControl.hpp>
 
namespace dfx::FdWatch
{
class Poller;
} // !namespace dfx::FdWatch
 
namespace dfx::Server
{
class UnixRouter;
class UnixServer;

class UnixSession : public std::enable_shared_from_this<UnixSession>
{
public:
    UnixSession(UnixServer & server, UnixRouter & router, std::chrono::milliseconds timeout);

    DISABLE_COPY_AND_MOVE(UnixSession);

    FdWatch::BorrowedFd socket() const noexcept { return _socket.borrow(); }

    uint64_t inMsgCount() const noexcept                { return _inMsgCount.load(std::memory_order::relaxed); }
    uint64_t outMsgCount() const noexcept               { return _outMsgCount.load(std::memory_order::relaxed); }
    uint64_t outErrMsgCount() const noexcept            { return _outErrMsgCount.load(std::memory_order::relaxed); }
    uint64_t outConsecutiveErrMsgCount() const noexcept { return _outConsecutiveErrMsgCount.load(std::memory_order::relaxed); }

    bool subtractTimeAndCheckTimeout(std::chrono::milliseconds time);

    std::chrono::milliseconds initialTimeout() const noexcept { return _initialTimeout; }

    void setInitialTimeout(std::chrono::milliseconds timeout);
 
public:
    void init(FdWatch::OwnedFd socket, FdWatch::Poller & poller);
 
public:
    void reply(nlohmann::json response, std::string_view id);

    void replyError(std::string message, std::string_view id, nlohmann::json baseJson = {})
    { replyError(std::vector<std::string>{ std::move(message) }, id, std::move(baseJson)); }

    void replyError(std::vector<std::string> const & messages, std::string_view id, nlohmann::json baseJson = {});

    void terminate();
 
private:
    void _onSocketReadyRead(FdWatch::EventTriggers events);
    bool _onIncommingMessage();
    bool _readAllPendingMessages(std::vector<std::string> & messages);
    void _sendMessage(std::string_view message);
 
private:
    FdWatch::PollerFd _socket;
    UnixServer & _server;
    UnixRouter & _router;
 
private:
    std::atomic_uint64_t _inMsgCount = 0;
    std::atomic_uint64_t _outMsgCount = 0;
    std::atomic_uint64_t _outErrMsgCount = 0;
    std::atomic_uint64_t _outConsecutiveErrMsgCount = 0;
 
    std::chrono::milliseconds _initialTimeout;
    std::chrono::milliseconds _timeBeforeTimeout{0};
};

using UnixSessionPtr  = std::shared_ptr<UnixSession>;
using UnixSessionWPtr = std::weak_ptr<UnixSession>;
} // !namespace dfx::Server