<commit_before>extern crate rust_base58;

use std::cell::RefCell;
use std::collections::HashMap;
use std::collections::HashSet;
use std::iter::FromIterator;
use std::rc::Rc;

use rmp_serde;
use serde_json;
use serde_json::Value as SJsonValue;

use commands::Command;
use commands::CommandExecutor;
use commands::ledger::LedgerCommand;
use errors::prelude::*;
use services::ledger::merkletree::merkletree::MerkleTree;
use services::pool::catchup::{build_catchup_req, CatchupProgress, check_cons_proofs, check_nodes_responses_on_status};
use services::pool::events::NetworkerEvent;
use services::pool::events::PoolEvent;
use services::pool::events::RequestEvent;
use services::pool::merkle_tree_factory;
use services::pool::networker::Networker;
use services::pool::state_proof;
use services::pool::types::CatchupRep;
use services::pool::types::HashableValue;

use super::indy_crypto::bls::Generator;
use super::indy_crypto::bls::VerKey;

use self::rust_base58::FromBase58;

struct RequestSM<T: Networker> {
    f: usize,
    cmd_ids: Vec<i32>,
    nodes: HashMap<String, Option<VerKey>>,
    generator: Generator,
    pool_name: String,
    timeout: i64,
    extended_timeout: i64,
    state: RequestState<T>,
}

/// Transitions of request state
/// Start -> Start, Single, Consensus, CatchupSingle, CatchupConsensus, Full, Finish
/// Single -> Single, Finish
/// Consensus -> Consensus, Finish
/// CatchupSingle -> CatchupSingle, Finish
/// CatchupConsensus -> CatchupConsensus, Finish
/// Full -> Full, Finish
/// Finish -> Finish
enum RequestState<T: Networker> {
    Start(StartState<T>),
    Single(SingleState<T>),
    Consensus(ConsensusState<T>),
    CatchupSingle(CatchupSingleState<T>),
    CatchupConsensus(CatchupConsensusState<T>),
    Full(FullState<T>),
    Finish(FinishState),
}

impl<T: Networker> RequestSM<T> {
    pub fn new(networker: Rc<RefCell<T>>,
               f: usize,
               cmd_ids: &Vec<i32>,
               nodes: &HashMap<String, Option<VerKey>>,
               generator: Option<Generator>,
               pool_name: &str, timeout: i64, extended_timeout: i64) -> Self {
        RequestSM {
            f,
            cmd_ids: cmd_ids.clone(),
            nodes: nodes.clone(),
            pool_name: pool_name.to_string(),
            generator: generator.unwrap_or(Generator::from_bytes(&"3LHpUjiyFC2q2hD7MnwwNmVXiuaFbQx2XkAFJWzswCjgN1utjsCeLzHsKk1nJvFEaS4fcrUmVAkdhtPCYbrVyATZcmzwJReTcJqwqBCPTmTQ9uWPwz6rEncKb2pYYYFcdHa8N17HzVyTqKfgPi4X9pMetfT3A5xCHq54R2pDNYWVLDX".from_base58().unwrap()).unwrap()),
            timeout,
            extended_timeout,
            state: RequestState::Start(StartState {
                networker
            }),
        }
    }

    pub fn step(f: usize,
                cmd_ids: Vec<i32>,
                nodes: HashMap<String, Option<VerKey>>,
                generator: Generator,
                pool_name: String,
                timeout: i64,
                extended_timeout: i64,
                state: RequestState<T>) -> Self {
        RequestSM {
            f,
            cmd_ids,
            nodes,
            pool_name,
            generator,
            timeout,
            extended_timeout,
            state,
        }
    }
}

struct StartState<T: Networker> {
    networker: Rc<RefCell<T>>
}

struct ConsensusState<T: Networker> {
    denied_nodes: HashSet<String> /* FIXME should be map, may be merged with replies */,
    replies: HashMap<HashableValue, HashSet<String>>,
    timeout_nodes: HashSet<String>,
    networker: Rc<RefCell<T>>,
}

struct CatchupConsensusState<T: Networker> {
    replies: HashMap<(String, usize, Option<Vec<String>>), HashSet<String>>,
    networker: Rc<RefCell<T>>,
    merkle_tree: MerkleTree,
}

struct CatchupSingleState<T: Networker> {
    target_mt_root: Vec<u8>,
    target_mt_size: usize,
    merkle_tree: MerkleTree,
    networker: Rc<RefCell<T>>,
    req_id: String,
}

struct SingleState<T: Networker> {
    denied_nodes: HashSet<String> /* FIXME should be map, may be merged with replies */,
    replies: HashMap<HashableValue, HashSet<String>>,
    timeout_nodes: HashSet<String>,
    networker: Rc<RefCell<T>>,
}

struct FullState<T: Networker> {
    accum_reply: Option<HashableValue>,
    nodes_to_send: Option<Vec<String>>,
    networker: Rc<RefCell<T>>,
}

struct FinishState {}

impl<T: Networker> From<StartState<T>> for SingleState<T> {
    fn from(state: StartState<T>) -> Self {
        SingleState {
            denied_nodes: HashSet::new(),
            replies: HashMap::new(),
            timeout_nodes: HashSet::new(),
            networker: state.networker.clone(),
        }
    }
}

impl<T: Networker> From<StartState<T>> for ConsensusState<T> {
    fn from(state: StartState<T>) -> Self {
        ConsensusState {
            denied_nodes: HashSet::new(),
            replies: HashMap::new(),
            timeout_nodes: HashSet::new(),
            networker: state.networker.clone(),
        }
    }
}

impl<T: Networker> From<(MerkleTree, StartState<T>)> for CatchupConsensusState<T> {
    fn from((merkle_tree, state): (MerkleTree, StartState<T>)) -> Self {
        CatchupConsensusState {
            replies: HashMap::new(),
            networker: state.networker.clone(),
            merkle_tree,
        }
    }
}

impl<T: Networker> From<(MerkleTree, StartState<T>, Vec<u8>, usize, String)> for CatchupSingleState<T> {
    fn from((merkle_tree, state, target_mt_root, target_mt_size, req_id): (MerkleTree, StartState<T>, Vec<u8>, usize, String)) -> Self {
        CatchupSingleState {
            target_mt_root,
            target_mt_size,
            networker: state.networker.clone(),
            merkle_tree,
            req_id,
        }
    }
}

impl<T: Networker> From<StartState<T>> for FullState<T> {
    fn from(state: StartState<T>) -> Self {
        FullState {
            accum_reply: None,
            nodes_to_send: None,
            networker: state.networker.clone(),
        }
    }
}

impl<T: Networker> From<(Option<Vec<String>>, StartState<T>)> for FullState<T> {
    fn from((nodes_to_send, state): (Option<Vec<String>>, StartState<T>)) -> Self {
        FullState {
            accum_reply: None,
            nodes_to_send,
            networker: state.networker.clone(),
        }
    }
}

impl<T: Networker> RequestState<T> {
    fn finish() -> RequestState<T> {
        RequestState::Finish(FinishState {})
    }
}

