fn reader_thread(
        network: Network,
        connection: TcpStream,
        reader_thread_responses: Arc<RwLock<ResponsesMap>>,
        reader_thread_writer: Arc<Mutex<TcpStream>>,
        reader_thread_mempool: Arc<Mempool>,
        reader_thread_connected: Arc<RwLock<bool>>,
    ) {
        macro_rules! check_disconnect {
            ($call:expr) => {
                match $call {
                    Ok(good) => good,
                    Err(e) => {
                        log::debug!("Error {:?}", e);
                        *reader_thread_connected.write().unwrap() = false;

                        break;
                    }
                }
            };
        }

        let mut reader = StreamReader::new(connection, None);
        loop {
            let raw_message: RawNetworkMessage = check_disconnect!(reader.read_next());

            let in_message = if raw_message.magic != network.magic() {
                continue;
            } else {
                raw_message.payload
            };

            log::trace!("<== {:?}", in_message);

            match in_message {
                NetworkMessage::Ping(nonce) => {
                    check_disconnect!(Self::_send(
                        &mut reader_thread_writer.lock().unwrap(),
                        network.magic(),
                        NetworkMessage::Pong(nonce),
                    ));

                    continue;
                }
                NetworkMessage::Alert(_) => continue,
                NetworkMessage::GetData(ref inv) => {
                    let (found, not_found): (Vec<_>, Vec<_>) = inv
                        .iter()
                        .map(|item| (*item, reader_thread_mempool.get_tx(item)))
                        .partition(|(_, d)| d.is_some());
                    for (_, found_tx) in found {
                        check_disconnect!(Self::_send(
                            &mut reader_thread_writer.lock().unwrap(),
                            network.magic(),
                            NetworkMessage::Tx(found_tx.unwrap()),
                        ));
                    }

                    if !not_found.is_empty() {
                        check_disconnect!(Self::_send(
                            &mut reader_thread_writer.lock().unwrap(),
                            network.magic(),
                            NetworkMessage::NotFound(
                                not_found.into_iter().map(|(i, _)| i).collect(),
                            ),
                        ));
                    }
                }
                _ => {}
            }

            let message_resp = {
                let mut lock = reader_thread_responses.write().unwrap();
                let message_resp = lock.entry(in_message.cmd()).or_default();
                Arc::clone(&message_resp)
            };

            let (lock, cvar) = &*message_resp;
            let mut messages = lock.lock().unwrap();
            messages.push(in_message);
            cvar.notify_all();
        }
    }
/// Internal function called once the `reader_thread` is spawned
Internal function called once the `reader_thread` is spawned
Internal function called once the ` reader_thread ` is spawned
fn reader_thread(
        network: Network,
        connection: TcpStream,
        reader_thread_responses: Arc<RwLock<ResponsesMap>>,
        reader_thread_writer: Arc<Mutex<TcpStream>>,
        reader_thread_mempool: Arc<Mempool>,
        reader_thread_connected: Arc<RwLock<bool>>,
    ) {
        macro_rules! check_disconnect {
            ($call:expr) => {
                match $call {
                    Ok(good) => good,
                    Err(e) => {
                        log::debug!("Error {:?}", e);
                        *reader_thread_connected.write().unwrap() = false;
                        break;
                    }
                }
            };
        }
        let mut reader = StreamReader::new(connection, None);
        loop {
            let raw_message: RawNetworkMessage = check_disconnect!(reader.read_next());
            let in_message = if raw_message.magic != network.magic() {
                continue;
            } else {
                raw_message.payload
            };
            log::trace!("<== {:?}", in_message);
            match in_message {
                NetworkMessage::Ping(nonce) => {
                    check_disconnect!(Self::_send(
                        &mut reader_thread_writer.lock().unwrap(),
                        network.magic(),
                        NetworkMessage::Pong(nonce),
                    ));
                    continue;
                }
                NetworkMessage::Alert(_) => continue,
                NetworkMessage::GetData(ref inv) => {
                    let (found, not_found): (Vec<_>, Vec<_>) = inv
                        .iter()
                        .map(|item| (*item, reader_thread_mempool.get_tx(item)))
                        .partition(|(_, d)| d.is_some());
                    for (_, found_tx) in found {
                        check_disconnect!(Self::_send(
                            &mut reader_thread_writer.lock().unwrap(),
                            network.magic(),
                            NetworkMessage::Tx(found_tx.unwrap()),
                        ));
                    }
                    if !not_found.is_empty() {
                        check_disconnect!(Self::_send(
                            &mut reader_thread_writer.lock().unwrap(),
                            network.magic(),
                            NetworkMessage::NotFound(
                                not_found.into_iter().map(|(i, _)| i).collect(),
                            ),
                        ));
                    }
                }
                _ => {}
            }
            let message_resp = {
                let mut lock = reader_thread_responses.write().unwrap();
                let message_resp = lock.entry(in_message.cmd()).or_default();
                Arc::clone(&message_resp)
            };
            let (lock, cvar) = &*message_resp;
            let mut messages = lock.lock().unwrap();
            messages.push(in_message);
            cvar.notify_all();
        }
    }
