channel.hpp

// Copyright (C) 2020-2025 Andrei Avram
 
#ifndef MSD_CHANNEL_HPP_
#define MSD_CHANNEL_HPP_
 
#include <atomic>
#include <condition_variable>
#include <cstdlib>
#include <mutex>
#include <queue>
#include <stdexcept>
#include <type_traits>
 
#include "blocking_iterator.hpp"
 
namespace msd {
 
#if (__cplusplus >= 201703L || (defined(_MSVC_LANG) && _MSVC_LANG >= 201703L))
#define NODISCARD [[nodiscard]]
#else
#define NODISCARD
#endif
 
class closed_channel : public std::runtime_error {
   public:
    explicit closed_channel(const char* msg) : std::runtime_error{msg} {}
};
 
template <typename T>
class channel {
   public:
    using value_type = T;
 
    using iterator = blocking_iterator<channel<T>>;
 
    using size_type = std::size_t;
 
    constexpr channel() = default;
 
    explicit constexpr channel(const size_type capacity) : cap_{capacity} {}
 
    template <typename Type>
    friend channel<typename std::decay<Type>::type>& operator<<(channel<typename std::decay<Type>::type>&, Type&&);
 
    template <typename Type>
    friend channel<Type>& operator>>(channel<Type>&, Type&);
 
    NODISCARD size_type constexpr size() const noexcept { return size_; }
 
    NODISCARD bool constexpr empty() const noexcept { return size_ == 0; }
 
    void close() noexcept
    {
        {
            std::unique_lock<std::mutex> lock{mtx_};
            is_closed_.store(true, std::memory_order_seq_cst);
        }
        cnd_.notify_all();
    }
 
    NODISCARD bool closed() const noexcept { return is_closed_.load(std::memory_order_seq_cst); }
 
    iterator begin() noexcept { return blocking_iterator<channel<T>>{*this}; }
 
    iterator end() noexcept { return blocking_iterator<channel<T>>{*this}; }
 
    channel(const channel&) = delete;
    channel& operator=(const channel&) = delete;
    channel(channel&&) = delete;
    channel& operator=(channel&&) = delete;
    virtual ~channel() = default;
 
   private:
    const size_type cap_{0};
    std::queue<T> queue_;
    std::atomic<std::size_t> size_{0};
    std::mutex mtx_;
    std::condition_variable cnd_;
    std::atomic<bool> is_closed_{false};
 
    void waitBeforeRead(std::unique_lock<std::mutex>& lock)
    {
        cnd_.wait(lock, [this]() { return !empty() || closed(); });
    };
 
    void waitBeforeWrite(std::unique_lock<std::mutex>& lock)
    {
        if (cap_ > 0 && size_ == cap_) {
            cnd_.wait(lock, [this]() { return size_ < cap_; });
        }
    }
 
    friend class blocking_iterator<channel>;
};
 
template <typename T>
channel<typename std::decay<T>::type>& operator<<(channel<typename std::decay<T>::type>& chan, T&& value)
{
    {
        std::unique_lock<std::mutex> lock{chan.mtx_};
        chan.waitBeforeWrite(lock);
 
        if (chan.closed()) {
            throw closed_channel{"cannot write on closed channel"};
        }
 
        chan.queue_.push(std::forward<T>(value));
        ++chan.size_;
    }
 
    chan.cnd_.notify_one();
 
    return chan;
}
 
template <typename T>
channel<T>& operator>>(channel<T>& chan, T& out)
{
    {
        std::unique_lock<std::mutex> lock{chan.mtx_};
        chan.waitBeforeRead(lock);
 
        if (chan.closed() && chan.empty()) {
            return chan;
        }
 
        if (!chan.empty()) {
            out = std::move(chan.queue_.front());
            chan.queue_.pop();
            --chan.size_;
        }
    }
 
    chan.cnd_.notify_one();
 
    return chan;
}
 
}  // namespace msd
 
#endif  // MSD_CHANNEL_HPP_