impl<T: Networker> RequestSM<T> {
    fn handle_event(self, re: RequestEvent) -> (Self, Option<PoolEvent>) {
        let RequestSM { state, f, cmd_ids, nodes, generator, pool_name, timeout, extended_timeout } = self;
        let (state, event) = match state {
            RequestState::Start(state) => {
                match re {
                    RequestEvent::LedgerStatus(ls, _, Some(merkle)) => {
                        let req_id = ls.merkleRoot.clone();
                        let ne = Some(NetworkerEvent::SendAllRequest(serde_json::to_string(&super::types::Message::LedgerStatus(ls)).expect("FIXME"),
                                                                     req_id, extended_timeout, None));
                        trace!("start catchup, ne: {:?}", ne);
                        state.networker.borrow_mut().process_event(ne);
                        (RequestState::CatchupConsensus((merkle, state).into()), None)
                    }
                    RequestEvent::CatchupReq(merkle, target_mt_size, target_mt_root) => {
                        match build_catchup_req(&merkle, target_mt_size) {
                            Ok(Some((req_id, req_json))) => {
                                state.networker.borrow_mut().process_event(Some(NetworkerEvent::SendOneRequest(req_json, req_id.clone(), timeout)));
                                (RequestState::CatchupSingle((merkle, state, target_mt_root, target_mt_size, req_id).into()), None)
                            }
                            Ok(None) => {
                                warn!("No transactions to catch up!");
                                (RequestState::finish(), Some(PoolEvent::Synced(merkle)))
                            }
                            Err(e) => {
                                _send_replies(&cmd_ids, Err(e));
                                (RequestState::finish(), None)
                            }
                        }
                    }
                    RequestEvent::CustomSingleRequest(msg, req_id) => {
                        state.networker.borrow_mut().process_event(Some(NetworkerEvent::SendOneRequest(msg, req_id, timeout)));
                        (RequestState::Single(state.into()), None)
                    }
                    RequestEvent::CustomFullRequest(msg, req_id, local_timeout, nodes_to_send) => {
                        let timeout = local_timeout.map(|to| to as i64).unwrap_or(extended_timeout);
                        if let Some(nodes_to_send) = nodes_to_send {
                            match serde_json::from_str::<Vec<String>>(&nodes_to_send) {
                                Ok(nodes_to_send) => {
                                    //TODO check empty list on API level?
                                    let is_nodes_to_send_known = !nodes_to_send.is_empty() && nodes_to_send.iter().all(|node| nodes.contains_key(node));
                                    if is_nodes_to_send_known {
                                        state.networker.borrow_mut().process_event(Some(NetworkerEvent::SendAllRequest(msg, req_id, timeout, Some(nodes_to_send.clone()))));
                                        (RequestState::Full((Some(nodes_to_send), state).into()), None)
                                    } else {
                                        _send_replies(&cmd_ids, Err(err_msg(IndyErrorKind::InvalidStructure,
                                                                            format!("There is no known node in list to send {:?}, known nodes are {:?}",
                                                                                    nodes_to_send, nodes.keys()))));
                                        (RequestState::finish(), None)
                                    }
                                }
                                Err(err) => {
                                    _send_replies(&cmd_ids, Err(err.to_indy(IndyErrorKind::InvalidStructure, "Invalid list of nodes to send")));
                                    (RequestState::finish(), None)
                                }
                            }
                        } else {
                            state.networker.borrow_mut().process_event(Some(NetworkerEvent::SendAllRequest(msg, req_id, timeout, None)));
                            (RequestState::Full((None, state).into()), None)
                        }
                    }
                    RequestEvent::CustomConsensusRequest(msg, req_id) => {
                        state.networker.borrow_mut().process_event(Some(NetworkerEvent::SendAllRequest(msg, req_id, timeout, None)));
                        (RequestState::Consensus(state.into()), None)
                    }
                    _ => {
                        (RequestState::Start(state), None)
                    }
                }
            }
            RequestState::Consensus(mut state) => {
                match re {
                    RequestEvent::Reply(_, raw_msg, node_alias, req_id) |
                    RequestEvent::ReqNACK(_, raw_msg, node_alias, req_id) |
                    RequestEvent::Reject(_, raw_msg, node_alias, req_id)
                        => {
                        if let Ok((_, result_without_proof)) = _get_msg_result_without_state_proof(&raw_msg) {
                            let hashable = HashableValue { inner: result_without_proof };

                            let cnt = {
                                let set = state.replies.entry(hashable).or_insert(HashSet::new());
                                set.insert(node_alias.clone());
                                set.len()
                            };

                            if cnt > f {
                                _send_ok_replies(&cmd_ids, &raw_msg);
                                state.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(req_id, None)));
                                (RequestState::finish(), None)
                            } else if state.is_consensus_reachable(f, nodes.len()) {
                                state.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(req_id, Some(node_alias))));
                                (RequestState::Consensus(state), None)
                            } else {
                                //TODO: maybe we should change the error, but it was made to escape changing of ErrorCode returned to client
                                _send_replies(&cmd_ids, Err(err_msg(IndyErrorKind::PoolTimeout, "Consensus is impossible")));
                                state.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(req_id, None)));
                                (RequestState::finish(), None)
                            }
                        } else {
                            (RequestState::Consensus(state), None)
                        }
                    }
                    RequestEvent::ReqACK(_, _, node_alias, req_id) => {
                        state.networker.borrow_mut().process_event(Some(NetworkerEvent::ExtendTimeout(req_id, node_alias, extended_timeout)));
                        (RequestState::Consensus(state), None)
                    }
                    RequestEvent::Timeout(req_id, node_alias) => {
                        state.timeout_nodes.insert(node_alias.clone());
                        if state.is_consensus_reachable(f, nodes.len()) {
                            state.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(req_id, Some(node_alias))));
                            (RequestState::Consensus(state.into()), None)
                        } else {
                            //TODO: maybe we should change the error, but it was made to escape changing of ErrorCode returned to client
                            _send_replies(&cmd_ids, Err(err_msg(IndyErrorKind::PoolTimeout, "Consensus is impossible")));
                            state.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(req_id, None)));
                            (RequestState::finish(), None)
                        }
                    }
                    RequestEvent::Terminate => {
                        _finish_request(&cmd_ids);
                        (RequestState::finish(), None)
                    }
                    _ => (RequestState::Consensus(state.into()), None)
                }
            }
            RequestState::Single(mut state) => {
                match re {
                    RequestEvent::Reply(_, raw_msg, node_alias, req_id) |
                    RequestEvent::ReqNACK(_, raw_msg, node_alias, req_id) |
                    RequestEvent::Reject(_, raw_msg, node_alias, req_id) => {
                        trace!("reply on single request");
                        state.timeout_nodes.remove(&node_alias);
                        if let Ok((result, result_without_proof)) = _get_msg_result_without_state_proof(&raw_msg) {
                            let hashable = HashableValue { inner: result_without_proof };

                            let cnt = {
                                let set = state.replies.entry(hashable).or_insert(HashSet::new());
                                set.insert(node_alias.clone());
                                set.len()
                            };

                            if cnt > f || _check_state_proof(&result, f, &generator, &nodes, &raw_msg) {
                                state.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(req_id, None)));
                                _send_ok_replies(&cmd_ids, &raw_msg);
                                (RequestState::finish(), None)
                            } else {
                                (state.try_to_continue(req_id, node_alias, &cmd_ids, nodes.len(), timeout), None)
                            }
                        } else {
                            state.denied_nodes.insert(node_alias.clone());
                            (state.try_to_continue(req_id, node_alias, &cmd_ids, nodes.len(), timeout), None)
                        }
                    }
                    RequestEvent::ReqACK(_, _, node_alias, req_id) => {
                        state.networker.borrow_mut().process_event(Some(NetworkerEvent::ExtendTimeout(req_id, node_alias, extended_timeout)));
                        (RequestState::Single(state), None)
                    }
                    RequestEvent::Timeout(req_id, node_alias) => {
                        state.timeout_nodes.insert(node_alias.clone());
                        (state.try_to_continue(req_id, node_alias, &cmd_ids, nodes.len(), timeout), None)
                    }
                    RequestEvent::Terminate => {
                        _finish_request(&cmd_ids);
                        (RequestState::finish(), None)
                    }
                    _ => (RequestState::Single(state), None)
                }
            }
            RequestState::CatchupConsensus(state) => {
                match re {
                    RequestEvent::LedgerStatus(ls, Some(node_alias), _) => {
                        RequestSM::_catchup_target_handle_consensus_state(
                            state,
                            ls.merkleRoot.clone(), ls.txnSeqNo, None,
                            node_alias, ls.merkleRoot, f, &nodes, &pool_name)
                    }
                    RequestEvent::ConsistencyProof(cp, node_alias) => {
                        RequestSM::_catchup_target_handle_consensus_state(
                            state,
                            cp.newMerkleRoot, cp.seqNoEnd, Some(cp.hashes),
                            node_alias, cp.oldMerkleRoot, f, &nodes, &pool_name)
                    }
                    RequestEvent::Timeout(req_id, node_alias) => {
                        RequestSM::_catchup_target_handle_consensus_state(
                            state,
                            "timeout".to_string(), 0, None,
                            node_alias, req_id, f, &nodes, &pool_name)
                    }

                    RequestEvent::Terminate => {
                        _finish_request(&cmd_ids);
                        (RequestState::finish(), None)
                    }
                    _ => (RequestState::CatchupConsensus(state), None)
                }
            }
            RequestState::CatchupSingle(mut state) => {
                match re {
                    RequestEvent::CatchupRep(mut cr, node_alias) => {
                        match _process_catchup_reply(&mut cr, &mut state.merkle_tree, &state.target_mt_root, state.target_mt_size, &pool_name) {
                            Ok(merkle) => {
                                state.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(state.req_id.clone(), None)));
                                (RequestState::finish(), Some(PoolEvent::Synced(merkle)))
                            }
                            Err(_) => {
                                state.networker.borrow_mut().process_event(Some(NetworkerEvent::Resend(state.req_id.clone(), timeout)));
                                state.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(state.req_id.clone(), Some(node_alias))));
                                (RequestState::CatchupSingle(state), None)
                            }
                        }
                    }
                    RequestEvent::Timeout(req_id, node_alias) => {
                        state.networker.borrow_mut().process_event(Some(NetworkerEvent::Resend(state.req_id.clone(), timeout)));
                        state.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(req_id, Some(node_alias))));
                        (RequestState::CatchupSingle(state), None)
                    }
                    RequestEvent::Terminate => {
                        _finish_request(&cmd_ids);
                        (RequestState::finish(), None)
                    }
                    _ => (RequestState::CatchupSingle(state), None)
                }
            }
            RequestState::Full(state) => {
                match re {
                    RequestEvent::Reply(_, raw_msg, node_alias, req_id) |
                    RequestEvent::ReqNACK(_, raw_msg, node_alias, req_id) |
                    RequestEvent::Reject(_, raw_msg, node_alias, req_id) =>
                        (RequestSM::_full_request_handle_consensus_state(
                            state, req_id, node_alias, raw_msg, &cmd_ids, &nodes), None),
                    RequestEvent::Timeout(req_id, node_alias) =>
                        (RequestSM::_full_request_handle_consensus_state(
                            state, req_id, node_alias, "timeout".to_string(), &cmd_ids, &nodes), None),

                    RequestEvent::Terminate => {
                        _finish_request(&cmd_ids);
                        (RequestState::finish(), None)
                    }
                    _ => (RequestState::Full(state), None),
                }
            }
            RequestState::Finish(state) => (RequestState::Finish(state), None)
        };
        (RequestSM::step(f, cmd_ids, nodes, generator, pool_name, timeout, extended_timeout, state), event)
    }

    fn is_terminal(&self) -> bool {
        match self.state {
            RequestState::Start(_) |
            RequestState::Consensus(_) |
            RequestState::Single(_) |
            RequestState::CatchupSingle(_) |
            RequestState::CatchupConsensus(_) |
            RequestState::Full(_) => false,
            RequestState::Finish(_) => true
        }
    }

    fn _full_request_handle_consensus_state(mut state: FullState<T>,
                                            req_id: String, node_alias: String, node_result: String,
                                            cmd_ids: &Vec<i32>,
                                            nodes: &HashMap<String, Option<VerKey>>) -> RequestState<T> {
        let is_first_resp = state.accum_reply.is_none();
        if is_first_resp {
            state.accum_reply = Some(HashableValue {
                inner: json!({node_alias.clone(): node_result})
            })
        } else {
            state.accum_reply.as_mut().unwrap()
                .inner.as_object_mut().unwrap()
                .insert(node_alias.clone(), SJsonValue::from(node_result));
        }

        let required_reply_cnt = state.nodes_to_send.as_ref().map(Vec::len).unwrap_or(nodes.len());

        let reply_cnt = state.accum_reply.as_ref().unwrap()
            .inner.as_object().unwrap().len();

        if reply_cnt == required_reply_cnt {
            state.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(req_id, None)));
            let reply = state.accum_reply.as_ref().unwrap().inner.to_string();
            _send_ok_replies(&cmd_ids, &reply);
            RequestState::Finish(FinishState {})
        } else {
            state.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(req_id, Some(node_alias))));
            RequestState::Full(state)
        }
    }

    fn _catchup_target_handle_consensus_state(mut state: CatchupConsensusState<T>,
                                              mt_root: String, sz: usize, cons_proof: Option<Vec<String>>,
                                              node_alias: String, req_id: String,
                                              f: usize, nodes: &HashMap<String, Option<VerKey>>,
                                              pool_name: &str) -> (RequestState<T>, Option<PoolEvent>) {
        let (finished, result) = RequestSM::_process_catchup_target(mt_root, sz, cons_proof,
                                                                    &node_alias, &mut state, f, nodes, pool_name);
        if finished {
            state.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(req_id, None)));
            (RequestState::finish(), result)
        } else {
            state.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(req_id, Some(node_alias))));
            (RequestState::CatchupConsensus(state), result)
        }
    }

    fn _process_catchup_target(merkle_root: String,
                               txn_seq_no: usize,
                               hashes: Option<Vec<String>>,
                               node_alias: &str,
                               state: &mut CatchupConsensusState<T>,
                               f: usize,
                               nodes: &HashMap<String, Option<VerKey>>,
                               pool_name: &str) -> (bool, Option<PoolEvent>) {
        let key = (merkle_root, txn_seq_no, hashes);
        let contains = state.replies.get_mut(&key)
            .map(|set| { set.insert(node_alias.to_string()); })
            .is_some();

        if !contains {
            state.replies.insert(key, HashSet::from_iter(vec![node_alias.to_string()]));
        }

        match check_nodes_responses_on_status(&state.replies,
                                              &state.merkle_tree,
                                              nodes.len(),
                                              f,
                                              &pool_name) {
            Ok(CatchupProgress::InProgress) => (false, None),
            Ok(CatchupProgress::NotNeeded(merkle_tree)) => (true, Some(PoolEvent::Synced(merkle_tree))),
            Ok(CatchupProgress::ShouldBeStarted(target_mt_root, target_mt_size, merkle_tree)) =>
                (true, Some(PoolEvent::CatchupTargetFound(target_mt_root, target_mt_size, merkle_tree))),
            Err(err) => (true, Some(PoolEvent::CatchupTargetNotFound(err))),
        }
    }
}

