cdrs_async/
transport_tcp.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_trait::async_trait;
14
15use super::transport::CDRSTransport;
16
17/// CDRS TCP transport.
18pub struct TransportTcp {
19  tcp: net::TcpStream,
20  _addr: String,
21}
22
23impl TransportTcp {
24  /// Constructs a new `TransportTcp`.
25  ///
26  /// # Examples
27  ///
28  /// ```no_run
29  /// use cdrs::transport::TransportTcp;
30  /// let addr = "127.0.0.1:9042";
31  /// let tcp_transport = TransportTcp::new(addr).unwrap();
32  /// ```
33  pub async fn new(addr: &str) -> io::Result<TransportTcp> {
34    net::TcpStream::connect(addr)
35      .await
36      .map(|socket| TransportTcp {
37        tcp: socket,
38        _addr: addr.to_string(),
39      })
40  }
41}
42
43impl Unpin for TransportTcp {}
44
45impl Read for TransportTcp {
46  fn poll_read(
47    mut self: Pin<&mut Self>,
48    cx: &mut Context,
49    buf: &mut [u8],
50  ) -> Poll<io::Result<usize>> {
51    Pin::new(&mut self.tcp).poll_read(cx, buf)
52  }
53
54  fn poll_read_vectored(
55    mut self: Pin<&mut Self>,
56    cx: &mut Context<'_>,
57    bufs: &mut [IoSliceMut<'_>],
58  ) -> Poll<io::Result<usize>> {
59    Pin::new(&mut self.tcp).poll_read_vectored(cx, bufs)
60  }
61}
62
63impl Write for TransportTcp {
64  fn poll_write(
65    mut self: Pin<&mut Self>,
66    cx: &mut Context<'_>,
67    buf: &[u8],
68  ) -> Poll<io::Result<usize>> {
69    Pin::new(&mut self.tcp).poll_write(cx, buf)
70  }
71
72  fn poll_write_vectored(
73    mut self: Pin<&mut Self>,
74    cx: &mut Context<'_>,
75    bufs: &[IoSlice<'_>],
76  ) -> Poll<io::Result<usize>> {
77    Pin::new(&mut self.tcp).poll_write_vectored(cx, bufs)
78  }
79
80  fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
81    Pin::new(&mut self.tcp).poll_flush(cx)
82  }
83
84  fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
85    Pin::new(&mut self.tcp).poll_close(cx)
86  }
87}
88
89#[async_trait]
90impl CDRSTransport for TransportTcp {
91  // FIXME:
92  // async fn try_clone(&self) -> io::Result<TransportTcp> {
93  //   net::TcpStream::connect(self.addr.as_str())
94  //     .await
95  //     .map(|socket| TransportTcp {
96  //       tcp: socket,
97  //       addr: self.addr.clone(),
98  //     })
99  // }
100
101  fn close(&mut self, close: net::Shutdown) -> io::Result<()> {
102    self.tcp.shutdown(close)
103  }
104
105  fn is_alive(&self) -> bool {
106    self.tcp.peer_addr().is_ok()
107  }
108}
cdrs_async/transport_tcp.rs

cdrs_async/
transport_tcp.rs
