channel.hpp

Go to the documentation of this file.

// Copyright (C) 2020-2025 Andrei Avram
 
#ifndef MSD_CHANNEL_CHANNEL_HPP_
#define MSD_CHANNEL_CHANNEL_HPP_
 
#include "blocking_iterator.hpp"
#include "nodiscard.hpp"
#include "storage.hpp"
 
#include <condition_variable>
#include <cstdlib>
#include <mutex>
#include <stdexcept>
#include <type_traits>
 
namespace msd {
 
class closed_channel : public std::runtime_error {
   public:
    explicit closed_channel(const char* msg) : std::runtime_error{msg} {}
};
 
template <typename T>
using default_storage = queue_storage<T>;
 
template <typename T>
struct is_supported_type {
    static constexpr bool value = std::is_default_constructible<T>::value && std::is_move_constructible<T>::value &&
                                  std::is_move_assignable<T>::value && std::is_destructible<T>::value;
};
 
template <typename, typename = void>
struct is_static_storage : std::false_type {};
 
template <typename Storage>
struct is_static_storage<Storage, decltype((void)Storage::capacity, void())> : std::true_type {};
 
template <typename T, typename Storage = default_storage<T>>
class channel {
   public:
    static_assert(is_supported_type<T>::value, "Type T does not meet all requirements.");
 
    using value_type = T;
 
    using iterator = blocking_iterator<channel<T, Storage>>;
 
    using size_type = std::size_t;
 
    template <typename S = Storage, typename std::enable_if<is_static_storage<S>::value, int>::type = 0>
    constexpr channel() : capacity_{Storage::capacity}
    {
    }
 
    template <typename S = Storage, typename std::enable_if<!is_static_storage<S>::value, int>::type = 0>
    constexpr channel() : storage_{0}
    {
    }
 
    template <typename S = Storage, typename std::enable_if<!is_static_storage<S>::value, int>::type = 0>
    explicit constexpr channel(const size_type capacity) : storage_{capacity}, capacity_{capacity}
    {
    }
 
    template <typename Type, typename Store>
    friend channel<typename std::decay<Type>::type, Store>& operator<<(
        channel<typename std::decay<Type>::type, Store>& chan, Type&& value);
 
    template <typename Type, typename Store>
    friend channel<Type, Store>& operator>>(channel<Type, Store>& chan, Type& out);
 
    template <typename Type>
    bool write(Type&& value)
    {
        {
            std::unique_lock<std::mutex> lock{mtx_};
            wait_before_write(lock);
 
            if (is_closed_) {
                return false;
            }
 
            storage_.push_back(std::forward<Type>(value));
        }
 
        cnd_.notify_one();
 
        return true;
    }
 
    bool read(T& out)
    {
        {
            std::unique_lock<std::mutex> lock{mtx_};
            wait_before_read(lock);
 
            if (storage_.size() == 0 && is_closed_) {
                return false;
            }
 
            storage_.pop_front(out);
        }
 
        cnd_.notify_one();
 
        return true;
    }
 
    NODISCARD size_type size() const noexcept
    {
        std::unique_lock<std::mutex> lock{mtx_};
        return storage_.size();
    }
 
    NODISCARD bool empty() const noexcept
    {
        std::unique_lock<std::mutex> lock{mtx_};
        return storage_.size() == 0;
    }
 
    void close() noexcept
    {
        {
            std::unique_lock<std::mutex> lock{mtx_};
            is_closed_ = true;
        }
        cnd_.notify_all();
    }
 
    NODISCARD bool closed() const noexcept
    {
        std::unique_lock<std::mutex> lock{mtx_};
        return is_closed_;
    }
 
    NODISCARD bool drained() noexcept
    {
        std::unique_lock<std::mutex> lock{mtx_};
        return storage_.size() == 0 && is_closed_;
    }
 
    iterator begin() noexcept { return blocking_iterator<channel<T, Storage>>{*this}; }
 
    iterator end() noexcept { return blocking_iterator<channel<T, Storage>>{*this, true}; }
 
    channel(const channel&) = delete;
    channel& operator=(const channel&) = delete;
    channel(channel&&) = delete;
    channel& operator=(channel&&) = delete;
    virtual ~channel() = default;
 
   private:
    Storage storage_;
    std::condition_variable cnd_;
    mutable std::mutex mtx_;
    std::size_t capacity_{};
    bool is_closed_{};
 
    void wait_before_read(std::unique_lock<std::mutex>& lock)
    {
        cnd_.wait(lock, [this]() { return storage_.size() > 0 || is_closed_; });
    };
 
    void wait_before_write(std::unique_lock<std::mutex>& lock)
    {
        if (capacity_ > 0) {
            cnd_.wait(lock, [this]() { return storage_.size() < capacity_; });
        }
    }
};
 
template <typename T, typename Storage>
channel<typename std::decay<T>::type, Storage>& operator<<(channel<typename std::decay<T>::type, Storage>& chan,
                                                           T&& value)
{
    if (!chan.write(std::forward<T>(value))) {
        throw closed_channel{"cannot write on closed channel"};
    }
 
    return chan;
}
 
template <typename T, typename Storage>
channel<T, Storage>& operator>>(channel<T, Storage>& chan, T& out)
{
    chan.read(out);
 
    return chan;
}
 
}  // namespace msd
 
#endif  // MSD_CHANNEL_CHANNEL_HPP_