fn read_msg<F, R>(
		&self,
		sender: UnboundedSender<Vec<u8>>,
		reader: R,
		handler: F,
		pool: CpuPool,
	) -> Box<Future<Item = R, Error = Error>>
	where
		F: Handler + 'static,
		R: AsyncRead + Send + 'static,
	{
		// infinite iterator stream so we repeat the message reading logic until the
		// peer is stopped
		let iter = stream::iter_ok(iter::repeat(()).map(Ok::<(), Error>));

		// setup the reading future, getting messages from the peer and processing them
		let recv_bytes = self.received_bytes.clone();
		let handler = Arc::new(handler);

		let mut count = 0;

		let read_msg = iter.buffered(1).fold(reader, move |reader, _| {
			count += 1;
			trace!(LOGGER, "read_msg: count (per buffered fold): {}", count);

			let recv_bytes = recv_bytes.clone();
			let handler = handler.clone();
			let sender_inner = sender.clone();
			let pool = pool.clone();

			// first read the message header
			read_exact(reader, vec![0u8; HEADER_LEN as usize])
				.from_err()
				.and_then(move |(reader, buf)| {
					trace!(LOGGER, "read_msg: start");

					let header = try!(ser::deserialize::<MsgHeader>(&mut &buf[..]));
					Ok((reader, header))
				})
				.and_then(move |(reader, header)| {
					// now that we have a size, proceed with the body
					read_exact(reader, vec![0u8; header.msg_len as usize])
						.map(|(reader, buf)| (reader, header, buf))
						.from_err()
				})
				.and_then(move |(reader, header, buf)| {
					// add the count of bytes received
					let mut recv_bytes = recv_bytes.lock().unwrap();
					*recv_bytes += header.serialized_len() + header.msg_len;

					pool.spawn_fn(move || {
						let msg_type = header.msg_type;
						if let Err(e) = handler.handle(sender_inner.clone(), header, buf) {
							debug!(LOGGER, "Invalid {:?} message: {}", msg_type, e);
							return Err(Error::Serialization(e));
						}

						trace!(LOGGER, "read_msg: done (via cpu_pool)");
						Ok(reader)
					})
				})
		});
		Box::new(read_msg)
	}
/// Prepares the future reading from the peer connection, parsing each
/// message and forwarding them appropriately based on their type
Prepares the future reading from the peer connection, parsing each
message and forwarding them appropriately based on their type
Prepares the future reading from the peer connection parsing each message and forwarding them appropriately based on their type
fn read_msg<F, R>(
		&self,
		sender: UnboundedSender<Vec<u8>>,
		reader: R,
		handler: F,
		pool: CpuPool,
	) -> Box<Future<Item = R, Error = Error>>
	where
		F: Handler + 'static,
		R: AsyncRead + Send + 'static,
	{
		// infinite iterator stream so we repeat the message reading logic until the
		// peer is stopped
		let iter = stream::iter_ok(iter::repeat(()).map(Ok::<(), Error>));
		// setup the reading future, getting messages from the peer and processing them
		let recv_bytes = self.received_bytes.clone();
		let handler = Arc::new(handler);
		let mut count = 0;
		let read_msg = iter.buffered(1).fold(reader, move |reader, _| {
			count += 1;
			trace!(LOGGER, "read_msg: count (per buffered fold): {}", count);
			let recv_bytes = recv_bytes.clone();
			let handler = handler.clone();
			let sender_inner = sender.clone();
			let pool = pool.clone();
			// first read the message header
			read_exact(reader, vec![0u8; HEADER_LEN as usize])
				.from_err()
				.and_then(move |(reader, buf)| {
					trace!(LOGGER, "read_msg: start");
					let header = try!(ser::deserialize::<MsgHeader>(&mut &buf[..]));
					Ok((reader, header))
				})
				.and_then(move |(reader, header)| {
					// now that we have a size, proceed with the body
					read_exact(reader, vec![0u8; header.msg_len as usize])
						.map(|(reader, buf)| (reader, header, buf))
						.from_err()
				})
				.and_then(move |(reader, header, buf)| {
					// add the count of bytes received
					let mut recv_bytes = recv_bytes.lock().unwrap();
					*recv_bytes += header.serialized_len() + header.msg_len;
					pool.spawn_fn(move || {
						let msg_type = header.msg_type;
						if let Err(e) = handler.handle(sender_inner.clone(), header, buf) {
							debug!(LOGGER, "Invalid {:?} message: {}", msg_type, e);
							return Err(Error::Serialization(e));
						}
						trace!(LOGGER, "read_msg: done (via cpu_pool)");
						Ok(reader)
					})
				})
		});
		Box::new(read_msg)
	}
fn read_msg < F , R > ( & self , sender : UnboundedSender < Vec < u8 > > , reader : R , handler : F , pool : CpuPool , ) -> Box < Future < Item = R , Error = Error > > where F : Handler + ' static , R : AsyncRead + Send + ' static , { // infinite iterator stream so we repeat the message reading logic until the // peer is stopped let iter = stream :: iter_ok ( iter :: repeat ( ( ) ) . map ( Ok :: < ( ) , Error > ) ) ; // setup the reading future, getting messages from the peer and processing them let recv_bytes = self . received_bytes . clone ( ) ; let handler = Arc :: new ( handler ) ; let mut count = 0 ; let read_msg = iter . buffered ( 1 ) . fold ( reader , move | reader , _ | { count += 1 ; trace ! ( LOGGER " " count ) ; let recv_bytes = recv_bytes . clone ( ) ; let handler = handler . clone ( ) ; let sender_inner = sender . clone ( ) ; let pool = pool . clone ( ) ; // first read the message header read_exact ( reader , vec ! [ 0u8 HEADER_LEN as usize ] ) . from_err ( ) . and_then ( move | ( reader , buf ) | { trace ! ( LOGGER " " ) ; let header = try ! ( ser deserialize MsgHeader ( mut buf [ ] ) ) ; Ok ( ( reader , header ) ) } ) . and_then ( move | ( reader , header ) | { // now that we have a size, proceed with the body read_exact ( reader , vec ! [ 0u8 header msg_len as usize ] ) . map ( | ( reader , buf ) | ( reader , header , buf ) ) . from_err ( ) } ) . and_then ( move | ( reader , header , buf ) | { // add the count of bytes received let mut recv_bytes = recv_bytes . lock ( ) . unwrap ( ) ; * recv_bytes += header . serialized_len ( ) + header . msg_len ; pool . spawn_fn ( move | | { let msg_type = header . msg_type ; if let Err ( e ) = handler . handle ( sender_inner . clone ( ) , header , buf ) { debug ! ( LOGGER " " msg_type e ) ; return Err ( Error :: Serialization ( e ) ) ; } trace ! ( LOGGER " " ) ; Ok ( reader ) } ) } ) } ) ; Box :: new ( read_msg ) }
Prepares the future reading from the peer connection, parsing each
message and forwarding them appropriately based on their type
Prepares the future reading from the peer connection parsing each message and forwarding them appropriately based on their type
&self
sender
UnboundedSender<Vec<u8>>
reader
R
handler
F
pool
CpuPool