Jungfraujoch/image_puller/TCPImagePuller.h

// SPDX-FileCopyrightText: 2025 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
// SPDX-License-Identifier: GPL-3.0-only

#pragma once

#include <thread>
#include <atomic>
#include <mutex>
#include <memory>
#include "ImagePuller.h"
#include "../common/Logger.h"
#include "../common/ThreadSafeFIFO.h"

class TCPImagePuller : public ImagePuller {
    int fd = -1;
    std::mutex fd_mutex;

    std::string addr;
    std::string host;
    uint16_t port = 0;
    std::optional<int32_t> receive_buffer_size;
    std::atomic<bool> disconnect{false};

    ThreadSafeFIFO<ImagePullerOutput> cbor_fifo{200};
    ThreadSafeFIFO<ImagePullerOutput> repub_fifo{200};

    std::unique_ptr<ZMQSocket> repub_socket;

    std::thread receiver_thread;
    std::thread cbor_thread;
    std::thread repub_thread;
    std::thread heartbeat_thread;

    // Serializes all writer->pusher sends (ACKs, keepalive pongs, busy heartbeats)
    // so that concurrently-sent frames cannot interleave on the socket.
    std::mutex send_mutex;

    Logger logger{"TCPImagePuller"};

    static constexpr uint32_t default_repub_watermark = 220;
    static constexpr auto RepubTimeout = std::chrono::milliseconds(100);
    static constexpr auto HeartbeatInterval = std::chrono::milliseconds(250);

    bool ReadExact(void *buf, size_t size);
    bool SendAll(const void *buf, size_t len);
    bool EnsureConnected();
    void CloseSocket();
    void ReceiverThread();
    void CBORThread();
    void RepubThread();
    void HeartbeatThread();
public:
    explicit TCPImagePuller(const std::string &tcp_addr,
                            std::optional<int32_t> rcv_buffer_size = {},
                            const std::string &repub_address = "",
                            const std::optional<int32_t> &repub_watermark = {});

    ~TCPImagePuller() override;
    bool SupportsAck() const override { return true; }
    bool SendAck(const PullerAckMessage &ack) override;
    void Disconnect() override;
};