net.rs (3503B)
1 use std::{ 2 io::{self, Read, Write}, 3 sync::{ 4 Arc, Mutex, 5 mpsc::{Receiver, Sender}, 6 }, 7 }; 8 9 use mio::{Events, Poll, net::TcpStream}; 10 use serde::{Serialize, de::DeserializeOwned}; 11 12 use crate::codec::Codec; 13 14 const MAX_NUM_ATTEMPTS: u8 = 3; 15 16 pub fn handle_network_read<T: DeserializeOwned>( 17 mut poll: Poll, 18 stream: Arc<Mutex<TcpStream>>, 19 tx: Sender<T>, 20 ) { 21 let mut events = Events::with_capacity(128); 22 let mut codec = Codec::default(); 23 let mut buffer = [0x00; 1024]; 24 let mut buffer_index; 25 loop { 26 // need an event loop here because otherwise the calling thread would hold a permanent lock 27 // on the stream while blocking on reading it 28 poll.poll(&mut events, None).unwrap(); 29 for event in events.iter() { 30 if event.is_readable() { 31 let mut stream = stream.lock().unwrap(); 32 let mut all_received = false; 33 let mut attempts = 0; 34 while !all_received && attempts < MAX_NUM_ATTEMPTS { 35 match stream.read(&mut buffer) { 36 Ok(0) => { 37 return; 38 } 39 Ok(i) => { 40 buffer_index = i; 41 let messages = codec.decode(&buffer[..buffer_index]); 42 for message in messages { 43 if let Ok(m) = postcard::from_bytes(message.as_slice()) 44 && tx.send(m).is_err() 45 { 46 break; 47 } 48 } 49 } 50 Err(e) => match e.kind() { 51 io::ErrorKind::WouldBlock => { 52 all_received = true; 53 } 54 io::ErrorKind::NotConnected 55 | io::ErrorKind::ResourceBusy 56 | io::ErrorKind::Interrupted => { 57 attempts += 1; 58 } 59 _ => { 60 return; 61 } 62 }, 63 } 64 } 65 } 66 } 67 } 68 } 69 70 pub fn handle_network_write<T: Serialize>(stream: Arc<Mutex<TcpStream>>, rx: Receiver<T>) { 71 loop { 72 match rx.recv() { 73 Ok(m) => { 74 if let Ok(to_write) = postcard::to_stdvec(&m) 75 && let Ok(mut s) = stream.lock() 76 { 77 let mut all_sent = false; 78 let mut attempts = 0; 79 let bytes = Codec::encode(&to_write); 80 while !all_sent && attempts < MAX_NUM_ATTEMPTS { 81 match s.write_all(bytes.as_slice()) { 82 Ok(_) => all_sent = true, 83 Err(e) => match e.kind() { 84 io::ErrorKind::WouldBlock 85 | io::ErrorKind::NotConnected 86 | io::ErrorKind::ResourceBusy 87 | io::ErrorKind::Interrupted => attempts += 1, 88 _ => return, 89 }, 90 } 91 } 92 } 93 } 94 Err(_) => return, 95 } 96 } 97 }