src/tls/client.rs

   1 use async_channel::{Receiver, Sender};
   2 use async_std::io::*;
   3 use async_std::net::*;
   4 use async_std::task;
   5 use async_tls::TlsConnector;
   6 use futures_util::io::AsyncReadExt;
   7 use log::*;
   8
   9 use crate::registry::StitchRegistry;
  10 use crate::StitchNetClient;
  11 use crate::{channel_factory, StitchMessage};
  12 use async_std::sync::{Arc, Condvar, Mutex};
  13
  14 impl StitchNetClient {
  15     pub fn tls_client<A: ToSocketAddrs + std::fmt::Display>(
  16         ip_addrs: A,
  17         domain: &str,
  18         connector: TlsConnector,
  19     ) -> Result<Self> {
  20         Self::tls_client_with_bound(ip_addrs, domain, connector, None)
  21     }
  22
  23     pub fn tls_client_with_bound<A: ToSocketAddrs + std::fmt::Display>(
  24         ip_addrs: A,
  25         domain: &str,
  26         connector: TlsConnector,
  27         cap: Option<usize>,
  28     ) -> Result<Self> {
  29         let stream = task::block_on(TcpStream::connect(&ip_addrs))?;
  30         stream.set_nodelay(true)?;
  31         info!("Established client TCP connection to {}", ip_addrs);
  32
  33         Self::tls_client_from_parts(stream, domain, connector, channel_factory(cap))
  34     }
  35
  36     pub fn tls_client_from_parts(
  37         stream: TcpStream,
  38         domain: &str,
  39         connector: TlsConnector,
  40         (tls_write_sender, tls_write_receiver): (Sender<StitchMessage>, Receiver<StitchMessage>),
  41     ) -> Result<Self> {
  42         let local_addr = stream.local_addr()?;
  43         let peer_addr = stream.peer_addr()?;
  44
  45         let encrypted_stream = task::block_on(connector.connect(domain, stream))?;
  46         let (read_stream, write_stream) = encrypted_stream.split();
  47         info!("Completed TLS handshake with {}", peer_addr);
  48
  49         let registry: StitchRegistry = crate::registry::new();
  50         let read_readiness = Arc::new((Mutex::new(false), Condvar::new()));
  51         let write_readiness = Arc::new((Mutex::new(false), Condvar::new()));
  52
  53         let read_task = task::spawn(crate::tasks::read_from_stream(
  54             registry.clone(),
  55             read_stream,
  56             read_readiness.clone(),
  57         ));
  58
  59         let write_task = task::spawn(crate::tasks::write_to_stream(
  60             tls_write_receiver.clone(),
  61             write_stream,
  62             write_readiness.clone(),
  63         ));
  64
  65         Ok(Self {
  66             local_addr,
  67             peer_addr,
  68             registry,
  69             stream_writer_chan: (tls_write_sender, tls_write_receiver),
  70             read_readiness,
  71             write_readiness,
  72             read_task,
  73             write_task,
  74         })
  75     }
  76 }