pub trait RequestHandler<T: Networker> {
    fn new(networker: Rc<RefCell<T>>, f: usize, cmd_ids: &Vec<i32>, nodes: &HashMap<String, Option<VerKey>>, generator: Option<Generator>, pool_name: &str, timeout: i64, extended_timeout: i64) -> Self;
    fn process_event(&mut self, ore: Option<RequestEvent>) -> Option<PoolEvent>;
    fn is_terminal(&self) -> bool;
}

pub struct RequestHandlerImpl<T: Networker> {
    request_wrapper: Option<RequestSM<T>>
}

impl<T: Networker> RequestHandler<T> for RequestHandlerImpl<T> {
    fn new(networker: Rc<RefCell<T>>, f: usize, cmd_ids: &Vec<i32>, nodes: &HashMap<String, Option<VerKey>>, generator: Option<Generator>, pool_name: &str, timeout: i64, extended_timeout: i64) -> Self {
        RequestHandlerImpl {
            request_wrapper: Some(RequestSM::new(networker, f, cmd_ids, nodes, generator, pool_name, timeout, extended_timeout)),
        }
    }

    fn process_event(&mut self, ore: Option<RequestEvent>) -> Option<PoolEvent> {
        match ore {
            Some(re) => {
                if let Some((rw, res)) = self.request_wrapper.take().map(|w| w.handle_event(re)) {
                    self.request_wrapper = Some(rw);
                    res
                } else {
                    self.request_wrapper = None;
                    None
                }
            }
            None => None
        }
    }

    fn is_terminal(&self) -> bool {
        self.request_wrapper.as_ref().map(|w| w.is_terminal()).unwrap_or(true)
    }
}

impl<T: Networker> SingleState<T> {
    fn is_consensus_reachable(&self, total_nodes_cnt: usize) -> bool {
        (self.timeout_nodes.len() + self.denied_nodes.len() + self.replies.values().map(|set| set.len()).sum::<usize>())
            < total_nodes_cnt
    }

    fn try_to_continue(self, req_id: String, node_alias: String, cmd_ids: &Vec<i32>, nodes_cnt: usize, timeout: i64) -> RequestState<T> {
        if self.is_consensus_reachable(nodes_cnt) {
            self.networker.borrow_mut().process_event(Some(NetworkerEvent::Resend(req_id.clone(), timeout)));
            self.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(req_id, Some(node_alias))));
            RequestState::Single(self)
        } else {
            //TODO: maybe we should change the error, but it was made to escape changing of ErrorCode returned to client
            _send_replies(cmd_ids, Err(err_msg(IndyErrorKind::PoolTimeout, "Consensus is impossible")));
            self.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(req_id, None)));
            RequestState::finish()
        }
    }
}

impl<T: Networker> ConsensusState<T> {
    fn is_consensus_reachable(&self, f: usize, total_nodes_cnt: usize) -> bool {
        let rep_no: usize = self.replies.values().map(|set| set.len()).sum();
        let max_no = self.replies.values().map(|set| set.len()).max().unwrap_or(0);
        max_no + total_nodes_cnt - rep_no - self.timeout_nodes.len() - self.denied_nodes.len() > f
    }
}

fn _parse_nack(denied_nodes: &mut HashSet<String>, f: usize, raw_msg: &str, cmd_ids: &Vec<i32>, node_alias: &str) -> bool {
    if denied_nodes.len() == f {
        _send_ok_replies(cmd_ids, raw_msg);
        true
    } else {
        denied_nodes.insert(node_alias.to_string());
        false
    }
}

fn _process_catchup_reply(rep: &mut CatchupRep, merkle: &MerkleTree, target_mt_root: &Vec<u8>, target_mt_size: usize, pool_name: &str) -> IndyResult<MerkleTree> {
    let mut txns_to_drop = vec![];
    let mut merkle = merkle.clone();

    while !rep.txns.is_empty() {
        let key = rep.min_tx()?;
        let txn = rep.txns.remove(&key.to_string()).unwrap();

        let txn = rmp_serde::to_vec_named(&txn)
            .to_indy(IndyErrorKind::InvalidStructure, "Invalid transaction -- can not transform to bytes")?;

        merkle.append(txn.clone())?;
        txns_to_drop.push(txn);
    }

    check_cons_proofs(&merkle, &rep.consProof, target_mt_root, target_mt_size)?;
    merkle_tree_factory::dump_new_txns(pool_name, &txns_to_drop)?;
    Ok(merkle)
}

fn _send_ok_replies(cmd_ids: &Vec<i32>, msg: &str) {
    _send_replies(cmd_ids, Ok(msg.to_string()))
}

fn _finish_request(cmd_ids: &Vec<i32>) {
    _send_replies(cmd_ids, Err(err_msg(IndyErrorKind::PoolTerminated, "Pool is terminated")))
}

fn _send_replies(cmd_ids: &Vec<i32>, msg: IndyResult<String>) {
    cmd_ids.into_iter().for_each(|id| {
        CommandExecutor::instance().send(
            Command::Ledger(
                LedgerCommand::SubmitAck(id.clone(), msg.clone()))
        ).unwrap();
    });
}

fn _get_msg_result_without_state_proof(msg: &str) -> IndyResult<(SJsonValue, SJsonValue)> {
    let msg = serde_json::from_str::<SJsonValue>(msg)
        .to_indy(IndyErrorKind::InvalidStructure, "Response is malformed json")?;

    let msg_result = msg["result"].clone();

    let mut msg_result_without_proof: SJsonValue = msg_result.clone();
    msg_result_without_proof.as_object_mut().map(|obj| obj.remove("state_proof"));

    if msg_result_without_proof["data"].is_object() {
        msg_result_without_proof["data"].as_object_mut().map(|obj| obj.remove("stateProofFrom"));
    }

    Ok((msg_result, msg_result_without_proof))
}

fn _check_state_proof(msg_result: &SJsonValue, f: usize, gen: &Generator, bls_keys: &HashMap<String, Option<VerKey>>, raw_msg: &str) -> bool {
    debug!("TransactionHandler::process_reply: Try to verify proof and signature >>");

    let res = match state_proof::parse_generic_reply_for_proof_checking(&msg_result, raw_msg) {
        Some(parsed_sps) => {
            debug!("TransactionHandler::process_reply: Proof and signature are present");
            state_proof::verify_parsed_sp(parsed_sps, bls_keys, f, gen)
        }
        None => false
    };

    debug!("TransactionHandler::process_reply: Try to verify proof and signature << {}", res);
    res
}

#[cfg(test)]
pub mod tests {
    use services::ledger::merkletree::tree::Tree;
    use services::pool::networker::MockNetworker;
    use services::pool::types::{ConsistencyProof, LedgerStatus, Reply, ReplyResultV1, ReplyTxnV1, ReplyV1, Response, ResponseMetadata, ResponseV1};
    use utils::test;

    use super::*;

    const MESSAGE: &'static str = "message";
    const REQ_ID: &'static str = "1";
    const NODE: &'static str = "n1";
    const NODE_2: &'static str = "n2";
    const NODE_3: &'static str = "n3";
    const NODE_4: &'static str = "n4";
    const SIMPLE_REPLY: &'static str = r#"{"result":{}}"#;
    const POOL: &'static str = "pool1";

    #[derive(Debug)]
    pub struct MockRequestHandler {}

    impl<T: Networker> RequestHandler<T> for MockRequestHandler {
        fn new(_networker: Rc<RefCell<T>>, _f: usize, _cmd_ids: &Vec<i32>, _nodes: &HashMap<String, Option<VerKey>>, _generator: Option<Generator>, _pool_name: &str, _timeout: i64, _extended_timeout: i64) -> Self {
            MockRequestHandler {}
        }

        fn process_event(&mut self, _ore: Option<RequestEvent>) -> Option<PoolEvent> {
            None
        }

        fn is_terminal(&self) -> bool {
            true
        }
    }

    impl Default for LedgerStatus {
        fn default() -> Self {
            LedgerStatus {
                txnSeqNo: 0,
                merkleRoot: String::new(),
                ledgerId: 0,
                ppSeqNo: None,
                viewNo: None,
                protocolVersion: None,
            }
        }
    }

    impl Default for MerkleTree {
        fn default() -> Self {
            MerkleTree {
                root: Tree::Empty { hash: Vec::new() },
                height: 0,
                count: 0,
                nodes_count: 0,
            }
        }
    }

