#include "ConcurrentQueue.h"
#include "Helpers.h"

namespace tns {

void ConcurrentQueue::Initialize(CFRunLoopRef runLoop, void (*performWork)(void*), void* info) {
    std::unique_lock<std::mutex> lock(initializationMutex_);
    if (terminated) {
        return;
    }
    this->runLoop_ = runLoop;
    CFRunLoopSourceContext sourceContext = { 0, info, 0, 0, 0, 0, 0, 0, 0, performWork };
    this->runLoopTasksSource_ = CFRunLoopSourceCreate(kCFAllocatorDefault, 0, &sourceContext);
    CFRunLoopAddSource(this->runLoop_, this->runLoopTasksSource_, kCFRunLoopCommonModes);
}

void ConcurrentQueue::Push(std::shared_ptr<worker::Message> message) {
    if (this->runLoopTasksSource_ != nullptr && !CFRunLoopSourceIsValid(this->runLoopTasksSource_)) {
        return;
    }

    {
        std::unique_lock<std::mutex> mlock(this->mutex_);
        this->messagesQueue_.push(message);
    }

    this->SignalAndWakeUp();
}

std::vector<std::shared_ptr<worker::Message>> ConcurrentQueue::PopAll() {
    std::unique_lock<std::mutex> mlock(this->mutex_);
    std::vector<std::shared_ptr<worker::Message>> messages;

    while (!this->messagesQueue_.empty()) {
        std::shared_ptr<worker::Message> message = this->messagesQueue_.front();
        this->messagesQueue_.pop();
        messages.push_back(message);
    }

    return messages;
}

void ConcurrentQueue::SignalAndWakeUp() {
    if (this->runLoopTasksSource_ != nullptr) {
        tns::Assert(CFRunLoopSourceIsValid(this->runLoopTasksSource_));
        CFRunLoopSourceSignal(this->runLoopTasksSource_);
    }

    if (this->runLoop_ != nullptr) {
        CFRunLoopWakeUp(this->runLoop_);
    }
}

void ConcurrentQueue::Terminate() {
    std::unique_lock<std::mutex> lock(initializationMutex_);
    terminated = true;
    CFRunLoopRef runLoop = this->runLoop_;
    CFRunLoopSourceRef source = this->runLoopTasksSource_;
    this->runLoopTasksSource_ = nullptr;
    this->runLoop_ = nullptr;

    if (runLoop) {
        CFRunLoopStop(runLoop);
    }

    if (source) {
        CFRunLoopRemoveSource(runLoop, source, kCFRunLoopCommonModes);
        CFRunLoopSourceInvalidate(source);
        CFRelease(source);
    }
}

}