subscriber_impl.hpp

Go to the documentation of this file.

#ifndef IROHA_SUBSCRIPTION_SUBSCRIBER_IMPL_HPP
#define IROHA_SUBSCRIPTION_SUBSCRIBER_IMPL_HPP

#include <atomic>
#include <functional>
#include <memory>
#include <mutex>

#include "common/common.hpp"
#include "subscription/subscriber.hpp"
#include "subscription/subscription_engine.hpp"

namespace iroha::subscription {

  template <typename EventKey,
            typename Dispatcher,
            typename Receiver,
            typename... Arguments>
  class SubscriberImpl : public Subscriber<EventKey, Dispatcher, Arguments...> {
   public:
    using ReceiverType = Receiver;
    using Hash = size_t;
    using Parent = Subscriber<EventKey, Dispatcher, Arguments...>;

    using SubscriptionEngineType = SubscriptionEngine<
        typename Parent::EventType,
        Dispatcher,
        Subscriber<typename Parent::EventType, Dispatcher, Arguments...>>;
    using SubscriptionEnginePtr = std::shared_ptr<SubscriptionEngineType>;
    using SubscriptionEngineWPtr = std::weak_ptr<SubscriptionEngineType>;

    using CallbackFnType =
        std::function<void(SubscriptionSetId,
                           ReceiverType &,
                           const typename Parent::EventType &,
                           Arguments &&...)>;

   private:
    using SubscriptionsContainer =
        std::unordered_map<typename Parent::EventType,
                           typename SubscriptionEngineType::IteratorType>;
    using SubscriptionsSets =
        std::unordered_map<SubscriptionSetId, SubscriptionsContainer>;

    std::atomic<SubscriptionSetId> next_id_;

    SubscriptionEngineWPtr engine_;

    ReceiverType object_;

    std::mutex subscriptions_cs_;

    SubscriptionsSets subscriptions_sets_;

    CallbackFnType on_notify_callback_;

    template <typename... SubscriberConstructorArgs>
    SubscriberImpl(SubscriptionEnginePtr const &ptr,
                   SubscriberConstructorArgs &&... args)
        : next_id_(0ull),
          engine_(ptr),
          object_(std::forward<SubscriberConstructorArgs>(args)...) {}

   public:
    template <typename... SubscriberConstructorArgs>
    static std::shared_ptr<SubscriberImpl> create(
        SubscriptionEnginePtr const &ptr,
        SubscriberConstructorArgs &&... args) {
      struct Resolver : SubscriberImpl {
        Resolver(SubscriptionEnginePtr const &ptr,
                 SubscriberConstructorArgs &&... args)
            : SubscriberImpl(
                  ptr, std::forward<SubscriberConstructorArgs>(args)...) {}
      };
      return std::make_shared<Resolver>(
          ptr, std::forward<SubscriberConstructorArgs>(args)...);
    }

    ~SubscriberImpl() {}

    void setCallback(CallbackFnType &&f) {
      on_notify_callback_ = std::move(f);
    }

    SubscriptionSetId generateSubscriptionSetId() {
      return ++next_id_;
    }

    void subscribe(SubscriptionSetId id,
                   const typename Parent::EventType &key,
                   typename Dispatcher::Tid tid = Dispatcher::kExecuteInPool) {
      if (auto engine = engine_.lock()) {
        std::lock_guard lock(subscriptions_cs_);
        auto &&[it, inserted] = subscriptions_sets_[id].emplace(
            key, typename SubscriptionEngineType::IteratorType{});

        if (inserted)
          it->second =
              engine->subscribe(tid, id, key, Parent::weak_from_this());
      }
    }

    bool unsubscribe(SubscriptionSetId id,
                     const typename Parent::EventType &key) {
      std::lock_guard<std::mutex> lock(subscriptions_cs_);
      if (auto set_it = subscriptions_sets_.find(id);
          set_it != subscriptions_sets_.end()) {
        auto &subscriptions = set_it->second;
        auto it = subscriptions.find(key);
        if (subscriptions.end() != it) {
          if (auto engine = engine_.lock())
            engine->unsubscribe(key, it->second);
          subscriptions.erase(it);
          return true;
        }
      }
      return false;
    }

    bool unsubscribe(SubscriptionSetId id) {
      std::lock_guard<std::mutex> lock(subscriptions_cs_);
      if (auto set_it = subscriptions_sets_.find(id);
          set_it != subscriptions_sets_.end()) {
        if (auto engine = engine_.lock()) {
          auto &subscriptions = set_it->second;
          for (auto &[key, it] : subscriptions) engine->unsubscribe(key, it);
        }

        subscriptions_sets_.erase(set_it);
        return true;
      }
      return false;
    }

    void unsubscribe() {
      std::lock_guard<std::mutex> lock(subscriptions_cs_);
      if (auto engine = engine_.lock())
        for (auto &[_, subscriptions] : subscriptions_sets_)
          for (auto &[key, it] : subscriptions) engine->unsubscribe(key, it);

      subscriptions_sets_.clear();
    }

    void on_notify(SubscriptionSetId set_id,
                   const typename Parent::EventType &key,
                   Arguments &&... args) override {
      if (nullptr != on_notify_callback_)
        on_notify_callback_(set_id, object_, key, std::move(args)...);
    }

    ReceiverType &get() {
      return object_;
    }
  };

}  // namespace iroha::subscription

#endif  // IROHA_SUBSCRIPTION_SUBSCRIBER_IMPL_HPP