fn run_thread<B: BlockT + 'static>(
	protocol_sender: Sender<FromNetworkMsg<B>>,
	network_service: Arc<Mutex<NetworkService<Message<B>>>>,
	network_port: NetworkPort<B>,
	peerset: PeersetHandle,
) -> impl Future<Item = (), Error = io::Error> {

	let network_service_2 = network_service.clone();

	// Protocol produces a stream of messages about what happens in sync.
	let protocol = stream::poll_fn(move || {
		match network_port.take_one_message() {
			Ok(Some(message)) => Ok(Async::Ready(Some(message))),
			Ok(None) => Ok(Async::NotReady),
			Err(_) => Err(())
		}
	}).for_each(move |msg| {
		// Handle message from Protocol.
		match msg {
			NetworkMsg::Outgoing(who, outgoing_message) => {
				network_service_2
					.lock()
					.send_custom_message(&who, outgoing_message);
			},
			NetworkMsg::ReportPeer(who, severity) => {
				match severity {
					Severity::Bad(message) => {
						debug!(target: "sync", "Banning {:?} because {:?}", who, message);
						network_service_2.lock().drop_node(&who);
						// temporary: make sure the peer gets dropped from the peerset
						peerset.report_peer(who, i32::min_value());
					},
					Severity::Useless(message) => {
						debug!(target: "sync", "Dropping {:?} because {:?}", who, message);
						network_service_2.lock().drop_node(&who)
					},
					Severity::Timeout => {
						debug!(target: "sync", "Dropping {:?} because it timed out", who);
						network_service_2.lock().drop_node(&who)
					},
				}
			},
			#[cfg(any(test, feature = "test-helpers"))]
			NetworkMsg::Synchronized => (),
		}
		Ok(())
	})
	.then(|res| {
		match res {
			Ok(()) => (),
			Err(_) => error!("Protocol disconnected"),
		};
		Ok(())
	});

	// The network service produces events about what happens on the network. Let's process them.
	let network = stream::poll_fn(move || network_service.lock().poll()).for_each(move |event| {
		match event {
			NetworkServiceEvent::OpenedCustomProtocol { peer_id, version, debug_info, .. } => {
				debug_assert_eq!(version, protocol::CURRENT_VERSION as u8);
				let _ = protocol_sender.send(FromNetworkMsg::PeerConnected(peer_id, debug_info));
			}
			NetworkServiceEvent::ClosedCustomProtocol { peer_id, debug_info, .. } => {
				let _ = protocol_sender.send(FromNetworkMsg::PeerDisconnected(peer_id, debug_info));
			}
			NetworkServiceEvent::CustomMessage { peer_id, message, .. } => {
				let _ = protocol_sender.send(FromNetworkMsg::CustomMessage(peer_id, message));
				return Ok(())
			}
			NetworkServiceEvent::Clogged { peer_id, messages, .. } => {
				debug!(target: "sync", "{} clogging messages:", messages.len());
				for msg in messages.into_iter().take(5) {
					debug!(target: "sync", "{:?}", msg);
					let _ = protocol_sender.send(FromNetworkMsg::PeerClogged(peer_id.clone(), Some(msg)));
				}
			}
		};
		Ok(())
	});

	// Merge all futures into one.
	let futures: Vec<Box<Future<Item = (), Error = io::Error> + Send>> = vec![
		Box::new(protocol) as Box<_>,
		Box::new(network) as Box<_>
	];

	futures::select_all(futures)
		.and_then(move |_| {
			debug!("Networking ended");
			Ok(())
		})
		.map_err(|(r, _, _)| r)
}
/// Runs the background thread that handles the networking.
Runs the background thread that handles the networking.
Runs the background thread that handles the networking .
fn run_thread<B: BlockT + 'static>(
	protocol_sender: Sender<FromNetworkMsg<B>>,
	network_service: Arc<Mutex<NetworkService<Message<B>>>>,
	network_port: NetworkPort<B>,
	peerset: PeersetHandle,
) -> impl Future<Item = (), Error = io::Error> {
	let network_service_2 = network_service.clone();
	// Protocol produces a stream of messages about what happens in sync.
	let protocol = stream::poll_fn(move || {
		match network_port.take_one_message() {
			Ok(Some(message)) => Ok(Async::Ready(Some(message))),
			Ok(None) => Ok(Async::NotReady),
			Err(_) => Err(())
		}
	}).for_each(move |msg| {
		// Handle message from Protocol.
		match msg {
			NetworkMsg::Outgoing(who, outgoing_message) => {
				network_service_2
					.lock()
					.send_custom_message(&who, outgoing_message);
			},
			NetworkMsg::ReportPeer(who, severity) => {
				match severity {
					Severity::Bad(message) => {
						debug!(target: "sync", "Banning {:?} because {:?}", who, message);
						network_service_2.lock().drop_node(&who);
						// temporary: make sure the peer gets dropped from the peerset
						peerset.report_peer(who, i32::min_value());
					},
					Severity::Useless(message) => {
						debug!(target: "sync", "Dropping {:?} because {:?}", who, message);
						network_service_2.lock().drop_node(&who)
					},
					Severity::Timeout => {
						debug!(target: "sync", "Dropping {:?} because it timed out", who);
						network_service_2.lock().drop_node(&who)
					},
				}
			},
			#[cfg(any(test, feature = "test-helpers"))]
			NetworkMsg::Synchronized => (),
		}
		Ok(())
	})
	.then(|res| {
		match res {
			Ok(()) => (),
			Err(_) => error!("Protocol disconnected"),
		};
		Ok(())
	});
	// The network service produces events about what happens on the network. Let's process them.
	let network = stream::poll_fn(move || network_service.lock().poll()).for_each(move |event| {
		match event {
			NetworkServiceEvent::OpenedCustomProtocol { peer_id, version, debug_info, .. } => {
				debug_assert_eq!(version, protocol::CURRENT_VERSION as u8);
				let _ = protocol_sender.send(FromNetworkMsg::PeerConnected(peer_id, debug_info));
			}
			NetworkServiceEvent::ClosedCustomProtocol { peer_id, debug_info, .. } => {
				let _ = protocol_sender.send(FromNetworkMsg::PeerDisconnected(peer_id, debug_info));
			}
			NetworkServiceEvent::CustomMessage { peer_id, message, .. } => {
				let _ = protocol_sender.send(FromNetworkMsg::CustomMessage(peer_id, message));
				return Ok(())
			}
			NetworkServiceEvent::Clogged { peer_id, messages, .. } => {
				debug!(target: "sync", "{} clogging messages:", messages.len());
				for msg in messages.into_iter().take(5) {
					debug!(target: "sync", "{:?}", msg);
					let _ = protocol_sender.send(FromNetworkMsg::PeerClogged(peer_id.clone(), Some(msg)));
				}
			}
		};
		Ok(())
	});
	// Merge all futures into one.
	let futures: Vec<Box<Future<Item = (), Error = io::Error> + Send>> = vec![
		Box::new(protocol) as Box<_>,
		Box::new(network) as Box<_>
	];
	futures::select_all(futures)
		.and_then(move |_| {
			debug!("Networking ended");
			Ok(())
		})
		.map_err(|(r, _, _)| r)
}
fn run_thread < B : BlockT + ' static > ( protocol_sender : Sender < FromNetworkMsg < B > > , network_service : Arc < Mutex < NetworkService < Message < B > > > > , network_port : NetworkPort < B > , peerset : PeersetHandle , ) -> impl Future < Item = ( ) , Error = io :: Error > { let network_service_2 = network_service . clone ( ) ; // Protocol produces a stream of messages about what happens in sync. let protocol = stream :: poll_fn ( move | | { match network_port . take_one_message ( ) { Ok ( Some ( message ) ) => Ok ( Async :: Ready ( Some ( message ) ) ) , Ok ( None ) => Ok ( Async :: NotReady ) , Err ( _ ) => Err ( ( ) ) } } ) . for_each ( move | msg | { // Handle message from Protocol. match msg { NetworkMsg :: Outgoing ( who , outgoing_message ) => { network_service_2 . lock ( ) . send_custom_message ( & who , outgoing_message ) ; } , NetworkMsg :: ReportPeer ( who , severity ) => { match severity { Severity :: Bad ( message ) => { debug ! ( target " " " " who message ) ; network_service_2 . lock ( ) . drop_node ( & who ) ; // temporary: make sure the peer gets dropped from the peerset peerset . report_peer ( who , i32 :: min_value ( ) ) ; } , Severity :: Useless ( message ) => { debug ! ( target " " " " who message ) ; network_service_2 . lock ( ) . drop_node ( & who ) } , Severity :: Timeout => { debug ! ( target " " " " who ) ; network_service_2 . lock ( ) . drop_node ( & who ) } , } } , # [ cfg ( any ( test feature " " ) ) ] NetworkMsg :: Synchronized => ( ) , } Ok ( ( ) ) } ) . then ( | res | { match res { Ok ( ( ) ) => ( ) , Err ( _ ) => error ! ( " " ) , } ; Ok ( ( ) ) } ) ; // The network service produces events about what happens on the network. Let's process them. let network = stream :: poll_fn ( move | | network_service . lock ( ) . poll ( ) ) . for_each ( move | event | { match event { NetworkServiceEvent :: OpenedCustomProtocol { peer_id , version , debug_info , .. } => { debug_assert_eq ! ( version protocol CURRENT_VERSION as u8 ) ; let _ = protocol_sender . send ( FromNetworkMsg :: PeerConnected ( peer_id , debug_info ) ) ; } NetworkServiceEvent :: ClosedCustomProtocol { peer_id , debug_info , .. } => { let _ = protocol_sender . send ( FromNetworkMsg :: PeerDisconnected ( peer_id , debug_info ) ) ; } NetworkServiceEvent :: CustomMessage { peer_id , message , .. } => { let _ = protocol_sender . send ( FromNetworkMsg :: CustomMessage ( peer_id , message ) ) ; return Ok ( ( ) ) } NetworkServiceEvent :: Clogged { peer_id , messages , .. } => { debug ! ( target " " " " messages len ( ) ) ; for msg in messages . into_iter ( ) . take ( 5 ) { debug ! ( target " " " " msg ) ; let _ = protocol_sender . send ( FromNetworkMsg :: PeerClogged ( peer_id . clone ( ) , Some ( msg ) ) ) ; } } } ; Ok ( ( ) ) } ) ; // Merge all futures into one. let futures : Vec < Box < Future < Item = ( ) , Error = io :: Error > + Send > > = vec ! [ Box new ( protocol ) as Box Box new ( network ) as Box ] ; futures :: select_all ( futures ) . and_then ( move | _ | { debug ! ( " " ) ; Ok ( ( ) ) } ) . map_err ( | ( r , _ , _ ) | r ) }
Runs the background thread that handles the networking.
Runs the background thread that handles the networking .
protocol_sender
Sender<FromNetworkMsg<B>>
network_service
Arc<Mutex<NetworkService<Message<B>>>>
network_port
NetworkPort<B>
peerset
PeersetHandle
protocol_sender
Sender<FromNetworkMsg<B>>
network_service
Arc<Mutex<NetworkService<Message<B>>>>
network_port
NetworkPort<B>
peerset
PeersetHandle