distributor.h

#pragma once
 
#include "../coro/subscription.h"
#include "../coro/queue.h"
 
#include <vector>
#include <map>
 
namespace coro {
 
template<typename T, typename Lock, typename QueueImpl>
class distributor_queue;
 
struct nolock {
    void lock() {}
    void unlock() {}
};
 
template<typename From, typename To>
concept is_static_cast_convertible = requires(From from){
    {static_cast<To>(from)};
};
 
 
template<typename T, typename Lock = nolock>
class distributor {
public:
 
    using value_type = T;
    using reference_type = std::add_lvalue_reference_t<T>;
    using subscription = coro::subscription<reference_type>;
    using promise_t = promise<reference_type>;
    using pending_notify = typename promise<reference_type>::notify;
    using ID = const void *;
 
    distributor() = default;
    distributor(Lock &&lk) :
            _mx(std::forward<Lock>(lk)) {
    }
 
 
    template<typename ... Args>
    void publish(Args &&... args) {
        promise_t plist = collect();
        if constexpr(sizeof...(Args) == 1 && (is_static_cast_convertible<Args &&, reference_type> && ...)) {
            plist(args...);
        } else {
            value_type tmp(std::forward<Args>(args)...);
            plist(tmp);
        }
    }
 
 
    void cancel_all() {
        collect().cancel();
    }
 
    void reject_all(std::exception_ptr e) {
        collect().reject(std::move(e));
    }
 
    void reject_all() {
        auto e = std::current_exception();
        if (e) reject_all(std::move(e));
        else cancel_all();
    }
 
 
    subscription subscribe(ID id = { }) {
        return typename subscription::subscribe_fn([this, id](promise_t promise) ->prepared_coro {
            subscribe(std::move(promise), id);
            return {};
        });
    }
 
    void subscribe(promise_t &&prom, ID id = { }) {
        trace::awaiting_ref(*this, prom.get_future());
        std::lock_guard _(_mx);
        _subscribers.push_back(std::pair<promise_t, ID>(std::move(prom), id));
    }
 
 
    pending_notify drop(ID id) {
        pending_notify ntf;
        {
            std::lock_guard _(_mx);
            auto iter = std::find_if(_subscribers.begin(), _subscribers.end(),
                    [&](const auto &iter) {
                        return iter.second == id;
                    });
            if (iter == _subscribers.end())
                return {};
            ntf = iter->first.cancel();
            if (&(iter->first) != &_subscribers.back().first) {
                std::swap(*iter, _subscribers.back());
            }
            _subscribers.pop_back();
        }
        return ntf;
    }
 
 
    template<typename QueueImpl = typename queue<T>::behavior>
    using queue = distributor_queue<T, Lock, QueueImpl>;
 
protected:
    [[no_unique_address]] Lock _mx;
    std::vector<std::pair<promise_t, ID> > _subscribers;
 
    promise_t collect() {
        std::lock_guard _(_mx);
        promise_t p;
        while (!_subscribers.empty()) {
            p += _subscribers.back().first;
            _subscribers.pop_back();
        }
        return p;
    }
};
 
template<typename T, typename Lock, typename QueueImpl>
class distributor_queue: public queue<T, QueueImpl> {
public:
 
    distributor_queue() = default;
    distributor_queue(QueueImpl qimpl) :
            queue<T, QueueImpl>(std::move(qimpl)) {
    }
    distributor_queue(distributor<T, Lock> &dist) {
        subscribe(dist);
    }
    distributor_queue(const distributor_queue&) = delete;
    distributor_queue& operator=(const distributor_queue&) = delete;
 
 
    void subscribe(distributor<T, Lock> &dist) {
        _connection = &dist;
        charge();
    }
 
 
    void unsubscribe() {
        if (_connection) {
            _connection->drop(this);
            _connection = nullptr;
        }
 
    }
 
    ~distributor_queue() {
        unsubscribe();
    }
 
protected:
    typename distributor<T, Lock>::subscription _dist_value;
    distributor<T, Lock> *_connection = nullptr;
    void charge() {
        _dist_value = _connection->subscribe(this);
        _dist_value >> [&]{value_ready();};
    }
 
    void value_ready() {
        try {
            if (_dist_value.has_value()) {
                this->push(T(_dist_value.get()));
                charge();
            } else {
                this->close();
            }
        } catch (...) {
            this->close(std::current_exception());
        }
    }
};
 
 
template<typename T, std::invocable<T> FilterFn, std::invoke_result_t<FilterFn, T> important_message = 0>
class filtered_update_queue {
public:
    using FilterResult = std::invoke_result_t<FilterFn, T>;
 
    template<std::convertible_to<FilterFn> Fn>
    filtered_update_queue(Fn &&fn):_filter(std::forward<Fn>(fn)) {}
 
    filtered_update_queue() = default;
 
    bool empty() const {return _main_queue.empty() && _updates.empty();}
 
 
    T &front() {
        if (_main_queue.empty()) return _updates.begin()->second;
        return _main_queue.front();
    }
 
    const T &front() const {
        if (_main_queue.empty()) return _updates.begin()->second;
        return _main_queue.front();
    }
 
 
    void pop() {
        if (_main_queue.empty()) _updates.erase(_updates.begin());
        else _main_queue.pop_front();
    }
 
    void push(const T &x) {
        emplace(x);
    }
 
    void push(T &&x) {
        emplace(std::move(x));
    }
 
    template<typename ... Args>
    void emplace(Args && ... args) {
        //assume that message is important (we can't determine ID from args)
        //construct it in queue
        _main_queue.emplace_back(std::forward<Args>(args)...);
        //retrieve it constructed
        T &itm = _main_queue.back();
        //retrieve is ID
        FilterResult id = _filter(itm);
 
        //if it is important message, we are good
        if (id == important_message) return;
 
        //find whether there is message with same ID
        auto iter = _updates.find(id);
        //if not
        if (iter == _updates.end()) {
            //create it
            _updates.emplace(id, std::move(itm));
        } else {
            //otherwise replace it
            iter->second = std::move(itm);
        }
        //remove it from the queue )
        _main_queue.pop_back();
    }
 
protected:
    using MainQueue = std::deque<T>;
    using Updates = std::map<FilterResult, T>;
 
    FilterFn _filter;
    MainQueue _main_queue;
    Updates _updates;
 
};
 
}