    impl Default for Reply {
        fn default() -> Self {
            Reply::ReplyV1(ReplyV1 { result: ReplyResultV1 { txn: ReplyTxnV1 { metadata: ResponseMetadata { req_id: 1 } } } })
        }
    }

    impl Default for Response {
        fn default() -> Self {
            Response::ResponseV1(ResponseV1 { metadata: ResponseMetadata { req_id: 1 } })
        }
    }

    impl Default for ConsistencyProof {
        fn default() -> Self {
            ConsistencyProof {
                seqNoEnd: 0,
                seqNoStart: 0,
                ledgerId: 0,
                hashes: Vec::new(),
                oldMerkleRoot: String::new(),
                newMerkleRoot: String::new(),
            }
        }
    }

    impl Default for CatchupRep {
        fn default() -> Self {
            CatchupRep {
                ledgerId: 0,
                consProof: Vec::new(),
                txns: HashMap::new(),
            }
        }
    }

    fn _request_handler(f: usize, nodes_cnt: usize) -> RequestHandlerImpl<MockNetworker> {
        let networker = Rc::new(RefCell::new(MockNetworker::new(0, 0, vec![])));

        let mut default_nodes: HashMap<String, Option<VerKey>> = HashMap::new();
        default_nodes.insert(NODE.to_string(), None);

        let node_names = vec![NODE, NODE_2, "n3", "n4"];
        let mut nodes: HashMap<String, Option<VerKey>> = HashMap::new();

        for i in 0..nodes_cnt {
            nodes.insert(node_names[i].to_string(), None);
        }

        RequestHandlerImpl::new(networker,
                                f,
                                &vec![],
                                &nodes,
                                None,
                                POOL,
                                0,
                                0)
    }

    // required because of dumping txns to cache
    fn _create_pool(content: Option<String>) {
        use utils::environment;
        use std::fs;
        use std::fs::File;
        use std::io::Write;

        let mut path = environment::pool_path(POOL);

        path.push(POOL);
        path.set_extension("txn");

        fs::create_dir_all(path.parent().unwrap()).unwrap();

        let mut file = File::create(path).unwrap();
        file.write_all(content.unwrap_or("{}".to_string()).as_bytes()).unwrap();
    }

    #[test]
    fn request_handler_new_works() {
        let request_handler = _request_handler(0, 1);
        assert_match!(RequestState::Start(_), request_handler.request_wrapper.unwrap().state);
    }

    #[test]
    fn request_handler_process_event_works() {
        let mut request_handler = _request_handler(0, 1);
        request_handler.process_event(None);
    }

    mod start {
        use super::*;

        #[test]
        fn request_handler_process_ledger_status_event_from_start_works() {
            let mut request_handler = _request_handler(0, 1);
            request_handler.process_event(Some(RequestEvent::LedgerStatus(LedgerStatus::default(), Some(NODE.to_string()), Some(MerkleTree::default()))));
            assert_match!(RequestState::CatchupConsensus(_), request_handler.request_wrapper.unwrap().state);
        }

        #[test]
        fn request_handler_process_catchup_req_event_from_start_works() {
            let mut request_handler = _request_handler(0, 1);
            request_handler.process_event(Some(RequestEvent::CatchupReq(MerkleTree::default(), 1, vec![])));
            assert_match!(RequestState::CatchupSingle(_), request_handler.request_wrapper.unwrap().state);
        }

        #[test]
        fn request_handler_process_catchup_req_event_from_start_works_for_no_transactions_to_catchup() {
            let mut request_handler = _request_handler(0, 1);
            request_handler.process_event(Some(RequestEvent::CatchupReq(MerkleTree::default(), 0, vec![])));
            assert_match!(RequestState::Finish(_), request_handler.request_wrapper.unwrap().state);
        }

        #[test]
        fn request_handler_process_custom_single_req_event_from_start_works() {
            let mut request_handler = _request_handler(0, 1);
            request_handler.process_event(Some(RequestEvent::CustomSingleRequest(MESSAGE.to_string(), REQ_ID.to_string())));
            assert_match!(RequestState::Single(_), request_handler.request_wrapper.unwrap().state);
        }

        #[test]
        fn request_handler_process_consensus_full_req_event_from_start_works() {
            let mut request_handler = _request_handler(0, 1);
            request_handler.process_event(Some(RequestEvent::CustomFullRequest(MESSAGE.to_string(), REQ_ID.to_string(), None, None)));
            assert_match!(RequestState::Full(_), request_handler.request_wrapper.unwrap().state);
        }

