EPS.hpp

Go to the documentation of this file.

#pragma once
#include <cstddef>
#include <cstdint>
#include <functional>
#include <list>
#include <memory>
#include <mutex>
#include <unordered_map>
#include <utility>
#include <vector>
namespace hdk::grid::eps {
  template <typename... ArgTypes> class Conduit {
    private:
      struct State;
 
    public:
      typedef std::function<void(ArgTypes...)> Callback_t;
 
      class Tap {
        public:
          Tap() = default;
          Tap(const Tap&) = delete;
          Tap& operator=(const Tap&) = delete;
          Tap(Tap&& other) noexcept
            : state(std::move(other.state)), id(other.id) {
            other.id = 0;
          }
          Tap& operator=(Tap&& other) noexcept {
            if (this != &other) {
              Detach();
              state = std::move(other.state);
              id = other.id;
              other.id = 0;
            }
            return *this;
          }
 
          bool IsActive() const {
            auto locked = state.lock();
            if (!locked || id == 0) {
              return false;
            }
            std::lock_guard<std::mutex> guard(locked->mutex);
            return locked->index.find(id) != locked->index.end();
          }
 
          bool Detach() {
            auto locked = state.lock();
            if (!locked || id == 0) {
              return false;
            }
            std::lock_guard<std::mutex> guard(locked->mutex);
            auto it = locked->index.find(id);
            if (it == locked->index.end()) {
              id = 0;
              return false;
            }
            locked->entries.erase(it->second);
            locked->index.erase(it);
            id = 0;
            return true;
          }
 
          bool Pause() {
            auto locked = state.lock();
            if (!locked || id == 0) {
              return false;
            }
            std::lock_guard<std::mutex> guard(locked->mutex);
            auto it = locked->index.find(id);
            if (it == locked->index.end()) {
              id = 0;
              return false;
            }
            it->second->paused = true;
            return true;
          }
 
          bool Resume() {
            auto locked = state.lock();
            if (!locked || id == 0) {
              return false;
            }
            std::lock_guard<std::mutex> guard(locked->mutex);
            auto it = locked->index.find(id);
            if (it == locked->index.end()) {
              id = 0;
              return false;
            }
            it->second->paused = false;
            return true;
          }
 
          bool IsPaused() const {
            auto locked = state.lock();
            if (!locked || id == 0) {
              return false;
            }
            std::lock_guard<std::mutex> guard(locked->mutex);
            auto it = locked->index.find(id);
            if (it == locked->index.end()) {
              return false;
            }
            return it->second->paused;
          }
 
        private:
          friend class Conduit;
          Tap(std::weak_ptr<State> subscription_state, std::uint64_t subscription_id)
            : state(std::move(subscription_state)), id(subscription_id) { }
 
          std::weak_ptr<State> state;
          std::uint64_t id = 0;
      };
 
      virtual ~Conduit() = default;
 
      virtual Tap Attach(Callback_t callback) {
        std::lock_guard<std::mutex> guard(state->mutex);
        const std::uint64_t id = state->next_id++;
        auto it = state->entries.insert(state->entries.end(), Entry{ id, std::move(callback), false });
        state->index.emplace(id, it);
        return Tap(state, id);
      }
 
      virtual bool Detach(std::uint64_t id) {
        if (id == 0) {
          return false;
        }
        std::lock_guard<std::mutex> guard(state->mutex);
        auto it = state->index.find(id);
        if (it == state->index.end()) {
          return false;
        }
        state->entries.erase(it->second);
        state->index.erase(it);
        return true;
      }
 
      virtual bool Pause(std::uint64_t id) {
        if (id == 0) {
          return false;
        }
        std::lock_guard<std::mutex> guard(state->mutex);
        auto it = state->index.find(id);
        if (it == state->index.end()) {
          return false;
        }
        it->second->paused = true;
        return true;
      }
 
      virtual bool Resume(std::uint64_t id) {
        if (id == 0) {
          return false;
        }
        std::lock_guard<std::mutex> guard(state->mutex);
        auto it = state->index.find(id);
        if (it == state->index.end()) {
          return false;
        }
        it->second->paused = false;
        return true;
      }
 
      virtual bool IsPaused(std::uint64_t id) const {
        if (id == 0) {
          return false;
        }
        std::lock_guard<std::mutex> guard(state->mutex);
        auto it = state->index.find(id);
        if (it == state->index.end()) {
          return false;
        }
        return it->second->paused;
      }
 
      virtual void Trigger(ArgTypes... args) const {
        std::vector<Callback_t> callbacks;
        {
          std::lock_guard<std::mutex> guard(state->mutex);
          callbacks.reserve(state->entries.size());
          for (const auto& entry : state->entries) {
            if (!entry.paused) {
              callbacks.push_back(entry.callback);
            }
          }
        }
        for (auto& callback : callbacks) {
          callback(args...);
        }
      }
 
      void operator()(ArgTypes... args) const { Trigger(args...); }
      Tap operator%(Callback_t callback) { return Attach(std::move(callback)); }
 
    private:
      struct Entry {
        std::uint64_t id;
        Callback_t callback;
        bool paused;
      };
 
      struct State {
        std::mutex mutex;
        std::list<Entry> entries;
        std::unordered_map<std::uint64_t, typename std::list<Entry>::iterator> index;
        std::uint64_t next_id = 1;
      };
 
      std::shared_ptr<State> state = std::make_shared<State>();
  };
 
  class TapScope {
    public:
      TapScope() = default;
      TapScope(const TapScope&) = delete;
      TapScope& operator=(const TapScope&) = delete;
      TapScope(TapScope&&) = delete;
      TapScope& operator=(TapScope&&) = delete;
      ~TapScope() { DetachAll(); }
 
      template <typename TapT> void Add(TapT tap) {
        if (!tap.IsActive()) {
          return;
        }
        std::lock_guard<std::mutex> guard(mutex);
        entries.emplace_back(std::make_unique<TapEntry<TapT>>(std::move(tap)));
      }
 
      template <typename TapT> TapScope& operator<<(TapT tap) {
        Add(std::move(tap));
        return *this;
      }
 
      void DetachAll() {
        std::vector<std::unique_ptr<TapEntryBase>> to_run;
        {
          std::lock_guard<std::mutex> guard(mutex);
          to_run.swap(entries);
        }
        for (auto& tap : to_run) {
          tap->Detach();
        }
      }
 
      void PauseAll() {
        std::lock_guard<std::mutex> guard(mutex);
        for (auto& tap : entries) {
          tap->Pause();
        }
      }
 
      void ResumeAll() {
        std::lock_guard<std::mutex> guard(mutex);
        for (auto& tap : entries) {
          tap->Resume();
        }
      }
 
      void Clear() {
        std::lock_guard<std::mutex> guard(mutex);
        entries.clear();
      }
 
      std::size_t Size() const {
        std::lock_guard<std::mutex> guard(mutex);
        return entries.size();
      }
      
 
    private:
      struct TapEntryBase {
        virtual ~TapEntryBase() = default;
        virtual void Detach() = 0;
        virtual void Pause() = 0;
        virtual void Resume() = 0;
      };
 
      template <typename TapT> struct TapEntry : TapEntryBase {
        explicit TapEntry(TapT&& tap_value)
          : tap(std::move(tap_value)) { }
 
        void Detach() override { tap.Detach(); }
        void Pause() override { tap.Pause(); }
        void Resume() override { tap.Resume(); }
 
        TapT tap;
      };
 
      mutable std::mutex mutex;
      std::vector<std::unique_ptr<TapEntryBase>> entries;
      
  };
} // namespace hdk::grid::eps