Channel.h

Go to the documentation of this file.

#pragma once
#include <mutex>
#include <queue>
#include <deque>
#include <condition_variable>
#include <chrono>
#include <tuple>
#include <utility>
#include <scl/macros.h>
#include <scl/tools/meta/type_query.h>
#include <scl/tools/meta/type_mod.h>

//TODO: Refactor names i.e. sender -> emitter, send -> emit

namespace scl{
    namespace async{
        template <class, class, template<class> class, class>
        class Channel;

        namespace details{
            template <class>
            class ChannelSender;

            template <class>
            class ChannelReceiver;

            template <class Chan>
            struct channel_transport_traits{
                using channel_type = Chan;

                using sender_type = typename channel_type::sender_type;

                using receiver_type = typename channel_type::receiver_type;

                using type = std::tuple<sender_type&, receiver_type&>;

                static type factory(sender_type& sender, receiver_type& receiver){
                    return type{sender, receiver};
                }
            };

            template <class Actor>
            struct channel_actor_traits{
                using actor_type = Actor;

                using channel_type = typename actor_type::channel_type;

                using lock_type = typename channel_type::lock_type;

                template <class L>
                using guard_type = typename channel_type::template guard_type<L>;

                using value_type = typename channel_type::value_type;

                using queue_type = typename channel_type::queue_type;

                actor_type* actor;

                std::unique_lock<lock_type> lock_;

                explicit channel_actor_traits(actor_type& act) : actor{&act} {
                    this->lock_ = {this->actor->channel->lock, std::defer_lock};
                }

                void lock(){
                    this->lock_.lock();
                }

                void unlock(){
                    this->lock_.unlock();
                }

                void notify(){
                    this->actor->channel->condition.notify_one();
                }

                void wait(){
                    this->lock();
                    this->actor->channel->condition.wait(this->lock_);
                }

                template <class Pred>
                void waitUntil(Pred&& pred){
                    this->lock();
                    this->actor->channel->condition.wait(
                        this->lock_,
                        std::forward<Pred>(pred)
                    );
                }

                template <class Rep, class Period>
                std::cv_status waitFor(const std::chrono::duration<Rep, Period>& time){
                    this->lock();
                    return this->actor->channel->condition.wait_for(this->lock_, time);
                }

                template <class Rep, class Period, class Pred>
                bool waitFor(const std::chrono::duration<Rep, Period>& time, Pred&& pred){
                    this->lock();
                    return this->actor->channel->condition.wait_for(
                        this->lock_,
                        time,
                        std::forward<Pred>(pred)
                    );
                }
            };
        }

        template <class T, class Lock = std::mutex, template<class> class Guard = std::lock_guard, class Container = std::deque<T>>
        class Channel{
            public:
                using lock_type = Lock;

                template <class L>
                using guard_type = Guard<L>;

                using queue_type = std::queue<T, Container>;

                using value_type = typename queue_type::value_type;

                using size_type = typename queue_type::size_type;

                using sender_type = details::ChannelSender<Channel>;

                using receiver_type = details::ChannelReceiver<Channel>;

                using transport_traits = details::channel_transport_traits<Channel>;

                using transport_type = typename transport_traits::type;

                friend sender_type;
                friend receiver_type;
                friend details::channel_actor_traits<sender_type>;
                friend details::channel_actor_traits<receiver_type>;

                template <size_t I, class U, class L, template<class> class G, class C>
                friend auto std::get(Channel<U, L, G, C>& chan) -> META::tuple_element_t<I, typename Channel<U, L, G, C>::transport_type>;

            protected:
                mutable lock_type lock;

                mutable std::condition_variable_any condition;

                queue_type queue;

                sender_type sender_;

                receiver_type receiver_;

                transport_type interface(){
                    return transport_traits::factory(this->sender(), this->receiver());
                }

            public:
                Channel() : lock{}, condition{}, queue{}, sender_{*this}, receiver_{*this} {
                }

                Channel(const Channel&) = delete;
                Channel(Channel&&) = delete;
                Channel& operator=(const Channel&) = delete;
                Channel& operator=(Channel&&) = delete;

                size_type size() const{
                    guard_type<lock_type> _{lock};
                    return this->queue.size();
                }

                size_type length() const{
                    return this->size();
                }

                bool isEmpty() const{
                    return this->size() <= 0;
                }

                receiver_type& receiver(){
                    return this->receiver_;
//                  return receiver_type{this};
                }

                sender_type& sender(){
                    return this->sender_;
//                  return sender_type{this};
                }

                template <class U = value_type>
                sender_type& operator<<(U&& value){
                    return this->sender() << std::forward<U>(value);
                }

                template <class U = value_type>
                receiver_type& operator>>(U& value){
                    return this->receiver() >> value;
                }
        };

        template <class U, class T, class L, template<class> class G, class C>
        void operator>>(U&& value, Channel<T, L, G, C>& channel) {
            channel << std::forward<U>(value);
        }