        #[test]
        fn request_handler_process_consensus_full_req_event_from_start_works_for_list_nodes() {
            let mut request_handler = _request_handler(0, 1);
            request_handler.process_event(Some(RequestEvent::CustomFullRequest(MESSAGE.to_string(), REQ_ID.to_string(), None, Some(format!(r#"["{}"]"#, NODE)))));
            assert_match!(RequestState::Full(_), request_handler.request_wrapper.unwrap().state);
        }

        #[test]
        fn request_handler_process_consensus_full_req_event_from_start_works_for_empty_list_nodes() {
            let mut request_handler = _request_handler(0, 1);
            request_handler.process_event(Some(RequestEvent::CustomFullRequest(MESSAGE.to_string(), REQ_ID.to_string(), None, Some("[ ]".to_string()))));
            assert_match!(RequestState::Finish(_), request_handler.request_wrapper.unwrap().state);
        }

        #[test]
        fn request_handler_process_consensus_full_req_event_from_start_works_for_list_nodes_contains_unknown_node() {
            let mut request_handler = _request_handler(0, 1);
            request_handler.process_event(Some(RequestEvent::CustomFullRequest(MESSAGE.to_string(), REQ_ID.to_string(), None, Some("[Unknown Node]".to_string()))));
            assert_match!(RequestState::Finish(_), request_handler.request_wrapper.unwrap().state);
        }

        #[test]
        fn request_handler_process_consensus_full_req_event_from_start_works_for_invalid_list_nodes_format() {
            let mut request_handler = _request_handler(0, 1);
            request_handler.process_event(Some(RequestEvent::CustomFullRequest(MESSAGE.to_string(), REQ_ID.to_string(), None, Some(format!(r#""{}""#, NODE)))));
            assert_match!(RequestState::Finish(_), request_handler.request_wrapper.unwrap().state);
        }

        #[test]
        fn request_handler_process_custom_consensus_req_event_from_start_works() {
            let mut request_handler = _request_handler(0, 1);
            request_handler.process_event(Some(RequestEvent::CustomConsensusRequest(MESSAGE.to_string(), REQ_ID.to_string())));
            assert_match!(RequestState::Consensus(_), request_handler.request_wrapper.unwrap().state);
        }

        #[test]
        fn request_handler_process_other_event_from_start_works() {
            let mut request_handler = _request_handler(0, 1);
            request_handler.process_event(Some(RequestEvent::Timeout(REQ_ID.to_string(), NODE.to_string())));
            assert_match!(RequestState::Start(_), request_handler.request_wrapper.unwrap().state);
        }
    }

    mod consensus_state {
        use super::*;

        #[test]
        fn request_handler_process_reply_event_from_consensus_state_works_for_consensus_reached() {
            let mut request_handler = _request_handler(0, 1);
            request_handler.process_event(Some(RequestEvent::CustomConsensusRequest(MESSAGE.to_string(), REQ_ID.to_string())));
            request_handler.process_event(Some(RequestEvent::Reply(Reply::default(), SIMPLE_REPLY.to_string(), NODE.to_string(), REQ_ID.to_string())));
            assert_match!(RequestState::Finish(_), request_handler.request_wrapper.unwrap().state);
        }

        #[test]
        fn request_handler_process_reply_event_from_consensus_state_works_for_consensus_reached_with_mixed_msgs() {

            // the test will use 4 nodes, each node replying with a response to the "custom consensus request" message
            // some nodes accept, some reject and some nack.  the end result is consensus should not be reached
            let mut request_handler = _request_handler(1, 4);

            request_handler.process_event(Some(RequestEvent::CustomConsensusRequest(MESSAGE.to_string(), REQ_ID.to_string())));
            request_handler.process_event(Some(RequestEvent::Reply(Reply::default(), SIMPLE_REPLY.to_string(), NODE.to_string(), REQ_ID.to_string())));
            request_handler.process_event(Some(RequestEvent::Reject(Response::default(), "{}".to_string(), NODE_2.to_string(), REQ_ID.to_string())));
            request_handler.process_event(Some(RequestEvent::ReqNACK(Response::default(), "{}".to_string(), NODE_3.to_string(), REQ_ID.to_string())));

            // test state is already Finished because already have 2 nodes not in consensus
            {
                let request_handler_ref = request_handler.request_wrapper.as_ref().unwrap();
                assert_match!(RequestState::Finish(_), request_handler_ref.state);
            }
            
            request_handler.process_event(Some(RequestEvent::Reject(Response::default(), "{}".to_string(), NODE_4.to_string(), REQ_ID.to_string())));

            assert_match!(RequestState::Finish(_), request_handler.request_wrapper.unwrap().state);

        }
<commit_msg>changes per conversation with nikita.  see https://jira.hyperledger.org/browse/IS-1137<commit_after>extern crate rust_base58;

use std::cell::RefCell;
use std::collections::HashMap;
use std::collections::HashSet;
use std::iter::FromIterator;
use std::rc::Rc;

use rmp_serde;
use serde_json;
use serde_json::Value as SJsonValue;

use commands::Command;
use commands::CommandExecutor;
use commands::ledger::LedgerCommand;
use errors::prelude::*;
use services::ledger::merkletree::merkletree::MerkleTree;
use services::pool::catchup::{build_catchup_req, CatchupProgress, check_cons_proofs, check_nodes_responses_on_status};
use services::pool::events::NetworkerEvent;
use services::pool::events::PoolEvent;
use services::pool::events::RequestEvent;
use services::pool::merkle_tree_factory;
use services::pool::networker::Networker;
use services::pool::state_proof;
use services::pool::types::CatchupRep;
use services::pool::types::HashableValue;

use super::indy_crypto::bls::Generator;
use super::indy_crypto::bls::VerKey;

use self::rust_base58::FromBase58;

struct RequestSM<T: Networker> {
    f: usize,
    cmd_ids: Vec<i32>,
    nodes: HashMap<String, Option<VerKey>>,
    generator: Generator,
    pool_name: String,
    timeout: i64,
    extended_timeout: i64,
    state: RequestState<T>,
}

/// Transitions of request state
/// Start -> Start, Single, Consensus, CatchupSingle, CatchupConsensus, Full, Finish
/// Single -> Single, Finish
/// Consensus -> Consensus, Finish
/// CatchupSingle -> CatchupSingle, Finish
/// CatchupConsensus -> CatchupConsensus, Finish
/// Full -> Full, Finish
/// Finish -> Finish
enum RequestState<T: Networker> {
    Start(StartState<T>),
    Single(SingleState<T>),
    Consensus(ConsensusState<T>),
    CatchupSingle(CatchupSingleState<T>),
    CatchupConsensus(CatchupConsensusState<T>),
    Full(FullState<T>),
    Finish(FinishState),
}

impl<T: Networker> RequestSM<T> {
    pub fn new(networker: Rc<RefCell<T>>,
               f: usize,
               cmd_ids: &Vec<i32>,
               nodes: &HashMap<String, Option<VerKey>>,
               generator: Option<Generator>,
               pool_name: &str, timeout: i64, extended_timeout: i64) -> Self {
        RequestSM {
            f,
            cmd_ids: cmd_ids.clone(),
            nodes: nodes.clone(),
            pool_name: pool_name.to_string(),
            generator: generator.unwrap_or(Generator::from_bytes(&"3LHpUjiyFC2q2hD7MnwwNmVXiuaFbQx2XkAFJWzswCjgN1utjsCeLzHsKk1nJvFEaS4fcrUmVAkdhtPCYbrVyATZcmzwJReTcJqwqBCPTmTQ9uWPwz6rEncKb2pYYYFcdHa8N17HzVyTqKfgPi4X9pMetfT3A5xCHq54R2pDNYWVLDX".from_base58().unwrap()).unwrap()),
            timeout,
            extended_timeout,
            state: RequestState::Start(StartState {
                networker
            }),
        }
    }

    pub fn step(f: usize,
                cmd_ids: Vec<i32>,
                nodes: HashMap<String, Option<VerKey>>,
                generator: Generator,
                pool_name: String,
                timeout: i64,
                extended_timeout: i64,
                state: RequestState<T>) -> Self {
        RequestSM {
            f,
            cmd_ids,
            nodes,
            pool_name,
            generator,
            timeout,
            extended_timeout,
            state,
        }
    }
}

struct StartState<T: Networker> {
    networker: Rc<RefCell<T>>
}

struct ConsensusState<T: Networker> {
    denied_nodes: HashSet<String> /* FIXME should be map, may be merged with replies */,
    replies: HashMap<HashableValue, HashSet<String>>,
    timeout_nodes: HashSet<String>,
    networker: Rc<RefCell<T>>,
}

struct CatchupConsensusState<T: Networker> {
    replies: HashMap<(String, usize, Option<Vec<String>>), HashSet<String>>,
    networker: Rc<RefCell<T>>,
    merkle_tree: MerkleTree,
}

struct CatchupSingleState<T: Networker> {
    target_mt_root: Vec<u8>,
    target_mt_size: usize,
    merkle_tree: MerkleTree,
    networker: Rc<RefCell<T>>,
    req_id: String,
}

struct SingleState<T: Networker> {
    denied_nodes: HashSet<String> /* FIXME should be map, may be merged with replies */,
    replies: HashMap<HashableValue, HashSet<String>>,
    timeout_nodes: HashSet<String>,
    networker: Rc<RefCell<T>>,
}

struct FullState<T: Networker> {
    accum_reply: Option<HashableValue>,
    nodes_to_send: Option<Vec<String>>,
    networker: Rc<RefCell<T>>,
}

struct FinishState {}

impl<T: Networker> From<StartState<T>> for SingleState<T> {
    fn from(state: StartState<T>) -> Self {
        SingleState {
            denied_nodes: HashSet::new(),
            replies: HashMap::new(),
            timeout_nodes: HashSet::new(),
            networker: state.networker.clone(),
        }
    }
}

impl<T: Networker> From<StartState<T>> for ConsensusState<T> {
    fn from(state: StartState<T>) -> Self {
        ConsensusState {
            denied_nodes: HashSet::new(),
            replies: HashMap::new(),
            timeout_nodes: HashSet::new(),
            networker: state.networker.clone(),
        }
    }
}

impl<T: Networker> From<(MerkleTree, StartState<T>)> for CatchupConsensusState<T> {
    fn from((merkle_tree, state): (MerkleTree, StartState<T>)) -> Self {
        CatchupConsensusState {
            replies: HashMap::new(),
            networker: state.networker.clone(),
            merkle_tree,
        }
    }
}

impl<T: Networker> From<(MerkleTree, StartState<T>, Vec<u8>, usize, String)> for CatchupSingleState<T> {
    fn from((merkle_tree, state, target_mt_root, target_mt_size, req_id): (MerkleTree, StartState<T>, Vec<u8>, usize, String)) -> Self {
        CatchupSingleState {
            target_mt_root,
            target_mt_size,
            networker: state.networker.clone(),
            merkle_tree,
            req_id,
        }
    }
}

impl<T: Networker> From<StartState<T>> for FullState<T> {
    fn from(state: StartState<T>) -> Self {
        FullState {
            accum_reply: None,
            nodes_to_send: None,
            networker: state.networker.clone(),
        }
    }
}

impl<T: Networker> From<(Option<Vec<String>>, StartState<T>)> for FullState<T> {
    fn from((nodes_to_send, state): (Option<Vec<String>>, StartState<T>)) -> Self {
        FullState {
            accum_reply: None,
            nodes_to_send,
            networker: state.networker.clone(),
        }
    }
}

impl<T: Networker> RequestState<T> {
    fn finish() -> RequestState<T> {
        RequestState::Finish(FinishState {})
    }
}

impl<T: Networker> RequestSM<T> {
    fn handle_event(self, re: RequestEvent) -> (Self, Option<PoolEvent>) {
        let RequestSM { state, f, cmd_ids, nodes, generator, pool_name, timeout, extended_timeout } = self;
        let (state, event) = match state {
            RequestState::Start(state) => {
                match re {
                    RequestEvent::LedgerStatus(ls, _, Some(merkle)) => {
                        let req_id = ls.merkleRoot.clone();
                        let ne = Some(NetworkerEvent::SendAllRequest(serde_json::to_string(&super::types::Message::LedgerStatus(ls)).expect("FIXME"),
                                                                     req_id, extended_timeout, None));
                        trace!("start catchup, ne: {:?}", ne);
                        state.networker.borrow_mut().process_event(ne);
                        (RequestState::CatchupConsensus((merkle, state).into()), None)
                    }
                    RequestEvent::CatchupReq(merkle, target_mt_size, target_mt_root) => {
                        match build_catchup_req(&merkle, target_mt_size) {
                            Ok(Some((req_id, req_json))) => {
                                state.networker.borrow_mut().process_event(Some(NetworkerEvent::SendOneRequest(req_json, req_id.clone(), timeout)));
                                (RequestState::CatchupSingle((merkle, state, target_mt_root, target_mt_size, req_id).into()), None)
                            }
                            Ok(None) => {
                                warn!("No transactions to catch up!");
                                (RequestState::finish(), Some(PoolEvent::Synced(merkle)))
                            }
                            Err(e) => {
                                _send_replies(&cmd_ids, Err(e));
                                (RequestState::finish(), None)
                            }
                        }
                    }
                    RequestEvent::CustomSingleRequest(msg, req_id) => {
                        state.networker.borrow_mut().process_event(Some(NetworkerEvent::SendOneRequest(msg, req_id, timeout)));
                        (RequestState::Single(state.into()), None)
                    }
                    RequestEvent::CustomFullRequest(msg, req_id, local_timeout, nodes_to_send) => {
                        let timeout = local_timeout.map(|to| to as i64).unwrap_or(extended_timeout);
                        if let Some(nodes_to_send) = nodes_to_send {
                            match serde_json::from_str::<Vec<String>>(&nodes_to_send) {
                                Ok(nodes_to_send) => {
                                    //TODO check empty list on API level?
                                    let is_nodes_to_send_known = !nodes_to_send.is_empty() && nodes_to_send.iter().all(|node| nodes.contains_key(node));
                                    if is_nodes_to_send_known {
                                        state.networker.borrow_mut().process_event(Some(NetworkerEvent::SendAllRequest(msg, req_id, timeout, Some(nodes_to_send.clone()))));
                                        (RequestState::Full((Some(nodes_to_send), state).into()), None)
                                    } else {
                                        _send_replies(&cmd_ids, Err(err_msg(IndyErrorKind::InvalidStructure,
                                                                            format!("There is no known node in list to send {:?}, known nodes are {:?}",
                                                                                    nodes_to_send, nodes.keys()))));
                                        (RequestState::finish(), None)
                                    }
                                }
                                Err(err) => {
                                    _send_replies(&cmd_ids, Err(err.to_indy(IndyErrorKind::InvalidStructure, "Invalid list of nodes to send")));
                                    (RequestState::finish(), None)
                                }
                            }
                        } else {
                            state.networker.borrow_mut().process_event(Some(NetworkerEvent::SendAllRequest(msg, req_id, timeout, None)));
                            (RequestState::Full((None, state).into()), None)
                        }
                    }
                    RequestEvent::CustomConsensusRequest(msg, req_id) => {
                        state.networker.borrow_mut().process_event(Some(NetworkerEvent::SendAllRequest(msg, req_id, timeout, None)));
                        (RequestState::Consensus(state.into()), None)
                    }
                    _ => {
                        (RequestState::Start(state), None)
                    }
                }
            }
            RequestState::Consensus(mut state) => {
                match re {
                    RequestEvent::Reply(_, raw_msg, node_alias, req_id) |
                    RequestEvent::ReqNACK(_, raw_msg, node_alias, req_id) |
                    RequestEvent::Reject(_, raw_msg, node_alias, req_id)
                        => {
                        if let Ok((_, result_without_proof)) = _get_msg_result_without_state_proof(&raw_msg) {
                            let hashable = HashableValue { inner: result_without_proof };

                            let cnt = {
                                let set = state.replies.entry(hashable).or_insert(HashSet::new());
                                set.insert(node_alias.clone());
                                set.len()
                            };

                            if cnt > f {
                                _send_ok_replies(&cmd_ids, &raw_msg);
                                state.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(req_id, None)));
                                (RequestState::finish(), None)
                            } else if state.is_consensus_reachable(f, nodes.len()) {
                                state.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(req_id, Some(node_alias))));
                                (RequestState::Consensus(state), None)
                            } else {
                                //TODO: maybe we should change the error, but it was made to escape changing of ErrorCode returned to client
                                _send_replies(&cmd_ids, Err(err_msg(IndyErrorKind::PoolTimeout, "Consensus is impossible")));
                                state.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(req_id, None)));
                                (RequestState::finish(), None)
                            }
                        } else {
                            (RequestState::Consensus(state), None)
                        }
                    }
                    RequestEvent::ReqACK(_, _, node_alias, req_id) => {
                        state.networker.borrow_mut().process_event(Some(NetworkerEvent::ExtendTimeout(req_id, node_alias, extended_timeout)));
                        (RequestState::Consensus(state), None)
                    }
                    RequestEvent::Timeout(req_id, node_alias) => {
                        state.timeout_nodes.insert(node_alias.clone());
                        if state.is_consensus_reachable(f, nodes.len()) {
                            state.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(req_id, Some(node_alias))));
                            (RequestState::Consensus(state.into()), None)
                        } else {
                            //TODO: maybe we should change the error, but it was made to escape changing of ErrorCode returned to client
                            _send_replies(&cmd_ids, Err(err_msg(IndyErrorKind::PoolTimeout, "Consensus is impossible")));
                            state.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(req_id, None)));
                            (RequestState::finish(), None)
                        }
                    }
                    RequestEvent::Terminate => {
                        _finish_request(&cmd_ids);
                        (RequestState::finish(), None)
                    }
                    _ => (RequestState::Consensus(state.into()), None)
                }
            }
            RequestState::Single(mut state) => {
                match re {
                    RequestEvent::Reply(_, raw_msg, node_alias, req_id) |
                    RequestEvent::ReqNACK(_, raw_msg, node_alias, req_id) |
                    RequestEvent::Reject(_, raw_msg, node_alias, req_id) => {
                        trace!("reply on single request");
                        state.timeout_nodes.remove(&node_alias);
                        if let Ok((result, result_without_proof)) = _get_msg_result_without_state_proof(&raw_msg) {
                            let hashable = HashableValue { inner: result_without_proof };

                            let cnt = {
                                let set = state.replies.entry(hashable).or_insert(HashSet::new());
                                set.insert(node_alias.clone());
                                set.len()
                            };

                            if cnt > f || _check_state_proof(&result, f, &generator, &nodes, &raw_msg) {
                                state.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(req_id, None)));
                                _send_ok_replies(&cmd_ids, &raw_msg);
                                (RequestState::finish(), None)
                            } else {
                                (state.try_to_continue(req_id, node_alias, &cmd_ids, nodes.len(), timeout), None)
                            }
                        } else {
                            state.denied_nodes.insert(node_alias.clone());
                            (state.try_to_continue(req_id, node_alias, &cmd_ids, nodes.len(), timeout), None)
                        }
                    }
                    RequestEvent::ReqACK(_, _, node_alias, req_id) => {
                        state.networker.borrow_mut().process_event(Some(NetworkerEvent::ExtendTimeout(req_id, node_alias, extended_timeout)));
                        (RequestState::Single(state), None)
                    }
                    RequestEvent::Timeout(req_id, node_alias) => {
                        state.timeout_nodes.insert(node_alias.clone());
                        (state.try_to_continue(req_id, node_alias, &cmd_ids, nodes.len(), timeout), None)
                    }
                    RequestEvent::Terminate => {
                        _finish_request(&cmd_ids);
                        (RequestState::finish(), None)
                    }
                    _ => (RequestState::Single(state), None)
                }
            }
            RequestState::CatchupConsensus(state) => {
                match re {
                    RequestEvent::LedgerStatus(ls, Some(node_alias), _) => {
                        RequestSM::_catchup_target_handle_consensus_state(
                            state,
                            ls.merkleRoot.clone(), ls.txnSeqNo, None,
                            node_alias, ls.merkleRoot, f, &nodes, &pool_name)
                    }
                    RequestEvent::ConsistencyProof(cp, node_alias) => {
                        RequestSM::_catchup_target_handle_consensus_state(
                            state,
                            cp.newMerkleRoot, cp.seqNoEnd, Some(cp.hashes),
                            node_alias, cp.oldMerkleRoot, f, &nodes, &pool_name)
                    }
                    RequestEvent::Timeout(req_id, node_alias) => {
                        RequestSM::_catchup_target_handle_consensus_state(
                            state,
                            "timeout".to_string(), 0, None,
                            node_alias, req_id, f, &nodes, &pool_name)
                    }

                    RequestEvent::Terminate => {
                        _finish_request(&cmd_ids);
                        (RequestState::finish(), None)
                    }
                    _ => (RequestState::CatchupConsensus(state), None)
                }
            }
            RequestState::CatchupSingle(mut state) => {
                match re {
                    RequestEvent::CatchupRep(mut cr, node_alias) => {
                        match _process_catchup_reply(&mut cr, &mut state.merkle_tree, &state.target_mt_root, state.target_mt_size, &pool_name) {
                            Ok(merkle) => {
                                state.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(state.req_id.clone(), None)));
                                (RequestState::finish(), Some(PoolEvent::Synced(merkle)))
                            }
                            Err(_) => {
                                state.networker.borrow_mut().process_event(Some(NetworkerEvent::Resend(state.req_id.clone(), timeout)));
                                state.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(state.req_id.clone(), Some(node_alias))));
                                (RequestState::CatchupSingle(state), None)
                            }
                        }
                    }
                    RequestEvent::Timeout(req_id, node_alias) => {
                        state.networker.borrow_mut().process_event(Some(NetworkerEvent::Resend(state.req_id.clone(), timeout)));
                        state.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(req_id, Some(node_alias))));
                        (RequestState::CatchupSingle(state), None)
                    }
                    RequestEvent::Terminate => {
                        _finish_request(&cmd_ids);
                        (RequestState::finish(), None)
                    }
                    _ => (RequestState::CatchupSingle(state), None)
                }
            }
            RequestState::Full(state) => {
                match re {
                    RequestEvent::Reply(_, raw_msg, node_alias, req_id) |
                    RequestEvent::ReqNACK(_, raw_msg, node_alias, req_id) |
                    RequestEvent::Reject(_, raw_msg, node_alias, req_id) =>
                        (RequestSM::_full_request_handle_consensus_state(
                            state, req_id, node_alias, raw_msg, &cmd_ids, &nodes), None),
                    RequestEvent::Timeout(req_id, node_alias) =>
                        (RequestSM::_full_request_handle_consensus_state(
                            state, req_id, node_alias, "timeout".to_string(), &cmd_ids, &nodes), None),

                    RequestEvent::Terminate => {
                        _finish_request(&cmd_ids);
                        (RequestState::finish(), None)
                    }
                    _ => (RequestState::Full(state), None),
                }
            }
            RequestState::Finish(state) => (RequestState::Finish(state), None)
        };
        (RequestSM::step(f, cmd_ids, nodes, generator, pool_name, timeout, extended_timeout, state), event)
    }

    fn is_terminal(&self) -> bool {
        match self.state {
            RequestState::Start(_) |
            RequestState::Consensus(_) |
            RequestState::Single(_) |
            RequestState::CatchupSingle(_) |
            RequestState::CatchupConsensus(_) |
            RequestState::Full(_) => false,
            RequestState::Finish(_) => true
        }
    }

    fn _full_request_handle_consensus_state(mut state: FullState<T>,
                                            req_id: String, node_alias: String, node_result: String,
                                            cmd_ids: &Vec<i32>,
                                            nodes: &HashMap<String, Option<VerKey>>) -> RequestState<T> {
        let is_first_resp = state.accum_reply.is_none();
        if is_first_resp {
            state.accum_reply = Some(HashableValue {
                inner: json!({node_alias.clone(): node_result})
            })
        } else {
            state.accum_reply.as_mut().unwrap()
                .inner.as_object_mut().unwrap()
                .insert(node_alias.clone(), SJsonValue::from(node_result));
        }

        let required_reply_cnt = state.nodes_to_send.as_ref().map(Vec::len).unwrap_or(nodes.len());

        let reply_cnt = state.accum_reply.as_ref().unwrap()
            .inner.as_object().unwrap().len();

        if reply_cnt == required_reply_cnt {
            state.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(req_id, None)));
            let reply = state.accum_reply.as_ref().unwrap().inner.to_string();
            _send_ok_replies(&cmd_ids, &reply);
            RequestState::Finish(FinishState {})
        } else {
            state.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(req_id, Some(node_alias))));
            RequestState::Full(state)
        }
    }

    fn _catchup_target_handle_consensus_state(mut state: CatchupConsensusState<T>,
                                              mt_root: String, sz: usize, cons_proof: Option<Vec<String>>,
                                              node_alias: String, req_id: String,
                                              f: usize, nodes: &HashMap<String, Option<VerKey>>,
                                              pool_name: &str) -> (RequestState<T>, Option<PoolEvent>) {
        let (finished, result) = RequestSM::_process_catchup_target(mt_root, sz, cons_proof,
                                                                    &node_alias, &mut state, f, nodes, pool_name);
        if finished {
            state.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(req_id, None)));
            (RequestState::finish(), result)
        } else {
            state.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(req_id, Some(node_alias))));
            (RequestState::CatchupConsensus(state), result)
        }
    }

    fn _process_catchup_target(merkle_root: String,
                               txn_seq_no: usize,
                               hashes: Option<Vec<String>>,
                               node_alias: &str,
                               state: &mut CatchupConsensusState<T>,
                               f: usize,
                               nodes: &HashMap<String, Option<VerKey>>,
                               pool_name: &str) -> (bool, Option<PoolEvent>) {
        let key = (merkle_root, txn_seq_no, hashes);
        let contains = state.replies.get_mut(&key)
            .map(|set| { set.insert(node_alias.to_string()); })
            .is_some();

        if !contains {
            state.replies.insert(key, HashSet::from_iter(vec![node_alias.to_string()]));
        }

        match check_nodes_responses_on_status(&state.replies,
                                              &state.merkle_tree,
                                              nodes.len(),
                                              f,
                                              &pool_name) {
            Ok(CatchupProgress::InProgress) => (false, None),
            Ok(CatchupProgress::NotNeeded(merkle_tree)) => (true, Some(PoolEvent::Synced(merkle_tree))),
            Ok(CatchupProgress::ShouldBeStarted(target_mt_root, target_mt_size, merkle_tree)) =>
                (true, Some(PoolEvent::CatchupTargetFound(target_mt_root, target_mt_size, merkle_tree))),
            Err(err) => (true, Some(PoolEvent::CatchupTargetNotFound(err))),
        }
    }
}

