channel.hpp

浏览该文件的文档.

#ifndef NEFORCE_CORE_ASYNC_CHANNEL_HPP__
#define NEFORCE_CORE_ASYNC_CHANNEL_HPP__

 
#include "NeForce/core/async/condition_variable.hpp"
#include "NeForce/core/async/mutex.hpp"
#include "NeForce/core/container/deque.hpp"
NEFORCE_BEGIN_NAMESPACE__


template <typename T>
class channel {
private:
    size_t capacity_;
    bool closed_{false};
    mutable mutex mutex_;
    condition_variable cv_;
    deque<T> buffer_;
 
public:
    explicit channel(size_t capacity = capacity_max) :
    capacity_(capacity) {}
 
    channel(const channel&) = delete;
    channel& operator=(const channel&) = delete;

    static constexpr size_t capacity_max = numeric_traits<size_t>::max();

    bool try_write(T value) {
        unique_lock<mutex> lock{mutex_};
        if (closed_) {
            return false;
        }
        if (capacity_ > 0 && buffer_.size() >= capacity_) {
            return false;
        }
        buffer_.push_back(move(value));
        cv_.notify_one();
        return true;
    }

    bool write(T value) {
        unique_lock<mutex> lock{mutex_};
        cv_.wait(lock, [this] {
            return closed_ || (capacity_ == 0 && buffer_.empty()) || (capacity_ > 0 && buffer_.size() < capacity_);
        });
        if (closed_) {
            return false;
        }
        buffer_.push_back(move(value));
        cv_.notify_one();
        return true;
    }

    bool try_read(T& out) {
        unique_lock<mutex> lock{mutex_};
        if (buffer_.empty()) {
            return false;
        }
        out = move(buffer_.front());
        buffer_.pop_front();
        cv_.notify_one();
        return true;
    }

    bool read(T& out) {
        unique_lock<mutex> lock{mutex_};
        cv_.wait(lock, [this] { return closed_ || !buffer_.empty(); });
        if (buffer_.empty()) {
            return false;
        }
        out = move(buffer_.front());
        buffer_.pop_front();
        cv_.notify_one();
        return true;
    }

    void close() {
        lock<mutex> lock{mutex_};
        closed_ = true;
        cv_.notify_all();
    }

    NEFORCE_NODISCARD bool is_closed() const noexcept {
        lock<mutex> lock{mutex_};
        return closed_;
    }

    NEFORCE_NODISCARD size_t size() const noexcept {
        lock<mutex> lock{mutex_};
        return buffer_.size();
    }

    NEFORCE_NODISCARD bool empty() const noexcept {
        lock<mutex> lock{mutex_};
        return buffer_.empty();
    }
};
 // Channel
 
NEFORCE_END_NAMESPACE__
#endif // NEFORCE_CORE_ASYNC_CHANNEL_HPP__