subscription_manager.hpp

Go to the documentation of this file.

#ifndef IROHA_SUBSCRIPTION_SUBSCRIPTION_MANAGER_HPP
#define IROHA_SUBSCRIPTION_SUBSCRIPTION_MANAGER_HPP

#include <assert.h>
#include <memory>
#include <shared_mutex>
#include <unordered_map>

#include "common/common.hpp"
#include "common/compile-time_murmur2.hpp"
#include "subscription/dispatcher.hpp"
#include "subscription/subscriber.hpp"
#include "subscription/subscription_engine.hpp"

namespace iroha::subscription {

  template <uint32_t kHandlersCount, uint32_t kPoolSize>
  class SubscriptionManager final
      : public std::enable_shared_from_this<
            SubscriptionManager<kHandlersCount, kPoolSize>>,
        utils::NoMove,
        utils::NoCopy {
   public:
    using Dispatcher = subscription::IDispatcher;

   private:
    using EngineHash = uint64_t;
    using DispatcherPtr = std::shared_ptr<Dispatcher>;
    using EnginesList = std::unordered_map<EngineHash, std::shared_ptr<void>>;

   private:
    DispatcherPtr dispatcher_;
    std::shared_mutex engines_cs_;
    EnginesList engines_;
    std::atomic_flag disposed_;

   private:
    template <typename... Args>
    static constexpr EngineHash getSubscriptionHash() {
#ifdef _WIN32
      constexpr EngineHash value = CT_MURMUR2(__FUNCSIG__);
#else   //_WIN32
      constexpr EngineHash value = CT_MURMUR2(__PRETTY_FUNCTION__);
#endif  //_WIN32
      return value;
    }

   public:
    SubscriptionManager(DispatcherPtr dispatcher)
        : dispatcher_(std::move(dispatcher)) {
      disposed_.clear();
    }

    void dispose() {
      if (!disposed_.test_and_set()) {
        {
          std::shared_lock lock(engines_cs_);
          for (auto &descriptor : engines_)
            utils::reinterpret_pointer_cast<IDisposable>(descriptor.second)
                ->dispose();
        }
        dispatcher_->dispose();
      }
    }

    template <typename EventKey, typename... Args>
    auto getEngine() {
      using EngineType =
          SubscriptionEngine<EventKey,
                             Dispatcher,
                             Subscriber<EventKey, Dispatcher, Args...>>;
      constexpr auto engineId = getSubscriptionHash<Args...>();
      {
        std::shared_lock lock(engines_cs_);
        if (auto it = engines_.find(engineId); it != engines_.end()) {
          return utils::reinterpret_pointer_cast<EngineType>(it->second);
        }
      }
      std::unique_lock lock(engines_cs_);
      if (auto it = engines_.find(engineId); it != engines_.end()) {
        return utils::reinterpret_pointer_cast<EngineType>(it->second);
      }

      static_assert(std::is_base_of_v<IDisposable, EngineType>,
                    "Engine type must be derived from IDisposable.");
      assert(uintptr_t(reinterpret_cast<EngineType *>(0x1))
             == uintptr_t(static_cast<IDisposable *>(
                    reinterpret_cast<EngineType *>(0x1))));

      auto obj = std::make_shared<EngineType>(dispatcher_);
      engines_[engineId] = utils::reinterpret_pointer_cast<void>(obj);
      return obj;
    }

    template <typename EventKey, typename... Args>
    void notify(const EventKey &key, Args const &... args) {
      notifyDelayed(std::chrono::microseconds(0ull), key, args...);
    }

    template <typename EventKey, typename... Args>
    void notifyDelayed(std::chrono::microseconds timeout,
                       const EventKey &key,
                       Args const &... args) {
      using EngineType =
          SubscriptionEngine<EventKey,
                             Dispatcher,
                             Subscriber<EventKey, Dispatcher, Args...>>;
      constexpr auto engineId = getSubscriptionHash<Args...>();
      std::shared_ptr<EngineType> engine;
      {
        std::shared_lock lock(engines_cs_);
        if (auto it = engines_.find(engineId); it != engines_.end())
          engine = utils::reinterpret_pointer_cast<EngineType>(it->second);
        else
          return;
      }
      assert(engine);
      engine->notifyDelayed(timeout, key, args...);
    }

    DispatcherPtr dispatcher() const {
      return dispatcher_;
    }
  };
}  // namespace iroha::subscription

#endif  // IROHA_SUBSCRIPTION_SUBSCRIPTION_MANAGER_HPP