pub trait RequestHandler<T: Networker> {
    fn new(networker: Rc<RefCell<T>>, f: usize, cmd_ids: &Vec<i32>, nodes: &HashMap<String, Option<VerKey>>, generator: Option<Generator>, pool_name: &str, timeout: i64, extended_timeout: i64) -> Self;
    fn process_event(&mut self, ore: Option<RequestEvent>) -> Option<PoolEvent>;
    fn is_terminal(&self) -> bool;
}

pub struct RequestHandlerImpl<T: Networker> {
    request_wrapper: Option<RequestSM<T>>
}

impl<T: Networker> RequestHandler<T> for RequestHandlerImpl<T> {
    fn new(networker: Rc<RefCell<T>>, f: usize, cmd_ids: &Vec<i32>, nodes: &HashMap<String, Option<VerKey>>, generator: Option<Generator>, pool_name: &str, timeout: i64, extended_timeout: i64) -> Self {
        RequestHandlerImpl {
            request_wrapper: Some(RequestSM::new(networker, f, cmd_ids, nodes, generator, pool_name, timeout, extended_timeout)),
        }
    }

    fn process_event(&mut self, ore: Option<RequestEvent>) -> Option<PoolEvent> {
        match ore {
            Some(re) => {
                if let Some((rw, res)) = self.request_wrapper.take().map(|w| w.handle_event(re)) {
                    self.request_wrapper = Some(rw);
                    res
                } else {
                    self.request_wrapper = None;
                    None
                }
            }
            None => None
        }
    }

