1use std::io::{IoSlice, IoSliceMut};
2use std::net::SocketAddr;
3use std::pin::Pin;
4
5use async_io::Async;
6
7use crate::io::{self, Read, Write};
8use crate::net::ToSocketAddrs;
9use crate::sync::Arc;
10use crate::task::{Context, Poll};
11
12/// A TCP stream between a local and a remote socket.
13///
14/// A `TcpStream` can either be created by connecting to an endpoint, via the [`connect`] method,
15/// or by [accepting] a connection from a [listener]. It can be read or written to using the
16/// [`AsyncRead`], [`AsyncWrite`], and related extension traits in [`futures::io`].
17///
18/// The connection will be closed when the value is dropped. The reading and writing portions of
19/// the connection can also be shut down individually with the [`shutdown`] method.
20///
21/// This type is an async version of [`std::net::TcpStream`].
22///
23/// [`connect`]: struct.TcpStream.html#method.connect
24/// [accepting]: struct.TcpListener.html#method.accept
25/// [listener]: struct.TcpListener.html
26/// [`AsyncRead`]: https://docs.rs/futures/0.3/futures/io/trait.AsyncRead.html
27/// [`AsyncWrite`]: https://docs.rs/futures/0.3/futures/io/trait.AsyncWrite.html
28/// [`futures::io`]: https://docs.rs/futures/0.3/futures/io/index.html
29/// [`shutdown`]: struct.TcpStream.html#method.shutdown
30/// [`std::net::TcpStream`]: https://doc.rust-lang.org/std/net/struct.TcpStream.html
31///
32/// ## Examples
33///
34/// ```no_run
35/// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
36/// #
37/// use async_std::net::TcpStream;
38/// use async_std::prelude::*;
39///
40/// let mut stream = TcpStream::connect("127.0.0.1:8080").await?;
41/// stream.write_all(b"hello world").await?;
42///
43/// let mut buf = vec![0u8; 1024];
44/// let n = stream.read(&mut buf).await?;
45/// #
46/// # Ok(()) }) }
47/// ```
48#[derive(Debug, Clone)]
49pub struct TcpStream {
50 pub(super) watcher: Arc<Async<std::net::TcpStream>>,
51}
52
53impl TcpStream {
54 /// Creates a new TCP stream connected to the specified address.
55 ///
56 /// This method will create a new TCP socket and attempt to connect it to the `addr`
57 /// provided. The [returned future] will be resolved once the stream has successfully
58 /// connected, or it will return an error if one occurs.
59 ///
60 /// [returned future]: struct.Connect.html
61 ///
62 /// # Examples
63 ///
64 /// ```no_run
65 /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
66 /// #
67 /// use async_std::net::TcpStream;
68 ///
69 /// let stream = TcpStream::connect("127.0.0.1:0").await?;
70 /// #
71 /// # Ok(()) }) }
72 /// ```
73 pub async fn connect<A: ToSocketAddrs>(addrs: A) -> io::Result<TcpStream> {
74 let mut last_err = None;
75 let addrs = addrs.to_socket_addrs().await?;
76
77 for addr in addrs {
78 match Async::<std::net::TcpStream>::connect(addr).await {
79 Ok(stream) => {
80 return Ok(TcpStream {
81 watcher: Arc::new(stream),
82 });
83 }
84 Err(e) => {
85 last_err = Some(e);
86 continue;
87 }
88 }
89 }
90
91 Err(last_err.unwrap_or_else(|| {
92 io::Error::new(
93 io::ErrorKind::InvalidInput,
94 "could not resolve to any addresses",
95 )
96 }))
97 }
98
99 /// Returns the local address that this stream is connected to.
100 ///
101 /// ## Examples
102 ///
103 /// ```no_run
104 /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
105 /// #
106 /// use async_std::net::TcpStream;
107 ///
108 /// let stream = TcpStream::connect("127.0.0.1:8080").await?;
109 /// let addr = stream.local_addr()?;
110 /// #
111 /// # Ok(()) }) }
112 /// ```
113 pub fn local_addr(&self) -> io::Result<SocketAddr> {
114 self.watcher.get_ref().local_addr()
115 }
116
117 /// Returns the remote address that this stream is connected to.
118 ///
119 /// ## Examples
120 ///
121 /// ```no_run
122 /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
123 /// #
124 /// use async_std::net::TcpStream;
125 ///
126 /// let stream = TcpStream::connect("127.0.0.1:8080").await?;
127 /// let peer = stream.peer_addr()?;
128 /// #
129 /// # Ok(()) }) }
130 /// ```
131 pub fn peer_addr(&self) -> io::Result<SocketAddr> {
132 self.watcher.get_ref().peer_addr()
133 }
134
135 /// Gets the value of the `IP_TTL` option for this socket.
136 ///
137 /// For more information about this option, see [`set_ttl`].
138 ///
139 /// [`set_ttl`]: #method.set_ttl
140 ///
141 /// # Examples
142 ///
143 /// ```no_run
144 /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
145 /// #
146 /// use async_std::net::TcpStream;
147 ///
148 /// let stream = TcpStream::connect("127.0.0.1:8080").await?;
149 ///
150 /// stream.set_ttl(100)?;
151 /// assert_eq!(stream.ttl()?, 100);
152 /// #
153 /// # Ok(()) }) }
154 /// ```
155 pub fn ttl(&self) -> io::Result<u32> {
156 self.watcher.get_ref().ttl()
157 }
158
159 /// Sets the value for the `IP_TTL` option on this socket.
160 ///
161 /// This value sets the time-to-live field that is used in every packet sent
162 /// from this socket.
163 ///
164 /// # Examples
165 ///
166 /// ```no_run
167 /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
168 /// #
169 /// use async_std::net::TcpStream;
170 ///
171 /// let stream = TcpStream::connect("127.0.0.1:8080").await?;
172 ///
173 /// stream.set_ttl(100)?;
174 /// assert_eq!(stream.ttl()?, 100);
175 /// #
176 /// # Ok(()) }) }
177 /// ```
178 pub fn set_ttl(&self, ttl: u32) -> io::Result<()> {
179 self.watcher.get_ref().set_ttl(ttl)
180 }
181
182 /// Receives data on the socket from the remote address to which it is connected, without
183 /// removing that data from the queue.
184 ///
185 /// On success, returns the number of bytes peeked.
186 ///
187 /// Successive calls return the same data. This is accomplished by passing `MSG_PEEK` as a flag
188 /// to the underlying `recv` system call.
189 ///
190 /// # Examples
191 ///
192 /// ```no_run
193 /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
194 /// #
195 /// use async_std::net::TcpStream;
196 ///
197 /// let stream = TcpStream::connect("127.0.0.1:8000").await?;
198 ///
199 /// let mut buf = vec![0; 1024];
200 /// let n = stream.peek(&mut buf).await?;
201 /// #
202 /// # Ok(()) }) }
203 /// ```
204 pub async fn peek(&self, buf: &mut [u8]) -> io::Result<usize> {
205 self.watcher.peek(buf).await
206 }
207
208 /// Gets the value of the `TCP_NODELAY` option on this socket.
209 ///
210 /// For more information about this option, see [`set_nodelay`].
211 ///
212 /// [`set_nodelay`]: #method.set_nodelay
213 ///
214 /// # Examples
215 ///
216 /// ```no_run
217 /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
218 /// #
219 /// use async_std::net::TcpStream;
220 ///
221 /// let stream = TcpStream::connect("127.0.0.1:8080").await?;
222 ///
223 /// stream.set_nodelay(true)?;
224 /// assert_eq!(stream.nodelay()?, true);
225 /// #
226 /// # Ok(()) }) }
227 /// ```
228 pub fn nodelay(&self) -> io::Result<bool> {
229 self.watcher.get_ref().nodelay()
230 }
231
232 /// Sets the value of the `TCP_NODELAY` option on this socket.
233 ///
234 /// If set, this option disables the Nagle algorithm. This means that
235 /// segments are always sent as soon as possible, even if there is only a
236 /// small amount of data. When not set, data is buffered until there is a
237 /// sufficient amount to send out, thereby avoiding the frequent sending of
238 /// small packets.
239 ///
240 /// # Examples
241 ///
242 /// ```no_run
243 /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
244 /// #
245 /// use async_std::net::TcpStream;
246 ///
247 /// let stream = TcpStream::connect("127.0.0.1:8080").await?;
248 ///
249 /// stream.set_nodelay(true)?;
250 /// assert_eq!(stream.nodelay()?, true);
251 /// #
252 /// # Ok(()) }) }
253 /// ```
254 pub fn set_nodelay(&self, nodelay: bool) -> io::Result<()> {
255 self.watcher.get_ref().set_nodelay(nodelay)
256 }
257
258 /// Shuts down the read, write, or both halves of this connection.
259 ///
260 /// This method will cause all pending and future I/O on the specified portions to return
261 /// immediately with an appropriate value (see the documentation of [`Shutdown`]).
262 ///
263 /// [`Shutdown`]: https://doc.rust-lang.org/std/net/enum.Shutdown.html
264 ///
265 /// # Examples
266 ///
267 /// ```no_run
268 /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
269 /// #
270 /// use std::net::Shutdown;
271 ///
272 /// use async_std::net::TcpStream;
273 ///
274 /// let stream = TcpStream::connect("127.0.0.1:8080").await?;
275 /// stream.shutdown(Shutdown::Both)?;
276 /// #
277 /// # Ok(()) }) }
278 /// ```
279 pub fn shutdown(&self, how: std::net::Shutdown) -> std::io::Result<()> {
280 self.watcher.get_ref().shutdown(how)
281 }
282}
283
284impl Read for TcpStream {
285 fn poll_read(
286 self: Pin<&mut Self>,
287 cx: &mut Context<'_>,
288 buf: &mut [u8],
289 ) -> Poll<io::Result<usize>> {
290 Pin::new(&mut &*self).poll_read(cx, buf)
291 }
292
293 fn poll_read_vectored(
294 self: Pin<&mut Self>,
295 cx: &mut Context<'_>,
296 bufs: &mut [IoSliceMut<'_>],
297 ) -> Poll<io::Result<usize>> {
298 Pin::new(&mut &*self).poll_read_vectored(cx, bufs)
299 }
300}
301
302impl Read for &TcpStream {
303 fn poll_read(
304 self: Pin<&mut Self>,
305 cx: &mut Context<'_>,
306 buf: &mut [u8],
307 ) -> Poll<io::Result<usize>> {
308 Pin::new(&mut &*self.watcher).poll_read(cx, buf)
309 }
310
311 fn poll_read_vectored(
312 self: Pin<&mut Self>,
313 cx: &mut Context<'_>,
314 bufs: &mut [IoSliceMut<'_>],
315 ) -> Poll<io::Result<usize>> {
316 Pin::new(&mut &*self.watcher).poll_read_vectored(cx, bufs)
317 }
318}
319
320impl Write for TcpStream {
321 fn poll_write(
322 self: Pin<&mut Self>,
323 cx: &mut Context<'_>,
324 buf: &[u8],
325 ) -> Poll<io::Result<usize>> {
326 Pin::new(&mut &*self).poll_write(cx, buf)
327 }
328
329 fn poll_write_vectored(
330 self: Pin<&mut Self>,
331 cx: &mut Context<'_>,
332 bufs: &[IoSlice<'_>],
333 ) -> Poll<io::Result<usize>> {
334 Pin::new(&mut &*self).poll_write_vectored(cx, bufs)
335 }
336
337 fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
338 Pin::new(&mut &*self).poll_flush(cx)
339 }
340
341 fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
342 Pin::new(&mut &*self).poll_close(cx)
343 }
344}
345
346impl Write for &TcpStream {
347 fn poll_write(
348 self: Pin<&mut Self>,
349 cx: &mut Context<'_>,
350 buf: &[u8],
351 ) -> Poll<io::Result<usize>> {
352 Pin::new(&mut &*self.watcher).poll_write(cx, buf)
353 }
354
355 fn poll_write_vectored(
356 self: Pin<&mut Self>,
357 cx: &mut Context<'_>,
358 bufs: &[IoSlice<'_>],
359 ) -> Poll<io::Result<usize>> {
360 Pin::new(&mut &*self.watcher).poll_write_vectored(cx, bufs)
361 }
362
363 fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
364 Pin::new(&mut &*self.watcher).poll_flush(cx)
365 }
366
367 fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
368 Pin::new(&mut &*self.watcher).poll_close(cx)
369 }
370}
371
372impl From<std::net::TcpStream> for TcpStream {
373 /// Converts a `std::net::TcpStream` into its asynchronous equivalent.
374 fn from(stream: std::net::TcpStream) -> TcpStream {
375 TcpStream {
376 watcher: Arc::new(data:Async::new(stream).expect(msg:"TcpStream is known to be good")),
377 }
378 }
379}
380
381impl std::convert::TryFrom<TcpStream> for std::net::TcpStream {
382 type Error = io::Error;
383 /// Converts a `TcpStream` into its synchronous equivalent.
384 fn try_from(stream: TcpStream) -> io::Result<std::net::TcpStream> {
385 let inner: TcpStream = ArcAsync::try_unwrap(stream.watcher)
386 .map_err(|_| io::Error::new(
387 kind:io::ErrorKind::Other,
388 error:"Cannot convert TcpStream to synchronous: multiple references",
389 ))?
390 .into_inner()?;
391 inner.set_nonblocking(false)?;
392 Ok(inner)
393 }
394}
395
396cfg_unix! {
397 use crate::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
398
399 impl AsRawFd for TcpStream {
400 fn as_raw_fd(&self) -> RawFd {
401 self.watcher.get_ref().as_raw_fd()
402 }
403 }
404
405 impl FromRawFd for TcpStream {
406 unsafe fn from_raw_fd(fd: RawFd) -> TcpStream {
407 std::net::TcpStream::from_raw_fd(fd).into()
408 }
409 }
410
411 impl IntoRawFd for TcpStream {
412 fn into_raw_fd(self) -> RawFd {
413 // TODO(stjepang): This does not mean `RawFd` is now the sole owner of the file
414 // descriptor because it's possible that there are other clones of this `TcpStream`
415 // using it at the same time. We should probably document that behavior.
416 self.as_raw_fd()
417 }
418 }
419
420 cfg_io_safety! {
421 use crate::os::unix::io::{AsFd, BorrowedFd, OwnedFd};
422
423 impl AsFd for TcpStream {
424 fn as_fd(&self) -> BorrowedFd<'_> {
425 self.watcher.get_ref().as_fd()
426 }
427 }
428
429 impl From<OwnedFd> for TcpStream {
430 fn from(fd: OwnedFd) -> TcpStream {
431 std::net::TcpStream::from(fd).into()
432 }
433 }
434
435 impl From<TcpStream> for OwnedFd {
436 fn from(stream: TcpStream) -> OwnedFd {
437 stream.watcher.get_ref().try_clone().unwrap().into()
438 }
439 }
440 }
441}
442
443cfg_windows! {
444 use crate::os::windows::io::{
445 RawSocket, AsRawSocket, FromRawSocket, IntoRawSocket
446 };
447
448 impl AsRawSocket for TcpStream {
449 fn as_raw_socket(&self) -> RawSocket {
450 self.watcher.get_ref().as_raw_socket()
451 }
452 }
453
454 impl FromRawSocket for TcpStream {
455 unsafe fn from_raw_socket(handle: RawSocket) -> TcpStream {
456 std::net::TcpStream::from_raw_socket(handle).into()
457 }
458 }
459
460 impl IntoRawSocket for TcpStream {
461 fn into_raw_socket(self) -> RawSocket {
462 // TODO(stjepang): This does not mean `RawFd` is now the sole owner of the file
463 // descriptor because it's possible that there are other clones of this `TcpStream`
464 // using it at the same time. We should probably document that behavior.
465 self.as_raw_socket()
466 }
467 }
468
469 cfg_io_safety! {
470 use crate::os::windows::io::{AsSocket, BorrowedSocket, OwnedSocket};
471
472 impl AsSocket for TcpStream {
473 fn as_socket(&self) -> BorrowedSocket<'_> {
474 self.watcher.get_ref().as_socket()
475 }
476 }
477
478 impl From<OwnedSocket> for TcpStream {
479 fn from(fd: OwnedSocket) -> TcpStream {
480 std::net::TcpStream::from(fd).into()
481 }
482 }
483
484 impl From<TcpStream> for OwnedSocket {
485 fn from(stream: TcpStream) -> OwnedSocket {
486 stream.watcher.get_ref().try_clone().unwrap().into()
487 }
488 }
489 }
490}
491