command_service_impl.hpp

Go to the documentation of this file.

#ifndef TORII_COMMAND_SERVICE_IMPL_HPP
#define TORII_COMMAND_SERVICE_IMPL_HPP

#include "torii/command_service.hpp"

#include <rxcpp/rx-lite.hpp>
#include "ametsuchi/storage.hpp"
#include "ametsuchi/tx_presence_cache.hpp"
#include "cache/cache.hpp"
#include "cryptography/hash.hpp"
#include "interfaces/iroha_internal/tx_status_factory.hpp"
#include "logger/logger_fwd.hpp"
#include "torii/processor/transaction_processor.hpp"
#include "torii/status_bus.hpp"

namespace iroha {
  namespace torii {
    class CommandServiceImpl : public CommandService {
     public:
      // TODO: 2019-03-13 @muratovv fix with abstract cache type IR-397
      using CacheType = iroha::cache::Cache<
          shared_model::crypto::Hash,
          std::shared_ptr<shared_model::interface::TransactionResponse>,
          shared_model::crypto::Hash::Hasher>;

      CommandServiceImpl(
          std::shared_ptr<iroha::torii::TransactionProcessor> tx_processor,
          std::shared_ptr<iroha::ametsuchi::Storage> storage,
          std::shared_ptr<iroha::torii::StatusBus> status_bus,
          std::shared_ptr<shared_model::interface::TxStatusFactory>
              status_factory,
          std::shared_ptr<iroha::torii::CommandServiceImpl::CacheType> cache,
          std::shared_ptr<iroha::ametsuchi::TxPresenceCache> tx_presence_cache,
          logger::LoggerPtr log);

      ~CommandServiceImpl() override;

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

      void handleTransactionBatch(
          std::shared_ptr<shared_model::interface::TransactionBatch> batch)
          override;

      std::shared_ptr<shared_model::interface::TransactionResponse> getStatus(
          const shared_model::crypto::Hash &request) override;
      rxcpp::observable<
          std::shared_ptr<shared_model::interface::TransactionResponse>>
      getStatusStream(const shared_model::crypto::Hash &hash) override;

     private:
      inline void handleEvents(rxcpp::composite_subscription &subscription,
                               rxcpp::schedulers::run_loop &run_loop);

      void pushStatus(
          const std::string &who,
          std::shared_ptr<shared_model::interface::TransactionResponse>
              response);

      void processBatch(
          std::shared_ptr<shared_model::interface::TransactionBatch> batch);

      std::shared_ptr<iroha::torii::TransactionProcessor> tx_processor_;
      std::shared_ptr<iroha::ametsuchi::Storage> storage_;
      std::shared_ptr<iroha::torii::StatusBus> status_bus_;
      std::shared_ptr<CacheType> cache_;
      std::shared_ptr<shared_model::interface::TxStatusFactory> status_factory_;
      std::shared_ptr<iroha::ametsuchi::TxPresenceCache> tx_presence_cache_;

      rxcpp::composite_subscription status_subscription_;

      logger::LoggerPtr log_;
    };

  }  // namespace torii
}  // namespace iroha

#endif  // TORII_COMMAND_SERVICE_IMPL_HPP