    fn is_terminal(&self) -> bool {
        self.request_wrapper.as_ref().map(|w| w.is_terminal()).unwrap_or(true)
    }
}

impl<T: Networker> SingleState<T> {
    fn is_consensus_reachable(&self, total_nodes_cnt: usize) -> bool {
        (self.timeout_nodes.len() + self.denied_nodes.len() + self.replies.values().map(|set| set.len()).sum::<usize>())
            < total_nodes_cnt
    }

    fn try_to_continue(self, req_id: String, node_alias: String, cmd_ids: &Vec<i32>, nodes_cnt: usize, timeout: i64) -> RequestState<T> {
        if self.is_consensus_reachable(nodes_cnt) {
            self.networker.borrow_mut().process_event(Some(NetworkerEvent::Resend(req_id.clone(), timeout)));
            self.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(req_id, Some(node_alias))));
            RequestState::Single(self)
        } else {
            //TODO: maybe we should change the error, but it was made to escape changing of ErrorCode returned to client
            _send_replies(cmd_ids, Err(err_msg(IndyErrorKind::PoolTimeout, "Consensus is impossible")));
            self.networker.borrow_mut().process_event(Some(NetworkerEvent::CleanTimeout(req_id, None)));
            RequestState::finish()
        }
    }
}

impl<T: Networker> ConsensusState<T> {
    fn is_consensus_reachable(&self, f: usize, total_nodes_cnt: usize) -> bool {
        let rep_no: usize = self.replies.values().map(|set| set.len()).sum();
        let max_no = self.replies.values().map(|set| set.len()).max().unwrap_or(0);
        max_no + total_nodes_cnt - rep_no - self.timeout_nodes.len() - self.denied_nodes.len() > f
    }
}

fn _parse_nack(denied_nodes: &mut HashSet<String>, f: usize, raw_msg: &str, cmd_ids: &Vec<i32>, node_alias: &str) -> bool {
    if denied_nodes.len() == f {
        _send_ok_replies(cmd_ids, raw_msg);
        true
    } else {
        denied_nodes.insert(node_alias.to_string());
        false
    }
}

fn _process_catchup_reply(rep: &mut CatchupRep, merkle: &MerkleTree, target_mt_root: &Vec<u8>, target_mt_size: usize, pool_name: &str) -> IndyResult<MerkleTree> {
    let mut txns_to_drop = vec![];
    let mut merkle = merkle.clone();

    while !rep.txns.is_empty() {
        let key = rep.min_tx()?;
        let txn = rep.txns.remove(&key.to_string()).unwrap();

        let txn = rmp_serde::to_vec_named(&txn)
            .to_indy(IndyErrorKind::InvalidStructure, "Invalid transaction -- can not transform to bytes")?;

        merkle.append(txn.clone())?;
        txns_to_drop.push(txn);
    }

    check_cons_proofs(&merkle, &rep.consProof, target_mt_root, target_mt_size)?;
    merkle_tree_factory::dump_new_txns(pool_name, &txns_to_drop)?;
    Ok(merkle)
}

fn _send_ok_replies(cmd_ids: &Vec<i32>, msg: &str) {
    _send_replies(cmd_ids, Ok(msg.to_string()))
}

fn _finish_request(cmd_ids: &Vec<i32>) {
    _send_replies(cmd_ids, Err(err_msg(IndyErrorKind::PoolTerminated, "Pool is terminated")))
}

fn _send_replies(cmd_ids: &Vec<i32>, msg: IndyResult<String>) {
    cmd_ids.into_iter().for_each(|id| {
        CommandExecutor::instance().send(
            Command::Ledger(
                LedgerCommand::SubmitAck(id.clone(), msg.clone()))
        ).unwrap();
    });
}

fn _get_msg_result_without_state_proof(msg: &str) -> IndyResult<(SJsonValue, SJsonValue)> {
    let msg = serde_json::from_str::<SJsonValue>(msg)
        .to_indy(IndyErrorKind::InvalidStructure, "Response is malformed json")?;

    let msg_result = msg["result"].clone();

    let mut msg_result_without_proof: SJsonValue = msg_result.clone();
    msg_result_without_proof.as_object_mut().map(|obj| obj.remove("state_proof"));

    if msg_result_without_proof["data"].is_object() {
        msg_result_without_proof["data"].as_object_mut().map(|obj| obj.remove("stateProofFrom"));
    }

    Ok((msg_result, msg_result_without_proof))
}

fn _check_state_proof(msg_result: &SJsonValue, f: usize, gen: &Generator, bls_keys: &HashMap<String, Option<VerKey>>, raw_msg: &str) -> bool {
    debug!("TransactionHandler::process_reply: Try to verify proof and signature >>");

    let res = match state_proof::parse_generic_reply_for_proof_checking(&msg_result, raw_msg) {
        Some(parsed_sps) => {
            debug!("TransactionHandler::process_reply: Proof and signature are present");
            state_proof::verify_parsed_sp(parsed_sps, bls_keys, f, gen)
        }
        None => false
    };

    debug!("TransactionHandler::process_reply: Try to verify proof and signature << {}", res);
    res
}

#[cfg(test)]
pub mod tests {
    use services::ledger::merkletree::tree::Tree;
    use services::pool::networker::MockNetworker;
    use services::pool::types::{ConsistencyProof, LedgerStatus, Reply, ReplyResultV1, ReplyTxnV1, ReplyV1, Response, ResponseMetadata, ResponseV1};
    use utils::test;

    use super::*;

    const MESSAGE: &'static str = "message";
    const REQ_ID: &'static str = "1";
    const NODE: &'static str = "n1";
    const NODE_2: &'static str = "n2";
    const NODE_3: &'static str = "n3";
    const NODE_4: &'static str = "n4";
    const SIMPLE_REPLY: &'static str = r#"{"result":{}}"#;
    const REJECT_REPLY: &'static str = r#"{"op":"REJECT"}"#;
    const NACK_REPLY: &'static str = r#"{"op":"REQNACK"}"#;
    const POOL: &'static str = "pool1";

    #[derive(Debug)]
    pub struct MockRequestHandler {}

    impl<T: Networker> RequestHandler<T> for MockRequestHandler {
        fn new(_networker: Rc<RefCell<T>>, _f: usize, _cmd_ids: &Vec<i32>, _nodes: &HashMap<String, Option<VerKey>>, _generator: Option<Generator>, _pool_name: &str, _timeout: i64, _extended_timeout: i64) -> Self {
            MockRequestHandler {}
        }

        fn process_event(&mut self, _ore: Option<RequestEvent>) -> Option<PoolEvent> {
            None
        }

        fn is_terminal(&self) -> bool {
            true
        }
    }

    impl Default for LedgerStatus {
        fn default() -> Self {
            LedgerStatus {
                txnSeqNo: 0,
                merkleRoot: String::new(),
                ledgerId: 0,
                ppSeqNo: None,
                viewNo: None,
                protocolVersion: None,
            }
        }
    }

    impl Default for MerkleTree {
        fn default() -> Self {
            MerkleTree {
                root: Tree::Empty { hash: Vec::new() },
                height: 0,
                count: 0,
                nodes_count: 0,
            }
        }
    }

    impl Default for Reply {
        fn default() -> Self {
            Reply::ReplyV1(ReplyV1 { result: ReplyResultV1 { txn: ReplyTxnV1 { metadata: ResponseMetadata { req_id: 1 } } } })
        }
    }

    impl Default for Response {
        fn default() -> Self {
            Response::ResponseV1(ResponseV1 { metadata: ResponseMetadata { req_id: 1 } })
        }
    }

    impl Default for ConsistencyProof {
        fn default() -> Self {
            ConsistencyProof {
                seqNoEnd: 0,
                seqNoStart: 0,
                ledgerId: 0,
                hashes: Vec::new(),
                oldMerkleRoot: String::new(),
                newMerkleRoot: String::new(),
            }
        }
    }

