#include "RNSkDispatchQueue.h"

#include <memory>
#include <mutex>
#include <utility>

namespace RNSkia
{

  RNSkDispatchQueue::~RNSkDispatchQueue()
  {
    // Signal to dispatch threads that it's time to wrap up
    std::unique_lock<std::mutex> lock(lock_);
    quit_ = true;
    lock.unlock();
    cv_.notify_all();

    // Wait for threads to finish before we exit
    for (size_t i = 0; i < threads_.size(); i++)
    {
      if (threads_[i].joinable())
      {
        threads_[i].join();
      }
    }
  }

  RNSkDispatchQueue::RNSkDispatchQueue(std::string name, size_t thread_cnt)
      : name_{std::move(name)}, threads_(thread_cnt)
  {
    for (size_t i = 0; i < threads_.size(); i++)
    {
      threads_[i] = std::thread(&RNSkDispatchQueue::dispatch_thread_handler, this);
    }
  }

  void RNSkDispatchQueue::dispatch(const fp_t &op)
  {
    std::unique_lock<std::mutex> lock(lock_);
    q_.push(op);

    // Manual unlocking is done before notifying, to avoid waking up
    // the waiting thread only to block again (see notify_one for details)
    lock.unlock();
    cv_.notify_one();
  }

  void RNSkDispatchQueue::dispatch(fp_t &&op)
  {
    std::unique_lock<std::mutex> lock(lock_);
    q_.push(std::move(op));

    // Manual unlocking is done before notifying, to avoid waking up
    // the waiting thread only to block again (see notify_one for details)
    lock.unlock();
    cv_.notify_one();
  }

  void RNSkDispatchQueue::dispatch_thread_handler(void)
  {
    std::unique_lock<std::mutex> lock(lock_);

    do
    {
      // Wait until we have data or a quit signal
      cv_.wait(lock, [this]
               { return (q_.size() || quit_); });

      // after wait, we own the lock
      if (!quit_ && q_.size())
      {
        auto op = std::move(q_.front());
        q_.pop();

        // unlock now that we're done messing with the queue
        lock.unlock();

        op();

        lock.lock();
      }
    } while (!quit_);
  }
} // namespace RNSkia