gossip_propagation_strategy.hpp

Go to the documentation of this file.

#ifndef IROHA_GOSSIP_PROPAGATION_STRATEGY_HPP
#define IROHA_GOSSIP_PROPAGATION_STRATEGY_HPP

#include <boost/optional.hpp>
#include <chrono>
#include <mutex>

#include <rxcpp/rx-lite.hpp>
#include "ametsuchi/peer_query_factory.hpp"
#include "multi_sig_transactions/gossip_propagation_strategy_params.hpp"
#include "multi_sig_transactions/mst_propagation_strategy.hpp"

#include <rxcpp/operators/rx-observe_on.hpp>

namespace iroha {

  class GossipPropagationStrategy : public PropagationStrategy {
   public:
    using PeerProviderFactory = std::shared_ptr<ametsuchi::PeerQueryFactory>;
    using OptPeer = boost::optional<PropagationData::value_type>;
    GossipPropagationStrategy(
        // TODO 30.01.2019 lebdron: IR-266 Remove PeerQueryFactory
        PeerProviderFactory peer_factory,
        rxcpp::observe_on_one_worker emit_worker,
        const GossipPropagationStrategyParams &params);

    ~GossipPropagationStrategy();

    // ------------------| PropagationStrategy override |------------------

    rxcpp::observable<PropagationData> emitter() override;

    // --------------------------| end override |---------------------------
   private:
    PeerProviderFactory peer_factory;

    PropagationData last_data;

    std::vector<size_t> non_visited;

    rxcpp::observe_on_one_worker emit_worker;

    /*
     * Observable for the emitting propagated data
     */
    rxcpp::observable<PropagationData> emitent;

    /*
     * Mutex for handling observable stopping
     */
    std::mutex m;

    bool initQueue();

    OptPeer visit();
  };
}  // namespace iroha

#endif  // IROHA_GOSSIP_PROPAGATION_STRATEGY_HPP