        template <class U, class T, class L, template<class> class G, class C>
        void operator<<(U& value, Channel<T, L, G, C>& channel){
            channel >> value;
        }

        namespace details{
            template <class Chan>
            class ChannelSender{
                public:
                    using channel_type = Chan;

                    using sender_traits = channel_actor_traits<ChannelSender>;

                    using lock_type = typename sender_traits::lock_type;

                    template <class L>
                    using guard_type = typename sender_traits::template guard_type<L>;

                    using queue_type = typename sender_traits::queue_type;

                    using value_type = typename sender_traits::value_type;

                    friend sender_traits;
                    friend channel_type;

                protected:
                    channel_type* channel;

                    sender_traits traits;

                    explicit ChannelSender(channel_type& chan) : channel{&chan}, traits{*this} {
                    }

                    template <class... Args>
                    ChannelSender& doPush(Args&&... args){
                        guard_type<sender_traits> _{this->traits};
                        this->channel->queue.emplace(std::forward<Args>(args)...);
                        this->traits.notify();
                        return *this;
                    }

                public:
                    ChannelSender(const ChannelSender&) = delete;
                    ChannelSender& operator=(const ChannelSender&) = delete;
                    ChannelSender(ChannelSender&&) = delete;
                    ChannelSender& operator=(ChannelSender&&) = delete;

                    template <class U = value_type>
                    ChannelSender& push(U&& value){
                        return this->doPush(std::forward<U>(value));
                    }

                    template <class... Args>
                    ChannelSender& pushEmplace(Args&&... args){
                        return this->doPush(std::forward<Args&&>(args)...);
                    }

                    template <class U = value_type>
                    ChannelSender& queue(U&& value){
                        return this->push(std::forward<U>(value));
                    }

                    template <class U = value_type>
                    ChannelSender& enqueue(U&& value){
                        return this->push(std::forward<U>(value));
                    }

                    template <class U = value_type>
                    ChannelSender& send(U&& value){ return this->push(value); }

                    template <class U = value_type>
                    ChannelSender& operator<<(U&& value){ return this->push(value); }
            };

            template <class Chan>
            class ChannelReceiver{
                public:
                    using channel_type = Chan;

                    using receiver_traits = channel_actor_traits<ChannelReceiver>;

                    using lock_type = typename receiver_traits::lock_type;

                    template <class L>
                    using guard_type = typename receiver_traits::template guard_type<L>;

                    using queue_type = typename receiver_traits::queue_type;

                    using value_type = typename receiver_traits::value_type;

                    using optional_type = scl::utils::Optional<value_type>;

                    friend receiver_traits;
                    friend channel_type;

                protected:
                    channel_type* channel;

                    receiver_traits traits;

                    explicit ChannelReceiver(channel_type& chan) : channel{&chan}, traits{*this} {
                    }

                public:
                    ChannelReceiver(const ChannelReceiver&) = delete;
                    ChannelReceiver& operator=(const ChannelReceiver&) = delete;
                    ChannelReceiver(ChannelReceiver&&) = delete;
                    ChannelReceiver& operator=(ChannelReceiver&&) = delete;

                    value_type pop(){
                        this->traits.waitUntil([&]{
                            return !this->channel->queue.empty();
                        });

                        auto value = this->channel->queue.front(); //get front
                        this->channel->queue.pop(); //remove from queue
                        this->traits.unlock();
                        return value;
                    }

                    template <class Rep, class Period>
                    optional_type tryPop(const std::chrono::duration<Rep, Period>& duration){
                        bool isEmpty = this->traits.waitFor(duration, [&]{
                            return !this->channel->queue.empty();
                        });

                        if(isEmpty)
                            return optional_type{};

                        auto value = this->channel->queue.front(); //get front
                        this->channel->queue.pop(); //remove from queue
                        this->traits.unlock();
                        return optional_type{std::move(value)};
                    }

                    value_type dequeue(){ return this->pop(); }

                    value_type receive(){ return this->pop(); }

                    template <class Rep, class Period>
                    optional_type tryDeque(const std::chrono::duration<Rep, Period>& duration){
                        return this->tryPop(duration);
                    }

                    template <class Rep, class Period>
                    optional_type tryReceive(const std::chrono::duration<Rep, Period>& duration){
                        return this->tryPop(duration);
                    }

                    template <class U = value_type>
                    ChannelReceiver& operator>>(U& value){
                        value = std::move(this->receive());
                        return *this;
                    }
            };
        }
    }
}

namespace std{
    template <size_t I, class T, class Lock, template<class> class Guard, class Container>
    auto get(scl::async::Channel<T, Lock, Guard, Container>& channel) -> META::tuple_element_t<I, typename scl::async::Channel<T, Lock, Guard, Container>::transport_type> {
        return std::get<I>(channel.interface());
    }

    template <class T, class Lock, template<class> class Guard, class Container>
    struct tuple_size<scl::async::Channel<T, Lock, Guard, Container>>
            : std::tuple_size<typename scl::async::Channel<T, Lock, Guard, Container>::transport_type>{
    };
}