    impl Default for CatchupRep {
        fn default() -> Self {
            CatchupRep {
                ledgerId: 0,
                consProof: Vec::new(),
                txns: HashMap::new(),
            }
        }
    }

    fn _request_handler(f: usize, nodes_cnt: usize) -> RequestHandlerImpl<MockNetworker> {
        let networker = Rc::new(RefCell::new(MockNetworker::new(0, 0, vec![])));

        let mut default_nodes: HashMap<String, Option<VerKey>> = HashMap::new();
        default_nodes.insert(NODE.to_string(), None);

        let node_names = vec![NODE, NODE_2, "n3", "n4"];
        let mut nodes: HashMap<String, Option<VerKey>> = HashMap::new();

        for i in 0..nodes_cnt {
            nodes.insert(node_names[i].to_string(), None);
        }

        RequestHandlerImpl::new(networker,
                                f,
                                &vec![],
                                &nodes,
                                None,
                                POOL,
                                0,
                                0)
    }

    // required because of dumping txns to cache
    fn _create_pool(content: Option<String>) {
        use utils::environment;
        use std::fs;
        use std::fs::File;
        use std::io::Write;

        let mut path = environment::pool_path(POOL);

        path.push(POOL);
        path.set_extension("txn");

        fs::create_dir_all(path.parent().unwrap()).unwrap();

        let mut file = File::create(path).unwrap();
        file.write_all(content.unwrap_or("{}".to_string()).as_bytes()).unwrap();
    }

    #[test]
    fn request_handler_new_works() {
        let request_handler = _request_handler(0, 1);
        assert_match!(RequestState::Start(_), request_handler.request_wrapper.unwrap().state);
    }

    #[test]
    fn request_handler_process_event_works() {
        let mut request_handler = _request_handler(0, 1);
        request_handler.process_event(None);
    }

    mod start {
        use super::*;

        #[test]
        fn request_handler_process_ledger_status_event_from_start_works() {
            let mut request_handler = _request_handler(0, 1);
            request_handler.process_event(Some(RequestEvent::LedgerStatus(LedgerStatus::default(), Some(NODE.to_string()), Some(MerkleTree::default()))));
            assert_match!(RequestState::CatchupConsensus(_), request_handler.request_wrapper.unwrap().state);
        }

        #[test]
        fn request_handler_process_catchup_req_event_from_start_works() {
            let mut request_handler = _request_handler(0, 1);
            request_handler.process_event(Some(RequestEvent::CatchupReq(MerkleTree::default(), 1, vec![])));
            assert_match!(RequestState::CatchupSingle(_), request_handler.request_wrapper.unwrap().state);
        }

        #[test]
        fn request_handler_process_catchup_req_event_from_start_works_for_no_transactions_to_catchup() {
            let mut request_handler = _request_handler(0, 1);
            request_handler.process_event(Some(RequestEvent::CatchupReq(MerkleTree::default(), 0, vec![])));
            assert_match!(RequestState::Finish(_), request_handler.request_wrapper.unwrap().state);
        }

        #[test]
        fn request_handler_process_custom_single_req_event_from_start_works() {
            let mut request_handler = _request_handler(0, 1);
            request_handler.process_event(Some(RequestEvent::CustomSingleRequest(MESSAGE.to_string(), REQ_ID.to_string())));
            assert_match!(RequestState::Single(_), request_handler.request_wrapper.unwrap().state);
        }

        #[test]
        fn request_handler_process_consensus_full_req_event_from_start_works() {
            let mut request_handler = _request_handler(0, 1);
            request_handler.process_event(Some(RequestEvent::CustomFullRequest(MESSAGE.to_string(), REQ_ID.to_string(), None, None)));
            assert_match!(RequestState::Full(_), request_handler.request_wrapper.unwrap().state);
        }

        #[test]
        fn request_handler_process_consensus_full_req_event_from_start_works_for_list_nodes() {
            let mut request_handler = _request_handler(0, 1);
            request_handler.process_event(Some(RequestEvent::CustomFullRequest(MESSAGE.to_string(), REQ_ID.to_string(), None, Some(format!(r#"["{}"]"#, NODE)))));
            assert_match!(RequestState::Full(_), request_handler.request_wrapper.unwrap().state);
        }

        #[test]
        fn request_handler_process_consensus_full_req_event_from_start_works_for_empty_list_nodes() {
            let mut request_handler = _request_handler(0, 1);
            request_handler.process_event(Some(RequestEvent::CustomFullRequest(MESSAGE.to_string(), REQ_ID.to_string(), None, Some("[ ]".to_string()))));
            assert_match!(RequestState::Finish(_), request_handler.request_wrapper.unwrap().state);
        }

        #[test]
        fn request_handler_process_consensus_full_req_event_from_start_works_for_list_nodes_contains_unknown_node() {
            let mut request_handler = _request_handler(0, 1);
            request_handler.process_event(Some(RequestEvent::CustomFullRequest(MESSAGE.to_string(), REQ_ID.to_string(), None, Some("[Unknown Node]".to_string()))));
            assert_match!(RequestState::Finish(_), request_handler.request_wrapper.unwrap().state);
        }

        #[test]
        fn request_handler_process_consensus_full_req_event_from_start_works_for_invalid_list_nodes_format() {
            let mut request_handler = _request_handler(0, 1);
            request_handler.process_event(Some(RequestEvent::CustomFullRequest(MESSAGE.to_string(), REQ_ID.to_string(), None, Some(format!(r#""{}""#, NODE)))));
            assert_match!(RequestState::Finish(_), request_handler.request_wrapper.unwrap().state);
        }

        #[test]
        fn request_handler_process_custom_consensus_req_event_from_start_works() {
            let mut request_handler = _request_handler(0, 1);
            request_handler.process_event(Some(RequestEvent::CustomConsensusRequest(MESSAGE.to_string(), REQ_ID.to_string())));
            assert_match!(RequestState::Consensus(_), request_handler.request_wrapper.unwrap().state);
        }

        #[test]
        fn request_handler_process_other_event_from_start_works() {
            let mut request_handler = _request_handler(0, 1);
            request_handler.process_event(Some(RequestEvent::Timeout(REQ_ID.to_string(), NODE.to_string())));
            assert_match!(RequestState::Start(_), request_handler.request_wrapper.unwrap().state);
        }
    }

    mod consensus_state {
        use super::*;

        #[test]
        fn request_handler_process_reply_event_from_consensus_state_works_for_consensus_reached() {
            let mut request_handler = _request_handler(0, 1);
            request_handler.process_event(Some(RequestEvent::CustomConsensusRequest(MESSAGE.to_string(), REQ_ID.to_string())));
            request_handler.process_event(Some(RequestEvent::Reply(Reply::default(), SIMPLE_REPLY.to_string(), NODE.to_string(), REQ_ID.to_string())));
            assert_match!(RequestState::Finish(_), request_handler.request_wrapper.unwrap().state);
        }

        #[test]
        fn request_handler_process_reply_event_from_consensus_state_works_for_consensus_reached_with_mixed_msgs() {

            // the test will use 4 nodes, each node replying with a response to the "custom consensus request" message
            // some nodes accept, some reject and some nack.  the end result is consensus should not be reached
            let mut request_handler = _request_handler(1, 4);

            request_handler.process_event(Some(RequestEvent::CustomConsensusRequest(MESSAGE.to_string(), REQ_ID.to_string())));
            request_handler.process_event(Some(RequestEvent::Reply(Reply::default(), SIMPLE_REPLY.to_string(), NODE.to_string(), REQ_ID.to_string())));
            request_handler.process_event(Some(RequestEvent::Reject(Response::default(), REJECT_REPLY.to_string(), NODE_2.to_string(), REQ_ID.to_string())));
            request_handler.process_event(Some(RequestEvent::ReqNACK(Response::default(), NACK_REPLY.to_string(), NODE_3.to_string(), REQ_ID.to_string())));

            // test state is already Finished because already have 2 nodes not in consensus
            {
                let request_handler_ref = request_handler.request_wrapper.as_ref().unwrap();
                assert_match!(RequestState::Finish(_), request_handler_ref.state);
            }

            // send one more message to ensure state isn't affected
            request_handler.process_event(Some(RequestEvent::Reject(Response::default(), "{}".to_string(), NODE_4.to_string(), REQ_ID.to_string())));

            assert_match!(RequestState::Finish(_), request_handler.request_wrapper.unwrap().state);

        }

        // this test is marked ignore until https://jira.hyperledger.org/browse/IS-1137 is resolved
        #[test]
        #[ignore]
        fn request_handler_process_reply_event_from_consensus_state_works_for_consensus_reached_with_0_concensus() {

            // the test will use 4 nodes, each node replying with a response to the "custom consensus request" message
            // some nodes accept, some reject and some nack.  the end result is consensus should not be reached
            let mut request_handler = _request_handler(0, 4);

            request_handler.process_event(Some(RequestEvent::CustomConsensusRequest(MESSAGE.to_string(), REQ_ID.to_string())));
            request_handler.process_event(Some(RequestEvent::Reply(Reply::default(), SIMPLE_REPLY.to_string(), NODE.to_string(), REQ_ID.to_string())));
            request_handler.process_event(Some(RequestEvent::Reject(Response::default(), "".to_string(), NODE_2.to_string(), REQ_ID.to_string())));
            request_handler.process_event(Some(RequestEvent::ReqNACK(Response::default(), "".to_string(), NODE_3.to_string(), REQ_ID.to_string())));

            // test state is already Finished because already have 2 nodes not in consensus
            {
                let request_handler_ref = request_handler.request_wrapper.as_ref().unwrap();
                assert_match!(RequestState::Finish(_), request_handler_ref.state);
            }

            request_handler.process_event(Some(RequestEvent::Reject(Response::default(), "{}".to_string(), NODE_4.to_string(), REQ_ID.to_string())));

            assert_match!(RequestState::Finish(_), request_handler.request_wrapper.unwrap().state);

        }

{'fraction_non_alphanumeric': 0.11878453038674033, 'fraction_numerical': 0.0038179939810447827, 'mean_word_length': 2.0475685294822217, 'pattern_counts': {'":': 4, '<': 365, '<?xml version=': 0, '>': 563, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 102, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3823110', 'n_tokens_mistral': 25588, 'n_tokens_neox': 25137, 'n_words': 4120}