UploadQueue.cpp

#include "UploadQueue.h"

#include <utility>

namespace vr_image_client
{

UploadQueue::UploadQueue(std::size_t max_size)
    : max_size_(max_size)
{
}

UploadQueue::~UploadQueue()
{
    Stop();
}

void UploadQueue::Start()
{
    std::lock_guard<std::mutex> lock(mutex_);
    stop_requested_ = false;
    if (worker_.joinable())
    {
        return;
    }
    worker_ = std::thread(&UploadQueue::WorkerLoop, this);
}

void UploadQueue::Stop()
{
    {
        std::lock_guard<std::mutex> lock(mutex_);
        stop_requested_ = true;
    }
    cond_.notify_all();
    if (worker_.joinable())
    {
        worker_.join();
    }
    std::lock_guard<std::mutex> lock(mutex_);
    queue_.clear();
}

void UploadQueue::Enqueue(EncodedImage payload, LogBuffer &logs)
{
    bool dropped_oldest = false;
    std::function<void()> drop_callback;
    {
        std::lock_guard<std::mutex> lock(mutex_);
        if (queue_.size() >= max_size_)
        {
            if (!queue_.empty())
            {
                queue_.pop_front();
                dropped_oldest = true;
            }
        }
        queue_.push_back(std::move(payload));
        drop_callback = drop_callback_;
    }
    if (dropped_oldest)
    {
        logs.Push("[VR_for_Mac] Upload queue full; dropped oldest frame to enqueue latest capture.\n");
        if (drop_callback)
        {
            drop_callback();
        }
    }
    cond_.notify_one();
}

void UploadQueue::SetUploader(std::function<void(const EncodedImage &)> uploader)
{
    uploader_ = std::move(uploader);
}

void UploadQueue::SetDropCallback(std::function<void()> drop_callback)
{
    drop_callback_ = std::move(drop_callback);
}

void UploadQueue::WorkerLoop()
{
    while (true)
    {
        EncodedImage payload;
        {
            std::unique_lock<std::mutex> lock(mutex_);
            cond_.wait(lock, [&] {
                return stop_requested_ || !queue_.empty();
            });
            if (stop_requested_ && queue_.empty())
            {
                break;
            }
            payload = std::move(queue_.front());
            queue_.pop_front();
        }
        if (uploader_)
        {
            uploader_(payload);
        }
    }
}

} // namespace vr_image_client