fn reader_thread ( network : Network , connection : TcpStream , reader_thread_responses : Arc < RwLock < ResponsesMap > > , reader_thread_writer : Arc < Mutex < TcpStream > > , reader_thread_mempool : Arc < Mempool > , reader_thread_connected : Arc < RwLock < bool > > , ) { macro_rules! check_disconnect { ( $call : expr ) => { match $call { Ok ( good ) good Err ( e ) { log debug ( " " e ) reader_thread_connected write ( ) unwrap ( ) false break } } } ; } let mut reader = StreamReader :: new ( connection , None ) ; loop { let raw_message : RawNetworkMessage = check_disconnect ! ( reader read_next ( ) ) ; let in_message = if raw_message . magic != network . magic ( ) { continue ; } else { raw_message . payload } ; log :: trace ! ( " " in_message ) ; match in_message { NetworkMessage :: Ping ( nonce ) => { check_disconnect ! ( Self send ( mut reader_thread_writer lock ( ) unwrap ( ) network magic ( ) NetworkMessage Pong ( nonce ) ) ) ; continue ; } NetworkMessage :: Alert ( _ ) => continue , NetworkMessage :: GetData ( ref inv ) => { let ( found , not_found ) : ( Vec < _ > , Vec < _ > ) = inv . iter ( ) . map ( | item | ( * item , reader_thread_mempool . get_tx ( item ) ) ) . partition ( | ( _ , d ) | d . is_some ( ) ) ; for ( _ , found_tx ) in found { check_disconnect ! ( Self send ( mut reader_thread_writer lock ( ) unwrap ( ) network magic ( ) NetworkMessage Tx ( found_tx unwrap ( ) ) ) ) ; } if ! not_found . is_empty ( ) { check_disconnect ! ( Self send ( mut reader_thread_writer lock ( ) unwrap ( ) network magic ( ) NetworkMessage NotFound ( not_found into_iter ( ) map ( ( i ) i ) collect ( ) ) ) ) ; } } _ => { } } let message_resp = { let mut lock = reader_thread_responses . write ( ) . unwrap ( ) ; let message_resp = lock . entry ( in_message . cmd ( ) ) . or_default ( ) ; Arc :: clone ( & message_resp ) } ; let ( lock , cvar ) = & * message_resp ; let mut messages = lock . lock ( ) . unwrap ( ) ; messages . push ( in_message ) ; cvar . notify_all ( ) ; } }
Internal function called once the `reader_thread` is spawned
Internal function called once the ` reader_thread ` is spawned
network
Network
connection
TcpStream
reader_thread_responses
Arc<RwLock<ResponsesMap>>
reader_thread_writer
Arc<Mutex<TcpStream>>
reader_thread_mempool
Arc<Mempool>
reader_thread_connected
Arc<RwLock<bool>>
network
Network
connection
TcpStream
reader_thread_responses
Arc<RwLock<ResponsesMap>>
reader_thread_writer
Arc<Mutex<TcpStream>>
reader_thread_mempool
Arc<Mempool>
reader_thread_connected
Arc<RwLock<bool>>