Jungfraujoch/reader/JFJochImageBufferReader.h

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

#pragma once

#include <atomic>
#include <chrono>
#include <condition_variable>
#include <mutex>
#include <string>
#include <vector>

#include "JFJochReader.h"
#include "../common/ImageBuffer.h"

// Reads a collection inside the broker while it arrives: the dataset comes from its start message,
// the images from the broker's image buffer, where the receiver puts each one as it forwards it. A
// read of an image not there yet waits for it; after Ended it returns false for an image that never
// came, and so does a read of an image still missing when one MAX_IMAGE_LAG past it is there. The
// buffer must hold the whole collection (slots >= images) - a slot taken again for a later
// image would hand out that image instead. However many threads wait, one at a time looks at the buffer,
// every POLL_PERIOD, so the waiting stays off the lock the receiver puts the images in under.
class JFJochImageBufferReader : public JFJochReader {
public:
    constexpr static int64_t MAX_IMAGE_LAG = 100;
    constexpr static auto POLL_PERIOD = std::chrono::milliseconds(2);

    explicit JFJochImageBufferReader(ImageBuffer &buffer);

    void Open(const StartMessage &start);
    // No image comes after this: the collection is over, or abandoned.
    void Ended();

    [[nodiscard]] uint64_t GetNumberOfImages() const override;
    void Close() override;
    bool ReadRawImage(int64_t image_number, JFJochReaderRawImage &image) override;
    std::vector<SpotToSave> ReadSpots(int64_t image) const override { return {}; }

private:
    bool LoadImage_i(std::shared_ptr<JFJochReaderDataset> &dataset, DataMessage &message,
                     std::vector<uint8_t> &buffer, int64_t image_number, bool update_dataset) override;
    // The image's CBOR message, as the receiver put it in the buffer; false when it never came.
    bool WaitForMessage(int64_t image_number, std::vector<uint8_t> &cbor);

    // Looks at the buffer (GetStatus) and notes the images it holds in present_; under m_, released while
    // the buffer is asked.
    void Poll(std::unique_lock<std::mutex> &ul);

    ImageBuffer &buffer_;
    uint64_t number_of_images_ = 0;
    std::atomic<bool> ended_ = false;

    std::mutex m_;
    std::condition_variable cv_;
    bool polling_ = false;
    std::chrono::steady_clock::time_point next_poll_;
    std::vector<bool> present_;
    int64_t max_present_ = -1;
};