cdrs_async/
transport_tls.rs

1use std::{
2  io::{IoSlice, IoSliceMut},
3  marker::Unpin,
4  pin::Pin,
5  task::{Context, Poll},
6};
7
8use async_std::{
9  io,
10  io::{Read, Write},
11  net,
12};
13use async_tls::{client::TlsStream, TlsConnector};
14use async_trait::async_trait;
15
16use super::transport::CDRSTransport;
17
18pub type Stream = TlsStream<net::TcpStream>;
19
20/// CDRS TLS transport.
21pub struct TransportTls {
22  stream: Stream,
23  _addr: String,
24}
25
26impl TransportTls {
27  /// Constructs a new `TransportTcp`.
28  ///
29  /// # Examples
30  ///
31  /// ```no_run
32  /// use cdrs::transport::TransportTls;
33  /// let addr = "127.0.0.1:9042";
34  /// let tcp_transport = TransportTls::new(addr).unwrap();
35  /// ```
36  pub async fn new(addr: &str, connector: TlsConnector) -> io::Result<TransportTls> {
37    let tcp_stream = net::TcpStream::connect(addr).await?;
38    let stream = connector.connect(addr, tcp_stream)?.await?;
39    Ok(TransportTls {
40      stream,
41      _addr: addr.to_string(),
42    })
43  }
44}
45
46impl Unpin for TransportTls {}
47
48impl Read for TransportTls {
49  fn poll_read(
50    mut self: Pin<&mut Self>,
51    cx: &mut Context,
52    buf: &mut [u8],
53  ) -> Poll<io::Result<usize>> {
54    Pin::new(&mut self.stream).poll_read(cx, buf)
55  }
56
57  fn poll_read_vectored(
58    mut self: Pin<&mut Self>,
59    cx: &mut Context<'_>,
60    bufs: &mut [IoSliceMut<'_>],
61  ) -> Poll<io::Result<usize>> {
62    Pin::new(&mut self.stream).poll_read_vectored(cx, bufs)
63  }
64}
65
66impl Write for TransportTls {
67  fn poll_write(
68    mut self: Pin<&mut Self>,
69    cx: &mut Context<'_>,
70    buf: &[u8],
71  ) -> Poll<io::Result<usize>> {
72    Pin::new(&mut self.stream).poll_write(cx, buf)
73  }
74
75  fn poll_write_vectored(
76    mut self: Pin<&mut Self>,
77    cx: &mut Context<'_>,
78    bufs: &[IoSlice<'_>],
79  ) -> Poll<io::Result<usize>> {
80    Pin::new(&mut self.stream).poll_write_vectored(cx, bufs)
81  }
82
83  fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
84    Pin::new(&mut self.stream).poll_flush(cx)
85  }
86
87  fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
88    Pin::new(&mut self.stream).poll_close(cx)
89  }
90}
91
92#[async_trait]
93impl CDRSTransport for TransportTls {
94  // FIXME:
95  // async fn try_clone(&self) -> io::Result<TransportTls> {
96  //   // TODO:
97  //   todo!()
98  // }
99
100  fn close(&mut self, close: net::Shutdown) -> io::Result<()> {
101    self.stream.get_mut().shutdown(close)
102  }
103
104  fn is_alive(&self) -> bool {
105    self.stream.get_ref().peer_addr().is_ok()
106  }
107}
cdrs_async/transport_tls.rs

cdrs_async/
transport_tls.rs
