interprocess_connection.hpp

Go to the documentation of this file.

#pragma once
 
// InterprocessConnection — bidirectional IPC over named pipes, TCP sockets,
// or credential-bearing OS-local sockets.
// Provides length-prefixed message framing, connection lifecycle callbacks,
// and background receive thread. Used for crash-isolated plugin scanning,
// multi-process architectures, and standalone↔plugin communication.
 
#include <pulp/runtime/socket.hpp>
#include <pulp/runtime/alive_token.hpp>
#include <string>
#include <string_view>
#include <vector>
#include <functional>
#include <memory>
#include <optional>
#include <atomic>
#include <thread>
#include <mutex>
#include <cstdint>
#include <chrono>
 
namespace pulp::events {

enum class IpcTransport { NamedPipe, Socket, LocalSocket };

enum class IpcState { Disconnected, Connecting, Connected, Error };

class InterprocessConnection {
public:
    InterprocessConnection();
    virtual ~InterprocessConnection();
 
    // ── Connection lifecycle ────────────────────────────────────────────

    bool connect(
        std::string_view name,
        IpcTransport transport = IpcTransport::NamedPipe,
        std::chrono::milliseconds timeout = {});

    bool create_server(std::string_view name, IpcTransport transport = IpcTransport::NamedPipe,
                       int timeout_ms = 0);

    std::string server_bind_address() const;

    bool attach_inherited_local_socket(std::intptr_t handle);

    void disconnect();

    bool is_connected() const { return state_.load() == IpcState::Connected; }

    IpcState state() const { return state_.load(); }

    std::optional<runtime::LocalPeerCredentials> local_peer_credentials() const;
 
    // ── Messaging ───────────────────────────────────────────────────────

    bool send_message(const void* data, size_t size);

    bool send_message(std::string_view message);

    void set_max_message_bytes(std::size_t bytes);
    std::size_t max_message_bytes() const {
        return max_message_bytes_.load(std::memory_order_relaxed);
    }

    void set_write_timeout(std::chrono::milliseconds timeout);

    void set_frame_read_timeout(std::chrono::milliseconds timeout);

    void set_secure_receive_buffer(bool enabled);
 
    // ── Callbacks (override or set) ─────────────────────────────────────

    virtual void connection_made() {}

    virtual void connection_lost() {}

    virtual void message_received(const void* data, size_t size) {
        (void)data; (void)size;
    }

    virtual void message_received(std::string_view message) {
        (void)message;
    }

    std::function<void()> on_connected;
    std::function<void()> on_disconnected;
    std::function<void(const void*, size_t)> on_message;
    std::function<void(std::string_view)> on_text_message;
 
    void set_on_connected(std::function<void()> callback);
    void set_on_disconnected(std::function<void()> callback);
    void set_on_message(std::function<void(const void*, size_t)> callback);
    void set_on_text_message(std::function<void(std::string_view)> callback);
 
    // No copy
    InterprocessConnection(const InterprocessConnection&) = delete;
    InterprocessConnection& operator=(const InterprocessConnection&) = delete;
 
    friend class InterprocessConnectionServer;  // Needs to inject accepted sockets
 
private:
    struct Impl;
    std::shared_ptr<Impl> impl_;
    std::atomic<IpcState> state_{IpcState::Disconnected};
    std::atomic<bool> running_{false};
    std::atomic<std::uint64_t> connection_generation_{0};
    std::atomic<std::uint64_t> read_generation_{0};
    std::atomic<std::size_t> max_message_bytes_{64u * 1024u * 1024u};
    std::atomic<std::int64_t> write_timeout_ms_{0};
    std::atomic<std::int64_t> frame_read_timeout_ms_{0};
    std::atomic<bool> secure_receive_buffer_{false};
    std::atomic<bool> write_poisoned_{false};
    std::atomic<bool> defer_first_dispatch_until_callback_{false};
    std::shared_ptr<std::atomic<bool>> first_dispatch_gate_;
    mutable std::mutex callback_mutex_;
    runtime::AliveToken alive_;
 
    void disconnect_impl(bool destroying);
    void release_first_dispatch_gate();
    void start_read_thread(bool allow_active_disconnect_owner = false,
                           std::uint64_t expected_connection_generation = 0);
    void read_loop(std::uint64_t generation);
};

class InterprocessConnectionServer {
public:
    InterprocessConnectionServer();
    virtual ~InterprocessConnectionServer();

    bool start(std::string_view name, IpcTransport transport = IpcTransport::Socket);

    void stop();

    void set_max_message_bytes(std::size_t bytes);
    void set_write_timeout(std::chrono::milliseconds timeout);
    void set_frame_read_timeout(std::chrono::milliseconds timeout);

    bool is_running() const { return running_.load(); }

    std::uint16_t bound_port() const;

    std::string bound_address() const;

    virtual void client_connected(std::unique_ptr<InterprocessConnection> connection);

    std::function<void(std::unique_ptr<InterprocessConnection>)> on_client_connected;
 
    // No copy
    InterprocessConnectionServer(const InterprocessConnectionServer&) = delete;
    InterprocessConnectionServer& operator=(const InterprocessConnectionServer&) = delete;
 
private:
    std::atomic<bool> running_{false};
    std::atomic<std::size_t> max_message_bytes_{64u * 1024u * 1024u};
    std::atomic<std::int64_t> write_timeout_ms_{0};
    std::atomic<std::int64_t> frame_read_timeout_ms_{0};
    std::thread accept_thread_;
    std::vector<std::unique_ptr<InterprocessConnection>> clients_;
    struct ServerImpl;
    std::unique_ptr<ServerImpl> server_impl_;
};
 
}  // namespace pulp::events