use std::hash::Hash; use std::io::Write; use std::marker::PhantomData; use std::net::{SocketAddr, TcpListener as StdTcpListener, TcpStream}; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use crate::codec::Codec; use crate::context::Context; use crate::transport::connection::Connection; use crate::transport::error::TransportError; use crate::transport::listener::Listener; use crate::transport::receiver::PacketReceiver; use crate::transport::sender::PacketSender; use crate::transport::{Result, Transport}; // TODO : check Tcp Transport #[derive(Clone)] pub struct TcpSender { stream: Arc>, } impl PacketSender for TcpSender { fn send>(&self, packet: P, ctx: &Context) -> Result<()> { let mut buf = Vec::new(); packet.encode(&mut buf, ctx)?; let mut guard = self .stream .lock() .map_err(|_| TransportError::LockPoisoned)?; guard.write_all(&buf)?; guard.flush()?; Ok(()) } } pub struct TcpReceiver { stream: Arc>, uid: Uid, } impl PacketReceiver for TcpReceiver where Uid: PartialEq + Clone + Send + Sync, { fn recv>(&self, ctx: &Context) -> Result<(Uid, P)> { let mut guard = self .stream .lock() .map_err(|_| TransportError::LockPoisoned)?; let packet = P::decode(&mut *guard, ctx)?; Ok((self.uid.clone(), packet)) } } pub struct TcpListener { listener: Arc, factory: Arc Uid + Send + Sync>, _p: PhantomData, } impl Listener for TcpListener where Uid: PartialEq + Eq + Hash + Clone + Send + Sync, Data: Send + Sync, { type Data = Data; type Sender = TcpSender; type Receiver = TcpReceiver; fn accept(&mut self) -> Result>> { let (stream, _peer) = self.listener.accept()?; let stream = Arc::new(Mutex::new(stream)); let uid = (self.factory)(); let sender = TcpSender { stream: stream.clone(), }; let receiver = TcpReceiver { stream, uid }; Ok(Connection::new(sender, receiver)) } } pub struct TcpTransport { factory: Arc Uid + Send + Sync>, _p: PhantomData<(Uid, Data)>, } impl TcpTransport where Uid: PartialEq + Eq + Hash + Clone + Send + Sync, { pub fn new Uid + Send + Sync + 'static>(factory: F) -> Self { Self { factory: Arc::new(factory), _p: PhantomData, } } } impl Default for TcpTransport { fn default() -> Self { let counter = Arc::new(AtomicU64::new(0)); TcpTransport::new(move || counter.fetch_add(1, Ordering::Relaxed)) } } impl Transport for TcpTransport where Uid: PartialEq + Eq + Hash + Clone + Send + Sync, Data: Send + Sync, { type Data = Data; type Addr = SocketAddr; type Sender = TcpSender; type Receiver = TcpReceiver; type Listener = TcpListener; fn connect( &self, addr: SocketAddr, ) -> Result>> { let stream = TcpStream::connect(addr)?; let stream = Arc::new(Mutex::new(stream)); let uid = (self.factory)(); let sender = TcpSender { stream: stream.clone(), }; let receiver = TcpReceiver { stream, uid }; Ok(Connection::new(sender, receiver)) } fn listen(&self, addr: SocketAddr) -> Result> { let listener = StdTcpListener::bind(addr)?; Ok(TcpListener { listener: Arc::new(listener), factory: self.factory.clone(), _p: PhantomData, }) } }