ferroui_remote_protocol/
tcp_transport_base.rs

1use std::io::{Read, Write};
2use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, Shutdown, SocketAddr, TcpListener, TcpStream};
3use std::sync::atomic::{AtomicBool, Ordering};
4use std::sync::Arc;
5use std::time::Duration;
6
7use crate::error::Error;
8use crate::i_message_type_resolver::IMessageTypeResolver;
9use crate::i_transport::IFerroRemoteTransportConnection;
10
11/// What a connection calls when it is disposed.
12pub type DisposeCallback = Box<dyn Fn() + Send + Sync>;
13
14/// `DisposableServer`: what [`TcpTransportBase::listen`] returns. Disposing
15/// it stops the listening; connections that were accepted stay open. As in
16/// the original, a server that is never disposed keeps listening.
17pub struct DisposableServer {
18    address: SocketAddr,
19    stopped: Arc<AtomicBool>,
20}
21
22impl DisposableServer {
23    /// The address the server listens on, with the port the system assigned
24    /// when the port asked for was zero. (An addition of the port: the
25    /// original hands out only the `IDisposable`.)
26    pub fn local_addr(&self) -> SocketAddr {
27        self.address
28    }
29
30    pub fn dispose(&self) {
31        if self.stopped.swap(true, Ordering::SeqCst) {
32            return;
33        }
34        // `_l.Stop()`: a blocking accept of the standard library cannot be
35        // stopped from another thread, so the listener thread is woken with a
36        // connection and ends when it sees the flag.
37        let mut address = self.address;
38        if address.ip().is_unspecified() {
39            let loopback = match address {
40                SocketAddr::V4(_) => IpAddr::V4(Ipv4Addr::LOCALHOST),
41                SocketAddr::V6(_) => IpAddr::V6(Ipv6Addr::LOCALHOST),
42            };
43            address.set_ip(loopback);
44        }
45        //Ignore
46        let _ = TcpStream::connect_timeout(&address, Duration::from_secs(1));
47    }
48}
49
50/// The two directions of a socket as the streams of a connection, and the
51/// callback that closes the socket.
52fn create_transport_of_stream<T: TcpTransportBase>(
53    transport: &T,
54    stream: TcpStream,
55) -> Result<Arc<dyn IFerroRemoteTransportConnection>, Error> {
56    let input = stream.try_clone()?;
57    let output = stream.try_clone()?;
58    let dispose_callback: DisposeCallback = Box::new(move || {
59        let _ = stream.shutdown(Shutdown::Both);
60    });
61    Ok(transport.create_transport(transport.resolver(), Box::new(input), Box::new(output), dispose_callback))
62}
63
64/// A transport over TCP sockets. The abstract class of the original is a
65/// trait: an implementor holds the resolver its constructor is given
66/// ([`TcpTransportBase::resolver`]) and makes the connection of a socket
67/// ([`TcpTransportBase::create_transport`]). It is `Clone` because the
68/// listener thread keeps one.
69pub trait TcpTransportBase: Clone + Send + 'static {
70    /// `_resolver`: the resolver the transport was constructed with.
71    fn resolver(&self) -> Arc<dyn IMessageTypeResolver>;
72
73    /// Makes the connection over the stream of a socket. The original passes
74    /// the one `Stream` of the socket, which is read and written; here the
75    /// two directions are two handles of the socket.
76    fn create_transport(
77        &self,
78        resolver: Arc<dyn IMessageTypeResolver>,
79        input_stream: Box<dyn Read + Send>,
80        output_stream: Box<dyn Write + Send>,
81        dispose_callback: DisposeCallback,
82    ) -> Arc<dyn IFerroRemoteTransportConnection>;
83
84    /// Listens on the address and calls `cb` with the connection of every
85    /// client. A port of zero lets the system choose
86    /// ([`DisposableServer::local_addr`]). An address that cannot be bound is
87    /// an error, as `TcpListener.Start` throws.
88    ///
89    /// Deviation (DEVIATIONS.md, Remote protocol): the original accepts with
90    /// `AcceptTcpClientAsync` and calls back on the thread pool. Here a
91    /// listener thread accepts, and each accepted connection gets a thread
92    /// that makes its transport and calls `cb`, so that a callback that takes
93    /// its time does not hold up the next client. The dispose callback of an
94    /// accepted connection closes its socket (the original only completes a
95    /// task), since that is what ends the blocking read of its reader thread.
96    fn listen(
97        &self,
98        address: IpAddr,
99        port: u16,
100        cb: impl Fn(Arc<dyn IFerroRemoteTransportConnection>) + Send + Sync + 'static,
101    ) -> Result<DisposableServer, Error> {
102        let server = TcpListener::bind(SocketAddr::new(address, port))?;
103        let local_address = server.local_addr()?;
104        let stopped = Arc::new(AtomicBool::new(false));
105        let cb = Arc::new(cb);
106        let this = self.clone();
107        let accept_stopped = stopped.clone();
108        std::thread::Builder::new().name("remote-protocol-listener".to_string()).spawn(move || loop {
109            let cl = match server.accept() {
110                Ok((cl, _)) => cl,
111                //Ignore and stop
112                Err(_) => break,
113            };
114            if accept_stopped.load(Ordering::SeqCst) {
115                break;
116            }
117            let this = this.clone();
118            let cb = cb.clone();
119            let _ = std::thread::Builder::new().name("remote-protocol-accept".to_string()).spawn(move || {
120                //Ignore
121                if let Ok(t) = create_transport_of_stream(&this, cl) {
122                    cb(t);
123                }
124            });
125        })?;
126        Ok(DisposableServer { address: local_address, stopped })
127    }
128
129    /// Connects to a listening transport.
130    ///
131    /// Deviation (DEVIATIONS.md, Remote protocol): the original returns a
132    /// task; this blocks until the connection is made or has failed.
133    fn connect(&self, address: IpAddr, port: u16) -> Result<Arc<dyn IFerroRemoteTransportConnection>, Error> {
134        let c = TcpStream::connect(SocketAddr::new(address, port))?;
135        create_transport_of_stream(self, c)
136    }
137}
ferroui_remote_protocol/tcp_transport_base.rs

ferroui_remote_protocol/
tcp_transport_base.rs
