1 //! Asynchronous I/O.
2 //!
3 //! This module is the asynchronous version of `std::io`. It defines four
4 //! traits, [`AsyncRead`], [`AsyncWrite`], [`AsyncSeek`], and [`AsyncBufRead`],
5 //! which mirror the `Read`, `Write`, `Seek`, and `BufRead` traits of the
6 //! standard library. However, these traits integrate with the asynchronous
7 //! task system, so that if an I/O object isn't ready for reading (or writing),
8 //! the thread is not blocked, and instead the current task is queued to be
9 //! woken when I/O is ready.
10 //!
11 //! In addition, the [`AsyncReadExt`], [`AsyncWriteExt`], [`AsyncSeekExt`], and
12 //! [`AsyncBufReadExt`] extension traits offer a variety of useful combinators
13 //! for operating with asynchronous I/O objects, including ways to work with
14 //! them using futures, streams and sinks.
15 //!
16 //! This module is only available when the `std` feature of this
17 //! library is activated, and it is activated by default.
18
19 #[cfg(feature = "io-compat")]
20 #[cfg_attr(docsrs, doc(cfg(feature = "io-compat")))]
21 use crate::compat::Compat;
22 use crate::future::assert_future;
23 use crate::stream::assert_stream;
24 use std::{pin::Pin, ptr};
25
26 // Re-export some types from `std::io` so that users don't have to deal
27 // with conflicts when `use`ing `futures::io` and `std::io`.
28 #[doc(no_inline)]
29 #[cfg(feature = "read-initializer")]
30 #[cfg_attr(docsrs, doc(cfg(feature = "read-initializer")))]
31 pub use std::io::Initializer;
32 #[doc(no_inline)]
33 pub use std::io::{Error, ErrorKind, IoSlice, IoSliceMut, Result, SeekFrom};
34
35 pub use futures_io::{AsyncBufRead, AsyncRead, AsyncSeek, AsyncWrite};
36
37 // used by `BufReader` and `BufWriter`
38 // https://github.com/rust-lang/rust/blob/master/src/libstd/sys_common/io.rs#L1
39 const DEFAULT_BUF_SIZE: usize = 8 * 1024;
40
41 /// Initializes a buffer if necessary.
42 ///
43 /// A buffer is always initialized if `read-initializer` feature is disabled.
44 #[inline]
initialize<R: AsyncRead>(_reader: &R, buf: &mut [u8])45 unsafe fn initialize<R: AsyncRead>(_reader: &R, buf: &mut [u8]) {
46 #[cfg(feature = "read-initializer")]
47 {
48 if !_reader.initializer().should_initialize() {
49 return;
50 }
51 }
52 ptr::write_bytes(buf.as_mut_ptr(), 0, buf.len())
53 }
54
55 mod allow_std;
56 pub use self::allow_std::AllowStdIo;
57
58 mod buf_reader;
59 pub use self::buf_reader::{BufReader, SeeKRelative};
60
61 mod buf_writer;
62 pub use self::buf_writer::BufWriter;
63
64 mod chain;
65 pub use self::chain::Chain;
66
67 mod close;
68 pub use self::close::Close;
69
70 mod copy;
71 pub use self::copy::{copy, Copy};
72
73 mod copy_buf;
74 pub use self::copy_buf::{copy_buf, CopyBuf};
75
76 mod cursor;
77 pub use self::cursor::Cursor;
78
79 mod empty;
80 pub use self::empty::{empty, Empty};
81
82 mod fill_buf;
83 pub use self::fill_buf::FillBuf;
84
85 mod flush;
86 pub use self::flush::Flush;
87
88 #[cfg(feature = "sink")]
89 #[cfg_attr(docsrs, doc(cfg(feature = "sink")))]
90 mod into_sink;
91 #[cfg(feature = "sink")]
92 #[cfg_attr(docsrs, doc(cfg(feature = "sink")))]
93 pub use self::into_sink::IntoSink;
94
95 mod lines;
96 pub use self::lines::Lines;
97
98 mod read;
99 pub use self::read::Read;
100
101 mod read_vectored;
102 pub use self::read_vectored::ReadVectored;
103
104 mod read_exact;
105 pub use self::read_exact::ReadExact;
106
107 mod read_line;
108 pub use self::read_line::ReadLine;
109
110 mod read_to_end;
111 pub use self::read_to_end::ReadToEnd;
112
113 mod read_to_string;
114 pub use self::read_to_string::ReadToString;
115
116 mod read_until;
117 pub use self::read_until::ReadUntil;
118
119 mod repeat;
120 pub use self::repeat::{repeat, Repeat};
121
122 mod seek;
123 pub use self::seek::Seek;
124
125 mod sink;
126 pub use self::sink::{sink, Sink};
127
128 mod split;
129 pub use self::split::{ReadHalf, ReuniteError, WriteHalf};
130
131 mod take;
132 pub use self::take::Take;
133
134 mod window;
135 pub use self::window::Window;
136
137 mod write;
138 pub use self::write::Write;
139
140 mod write_vectored;
141 pub use self::write_vectored::WriteVectored;
142
143 mod write_all;
144 pub use self::write_all::WriteAll;
145
146 #[cfg(feature = "write-all-vectored")]
147 mod write_all_vectored;
148 #[cfg(feature = "write-all-vectored")]
149 pub use self::write_all_vectored::WriteAllVectored;
150
151 /// An extension trait which adds utility methods to `AsyncRead` types.
152 pub trait AsyncReadExt: AsyncRead {
153 /// Creates an adaptor which will chain this stream with another.
154 ///
155 /// The returned `AsyncRead` instance will first read all bytes from this object
156 /// until EOF is encountered. Afterwards the output is equivalent to the
157 /// output of `next`.
158 ///
159 /// # Examples
160 ///
161 /// ```
162 /// # futures::executor::block_on(async {
163 /// use futures::io::{AsyncReadExt, Cursor};
164 ///
165 /// let reader1 = Cursor::new([1, 2, 3, 4]);
166 /// let reader2 = Cursor::new([5, 6, 7, 8]);
167 ///
168 /// let mut reader = reader1.chain(reader2);
169 /// let mut buffer = Vec::new();
170 ///
171 /// // read the value into a Vec.
172 /// reader.read_to_end(&mut buffer).await?;
173 /// assert_eq!(buffer, [1, 2, 3, 4, 5, 6, 7, 8]);
174 /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
175 /// ```
chain<R>(self, next: R) -> Chain<Self, R> where Self: Sized, R: AsyncRead,176 fn chain<R>(self, next: R) -> Chain<Self, R>
177 where
178 Self: Sized,
179 R: AsyncRead,
180 {
181 assert_read(Chain::new(self, next))
182 }
183
184 /// Tries to read some bytes directly into the given `buf` in asynchronous
185 /// manner, returning a future type.
186 ///
187 /// The returned future will resolve to the number of bytes read once the read
188 /// operation is completed.
189 ///
190 /// # Examples
191 ///
192 /// ```
193 /// # futures::executor::block_on(async {
194 /// use futures::io::{AsyncReadExt, Cursor};
195 ///
196 /// let mut reader = Cursor::new([1, 2, 3, 4]);
197 /// let mut output = [0u8; 5];
198 ///
199 /// let bytes = reader.read(&mut output[..]).await?;
200 ///
201 /// // This is only guaranteed to be 4 because `&[u8]` is a synchronous
202 /// // reader. In a real system you could get anywhere from 1 to
203 /// // `output.len()` bytes in a single read.
204 /// assert_eq!(bytes, 4);
205 /// assert_eq!(output, [1, 2, 3, 4, 0]);
206 /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
207 /// ```
read<'a>(&'a mut self, buf: &'a mut [u8]) -> Read<'a, Self> where Self: Unpin,208 fn read<'a>(&'a mut self, buf: &'a mut [u8]) -> Read<'a, Self>
209 where
210 Self: Unpin,
211 {
212 assert_future::<Result<usize>, _>(Read::new(self, buf))
213 }
214
215 /// Creates a future which will read from the `AsyncRead` into `bufs` using vectored
216 /// IO operations.
217 ///
218 /// The returned future will resolve to the number of bytes read once the read
219 /// operation is completed.
read_vectored<'a>(&'a mut self, bufs: &'a mut [IoSliceMut<'a>]) -> ReadVectored<'a, Self> where Self: Unpin,220 fn read_vectored<'a>(&'a mut self, bufs: &'a mut [IoSliceMut<'a>]) -> ReadVectored<'a, Self>
221 where
222 Self: Unpin,
223 {
224 assert_future::<Result<usize>, _>(ReadVectored::new(self, bufs))
225 }
226
227 /// Creates a future which will read exactly enough bytes to fill `buf`,
228 /// returning an error if end of file (EOF) is hit sooner.
229 ///
230 /// The returned future will resolve once the read operation is completed.
231 ///
232 /// In the case of an error the buffer and the object will be discarded, with
233 /// the error yielded.
234 ///
235 /// # Examples
236 ///
237 /// ```
238 /// # futures::executor::block_on(async {
239 /// use futures::io::{AsyncReadExt, Cursor};
240 ///
241 /// let mut reader = Cursor::new([1, 2, 3, 4]);
242 /// let mut output = [0u8; 4];
243 ///
244 /// reader.read_exact(&mut output).await?;
245 ///
246 /// assert_eq!(output, [1, 2, 3, 4]);
247 /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
248 /// ```
249 ///
250 /// ## EOF is hit before `buf` is filled
251 ///
252 /// ```
253 /// # futures::executor::block_on(async {
254 /// use futures::io::{self, AsyncReadExt, Cursor};
255 ///
256 /// let mut reader = Cursor::new([1, 2, 3, 4]);
257 /// let mut output = [0u8; 5];
258 ///
259 /// let result = reader.read_exact(&mut output).await;
260 ///
261 /// assert_eq!(result.unwrap_err().kind(), io::ErrorKind::UnexpectedEof);
262 /// # });
263 /// ```
read_exact<'a>(&'a mut self, buf: &'a mut [u8]) -> ReadExact<'a, Self> where Self: Unpin,264 fn read_exact<'a>(&'a mut self, buf: &'a mut [u8]) -> ReadExact<'a, Self>
265 where
266 Self: Unpin,
267 {
268 assert_future::<Result<()>, _>(ReadExact::new(self, buf))
269 }
270
271 /// Creates a future which will read all the bytes from this `AsyncRead`.
272 ///
273 /// On success the total number of bytes read is returned.
274 ///
275 /// # Examples
276 ///
277 /// ```
278 /// # futures::executor::block_on(async {
279 /// use futures::io::{AsyncReadExt, Cursor};
280 ///
281 /// let mut reader = Cursor::new([1, 2, 3, 4]);
282 /// let mut output = Vec::with_capacity(4);
283 ///
284 /// let bytes = reader.read_to_end(&mut output).await?;
285 ///
286 /// assert_eq!(bytes, 4);
287 /// assert_eq!(output, vec![1, 2, 3, 4]);
288 /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
289 /// ```
read_to_end<'a>(&'a mut self, buf: &'a mut Vec<u8>) -> ReadToEnd<'a, Self> where Self: Unpin,290 fn read_to_end<'a>(&'a mut self, buf: &'a mut Vec<u8>) -> ReadToEnd<'a, Self>
291 where
292 Self: Unpin,
293 {
294 assert_future::<Result<usize>, _>(ReadToEnd::new(self, buf))
295 }
296
297 /// Creates a future which will read all the bytes from this `AsyncRead`.
298 ///
299 /// On success the total number of bytes read is returned.
300 ///
301 /// # Examples
302 ///
303 /// ```
304 /// # futures::executor::block_on(async {
305 /// use futures::io::{AsyncReadExt, Cursor};
306 ///
307 /// let mut reader = Cursor::new(&b"1234"[..]);
308 /// let mut buffer = String::with_capacity(4);
309 ///
310 /// let bytes = reader.read_to_string(&mut buffer).await?;
311 ///
312 /// assert_eq!(bytes, 4);
313 /// assert_eq!(buffer, String::from("1234"));
314 /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
315 /// ```
read_to_string<'a>(&'a mut self, buf: &'a mut String) -> ReadToString<'a, Self> where Self: Unpin,316 fn read_to_string<'a>(&'a mut self, buf: &'a mut String) -> ReadToString<'a, Self>
317 where
318 Self: Unpin,
319 {
320 assert_future::<Result<usize>, _>(ReadToString::new(self, buf))
321 }
322
323 /// Helper method for splitting this read/write object into two halves.
324 ///
325 /// The two halves returned implement the `AsyncRead` and `AsyncWrite`
326 /// traits, respectively.
327 ///
328 /// # Examples
329 ///
330 /// ```
331 /// # futures::executor::block_on(async {
332 /// use futures::io::{self, AsyncReadExt, Cursor};
333 ///
334 /// // Note that for `Cursor` the read and write halves share a single
335 /// // seek position. This may or may not be true for other types that
336 /// // implement both `AsyncRead` and `AsyncWrite`.
337 ///
338 /// let reader = Cursor::new([1, 2, 3, 4]);
339 /// let mut buffer = Cursor::new(vec![0, 0, 0, 0, 5, 6, 7, 8]);
340 /// let mut writer = Cursor::new(vec![0u8; 5]);
341 ///
342 /// {
343 /// let (buffer_reader, mut buffer_writer) = (&mut buffer).split();
344 /// io::copy(reader, &mut buffer_writer).await?;
345 /// io::copy(buffer_reader, &mut writer).await?;
346 /// }
347 ///
348 /// assert_eq!(buffer.into_inner(), [1, 2, 3, 4, 5, 6, 7, 8]);
349 /// assert_eq!(writer.into_inner(), [5, 6, 7, 8, 0]);
350 /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
351 /// ```
split(self) -> (ReadHalf<Self>, WriteHalf<Self>) where Self: AsyncWrite + Sized,352 fn split(self) -> (ReadHalf<Self>, WriteHalf<Self>)
353 where
354 Self: AsyncWrite + Sized,
355 {
356 let (r, w) = split::split(self);
357 (assert_read(r), assert_write(w))
358 }
359
360 /// Creates an AsyncRead adapter which will read at most `limit` bytes
361 /// from the underlying reader.
362 ///
363 /// # Examples
364 ///
365 /// ```
366 /// # futures::executor::block_on(async {
367 /// use futures::io::{AsyncReadExt, Cursor};
368 ///
369 /// let reader = Cursor::new(&b"12345678"[..]);
370 /// let mut buffer = [0; 5];
371 ///
372 /// let mut take = reader.take(4);
373 /// let n = take.read(&mut buffer).await?;
374 ///
375 /// assert_eq!(n, 4);
376 /// assert_eq!(&buffer, b"1234\0");
377 /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
378 /// ```
take(self, limit: u64) -> Take<Self> where Self: Sized,379 fn take(self, limit: u64) -> Take<Self>
380 where
381 Self: Sized,
382 {
383 assert_read(Take::new(self, limit))
384 }
385
386 /// Wraps an [`AsyncRead`] in a compatibility wrapper that allows it to be
387 /// used as a futures 0.1 / tokio-io 0.1 `AsyncRead`. If the wrapped type
388 /// implements [`AsyncWrite`] as well, the result will also implement the
389 /// futures 0.1 / tokio 0.1 `AsyncWrite` trait.
390 ///
391 /// Requires the `io-compat` feature to enable.
392 #[cfg(feature = "io-compat")]
393 #[cfg_attr(docsrs, doc(cfg(feature = "io-compat")))]
compat(self) -> Compat<Self> where Self: Sized + Unpin,394 fn compat(self) -> Compat<Self>
395 where
396 Self: Sized + Unpin,
397 {
398 Compat::new(self)
399 }
400 }
401
402 impl<R: AsyncRead + ?Sized> AsyncReadExt for R {}
403
404 /// An extension trait which adds utility methods to `AsyncWrite` types.
405 pub trait AsyncWriteExt: AsyncWrite {
406 /// Creates a future which will entirely flush this `AsyncWrite`.
407 ///
408 /// # Examples
409 ///
410 /// ```
411 /// # futures::executor::block_on(async {
412 /// use futures::io::{AllowStdIo, AsyncWriteExt};
413 /// use std::io::{BufWriter, Cursor};
414 ///
415 /// let mut output = vec![0u8; 5];
416 ///
417 /// {
418 /// let writer = Cursor::new(&mut output);
419 /// let mut buffered = AllowStdIo::new(BufWriter::new(writer));
420 /// buffered.write_all(&[1, 2]).await?;
421 /// buffered.write_all(&[3, 4]).await?;
422 /// buffered.flush().await?;
423 /// }
424 ///
425 /// assert_eq!(output, [1, 2, 3, 4, 0]);
426 /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
427 /// ```
flush(&mut self) -> Flush<'_, Self> where Self: Unpin,428 fn flush(&mut self) -> Flush<'_, Self>
429 where
430 Self: Unpin,
431 {
432 assert_future::<Result<()>, _>(Flush::new(self))
433 }
434
435 /// Creates a future which will entirely close this `AsyncWrite`.
close(&mut self) -> Close<'_, Self> where Self: Unpin,436 fn close(&mut self) -> Close<'_, Self>
437 where
438 Self: Unpin,
439 {
440 assert_future::<Result<()>, _>(Close::new(self))
441 }
442
443 /// Creates a future which will write bytes from `buf` into the object.
444 ///
445 /// The returned future will resolve to the number of bytes written once the write
446 /// operation is completed.
write<'a>(&'a mut self, buf: &'a [u8]) -> Write<'a, Self> where Self: Unpin,447 fn write<'a>(&'a mut self, buf: &'a [u8]) -> Write<'a, Self>
448 where
449 Self: Unpin,
450 {
451 assert_future::<Result<usize>, _>(Write::new(self, buf))
452 }
453
454 /// Creates a future which will write bytes from `bufs` into the object using vectored
455 /// IO operations.
456 ///
457 /// The returned future will resolve to the number of bytes written once the write
458 /// operation is completed.
write_vectored<'a>(&'a mut self, bufs: &'a [IoSlice<'a>]) -> WriteVectored<'a, Self> where Self: Unpin,459 fn write_vectored<'a>(&'a mut self, bufs: &'a [IoSlice<'a>]) -> WriteVectored<'a, Self>
460 where
461 Self: Unpin,
462 {
463 assert_future::<Result<usize>, _>(WriteVectored::new(self, bufs))
464 }
465
466 /// Write data into this object.
467 ///
468 /// Creates a future that will write the entire contents of the buffer `buf` into
469 /// this `AsyncWrite`.
470 ///
471 /// The returned future will not complete until all the data has been written.
472 ///
473 /// # Examples
474 ///
475 /// ```
476 /// # futures::executor::block_on(async {
477 /// use futures::io::{AsyncWriteExt, Cursor};
478 ///
479 /// let mut writer = Cursor::new(vec![0u8; 5]);
480 ///
481 /// writer.write_all(&[1, 2, 3, 4]).await?;
482 ///
483 /// assert_eq!(writer.into_inner(), [1, 2, 3, 4, 0]);
484 /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
485 /// ```
write_all<'a>(&'a mut self, buf: &'a [u8]) -> WriteAll<'a, Self> where Self: Unpin,486 fn write_all<'a>(&'a mut self, buf: &'a [u8]) -> WriteAll<'a, Self>
487 where
488 Self: Unpin,
489 {
490 assert_future::<Result<()>, _>(WriteAll::new(self, buf))
491 }
492
493 /// Attempts to write multiple buffers into this writer.
494 ///
495 /// Creates a future that will write the entire contents of `bufs` into this
496 /// `AsyncWrite` using [vectored writes].
497 ///
498 /// The returned future will not complete until all the data has been
499 /// written.
500 ///
501 /// [vectored writes]: std::io::Write::write_vectored
502 ///
503 /// # Notes
504 ///
505 /// Unlike `io::Write::write_vectored`, this takes a *mutable* reference to
506 /// a slice of `IoSlice`s, not an immutable one. That's because we need to
507 /// modify the slice to keep track of the bytes already written.
508 ///
509 /// Once this futures returns, the contents of `bufs` are unspecified, as
510 /// this depends on how many calls to `write_vectored` were necessary. It is
511 /// best to understand this function as taking ownership of `bufs` and to
512 /// not use `bufs` afterwards. The underlying buffers, to which the
513 /// `IoSlice`s point (but not the `IoSlice`s themselves), are unchanged and
514 /// can be reused.
515 ///
516 /// # Examples
517 ///
518 /// ```
519 /// # futures::executor::block_on(async {
520 /// use futures::io::AsyncWriteExt;
521 /// use futures_util::io::Cursor;
522 /// use std::io::IoSlice;
523 ///
524 /// let mut writer = Cursor::new(Vec::new());
525 /// let bufs = &mut [
526 /// IoSlice::new(&[1]),
527 /// IoSlice::new(&[2, 3]),
528 /// IoSlice::new(&[4, 5, 6]),
529 /// ];
530 ///
531 /// writer.write_all_vectored(bufs).await?;
532 /// // Note: the contents of `bufs` is now unspecified, see the Notes section.
533 ///
534 /// assert_eq!(writer.into_inner(), &[1, 2, 3, 4, 5, 6]);
535 /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
536 /// ```
537 #[cfg(feature = "write-all-vectored")]
write_all_vectored<'a>( &'a mut self, bufs: &'a mut [IoSlice<'a>], ) -> WriteAllVectored<'a, Self> where Self: Unpin,538 fn write_all_vectored<'a>(
539 &'a mut self,
540 bufs: &'a mut [IoSlice<'a>],
541 ) -> WriteAllVectored<'a, Self>
542 where
543 Self: Unpin,
544 {
545 assert_future::<Result<()>, _>(WriteAllVectored::new(self, bufs))
546 }
547
548 /// Wraps an [`AsyncWrite`] in a compatibility wrapper that allows it to be
549 /// used as a futures 0.1 / tokio-io 0.1 `AsyncWrite`.
550 /// Requires the `io-compat` feature to enable.
551 #[cfg(feature = "io-compat")]
552 #[cfg_attr(docsrs, doc(cfg(feature = "io-compat")))]
compat_write(self) -> Compat<Self> where Self: Sized + Unpin,553 fn compat_write(self) -> Compat<Self>
554 where
555 Self: Sized + Unpin,
556 {
557 Compat::new(self)
558 }
559
560 /// Allow using an [`AsyncWrite`] as a [`Sink`](futures_sink::Sink)`<Item: AsRef<[u8]>>`.
561 ///
562 /// This adapter produces a sink that will write each value passed to it
563 /// into the underlying writer.
564 ///
565 /// Note that this function consumes the given writer, returning a wrapped
566 /// version.
567 ///
568 /// # Examples
569 ///
570 /// ```
571 /// # futures::executor::block_on(async {
572 /// use futures::io::AsyncWriteExt;
573 /// use futures::stream::{self, StreamExt};
574 ///
575 /// let stream = stream::iter(vec![Ok([1, 2, 3]), Ok([4, 5, 6])]);
576 ///
577 /// let mut writer = vec![];
578 ///
579 /// stream.forward((&mut writer).into_sink()).await?;
580 ///
581 /// assert_eq!(writer, vec![1, 2, 3, 4, 5, 6]);
582 /// # Ok::<(), Box<dyn std::error::Error>>(())
583 /// # })?;
584 /// # Ok::<(), Box<dyn std::error::Error>>(())
585 /// ```
586 #[cfg(feature = "sink")]
587 #[cfg_attr(docsrs, doc(cfg(feature = "sink")))]
into_sink<Item: AsRef<[u8]>>(self) -> IntoSink<Self, Item> where Self: Sized,588 fn into_sink<Item: AsRef<[u8]>>(self) -> IntoSink<Self, Item>
589 where
590 Self: Sized,
591 {
592 crate::sink::assert_sink::<Item, Error, _>(IntoSink::new(self))
593 }
594 }
595
596 impl<W: AsyncWrite + ?Sized> AsyncWriteExt for W {}
597
598 /// An extension trait which adds utility methods to `AsyncSeek` types.
599 pub trait AsyncSeekExt: AsyncSeek {
600 /// Creates a future which will seek an IO object, and then yield the
601 /// new position in the object and the object itself.
602 ///
603 /// In the case of an error the buffer and the object will be discarded, with
604 /// the error yielded.
seek(&mut self, pos: SeekFrom) -> Seek<'_, Self> where Self: Unpin,605 fn seek(&mut self, pos: SeekFrom) -> Seek<'_, Self>
606 where
607 Self: Unpin,
608 {
609 assert_future::<Result<u64>, _>(Seek::new(self, pos))
610 }
611
612 /// Creates a future which will return the current seek position from the
613 /// start of the stream.
614 ///
615 /// This is equivalent to `self.seek(SeekFrom::Current(0))`.
stream_position(&mut self) -> Seek<'_, Self> where Self: Unpin,616 fn stream_position(&mut self) -> Seek<'_, Self>
617 where
618 Self: Unpin,
619 {
620 self.seek(SeekFrom::Current(0))
621 }
622 }
623
624 impl<S: AsyncSeek + ?Sized> AsyncSeekExt for S {}
625
626 /// An extension trait which adds utility methods to `AsyncBufRead` types.
627 pub trait AsyncBufReadExt: AsyncBufRead {
628 /// Creates a future which will wait for a non-empty buffer to be available from this I/O
629 /// object or EOF to be reached.
630 ///
631 /// This method is the async equivalent to [`BufRead::fill_buf`](std::io::BufRead::fill_buf).
632 ///
633 /// ```rust
634 /// # futures::executor::block_on(async {
635 /// use futures::{io::AsyncBufReadExt as _, stream::{iter, TryStreamExt as _}};
636 ///
637 /// let mut stream = iter(vec![Ok(vec![1, 2, 3]), Ok(vec![4, 5, 6])]).into_async_read();
638 ///
639 /// assert_eq!(stream.fill_buf().await?, vec![1, 2, 3]);
640 /// stream.consume_unpin(2);
641 ///
642 /// assert_eq!(stream.fill_buf().await?, vec![3]);
643 /// stream.consume_unpin(1);
644 ///
645 /// assert_eq!(stream.fill_buf().await?, vec![4, 5, 6]);
646 /// stream.consume_unpin(3);
647 ///
648 /// assert_eq!(stream.fill_buf().await?, vec![]);
649 /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
650 /// ```
fill_buf(&mut self) -> FillBuf<'_, Self> where Self: Unpin,651 fn fill_buf(&mut self) -> FillBuf<'_, Self>
652 where
653 Self: Unpin,
654 {
655 assert_future::<Result<&[u8]>, _>(FillBuf::new(self))
656 }
657
658 /// A convenience for calling [`AsyncBufRead::consume`] on [`Unpin`] IO types.
659 ///
660 /// ```rust
661 /// # futures::executor::block_on(async {
662 /// use futures::{io::AsyncBufReadExt as _, stream::{iter, TryStreamExt as _}};
663 ///
664 /// let mut stream = iter(vec![Ok(vec![1, 2, 3])]).into_async_read();
665 ///
666 /// assert_eq!(stream.fill_buf().await?, vec![1, 2, 3]);
667 /// stream.consume_unpin(2);
668 ///
669 /// assert_eq!(stream.fill_buf().await?, vec![3]);
670 /// stream.consume_unpin(1);
671 ///
672 /// assert_eq!(stream.fill_buf().await?, vec![]);
673 /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
674 /// ```
consume_unpin(&mut self, amt: usize) where Self: Unpin,675 fn consume_unpin(&mut self, amt: usize)
676 where
677 Self: Unpin,
678 {
679 Pin::new(self).consume(amt)
680 }
681
682 /// Creates a future which will read all the bytes associated with this I/O
683 /// object into `buf` until the delimiter `byte` or EOF is reached.
684 /// This method is the async equivalent to [`BufRead::read_until`](std::io::BufRead::read_until).
685 ///
686 /// This function will read bytes from the underlying stream until the
687 /// delimiter or EOF is found. Once found, all bytes up to, and including,
688 /// the delimiter (if found) will be appended to `buf`.
689 ///
690 /// The returned future will resolve to the number of bytes read once the read
691 /// operation is completed.
692 ///
693 /// In the case of an error the buffer and the object will be discarded, with
694 /// the error yielded.
695 ///
696 /// # Examples
697 ///
698 /// ```
699 /// # futures::executor::block_on(async {
700 /// use futures::io::{AsyncBufReadExt, Cursor};
701 ///
702 /// let mut cursor = Cursor::new(b"lorem-ipsum");
703 /// let mut buf = vec![];
704 ///
705 /// // cursor is at 'l'
706 /// let num_bytes = cursor.read_until(b'-', &mut buf).await?;
707 /// assert_eq!(num_bytes, 6);
708 /// assert_eq!(buf, b"lorem-");
709 /// buf.clear();
710 ///
711 /// // cursor is at 'i'
712 /// let num_bytes = cursor.read_until(b'-', &mut buf).await?;
713 /// assert_eq!(num_bytes, 5);
714 /// assert_eq!(buf, b"ipsum");
715 /// buf.clear();
716 ///
717 /// // cursor is at EOF
718 /// let num_bytes = cursor.read_until(b'-', &mut buf).await?;
719 /// assert_eq!(num_bytes, 0);
720 /// assert_eq!(buf, b"");
721 /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
722 /// ```
read_until<'a>(&'a mut self, byte: u8, buf: &'a mut Vec<u8>) -> ReadUntil<'a, Self> where Self: Unpin,723 fn read_until<'a>(&'a mut self, byte: u8, buf: &'a mut Vec<u8>) -> ReadUntil<'a, Self>
724 where
725 Self: Unpin,
726 {
727 assert_future::<Result<usize>, _>(ReadUntil::new(self, byte, buf))
728 }
729
730 /// Creates a future which will read all the bytes associated with this I/O
731 /// object into `buf` until a newline (the 0xA byte) or EOF is reached,
732 /// This method is the async equivalent to [`BufRead::read_line`](std::io::BufRead::read_line).
733 ///
734 /// This function will read bytes from the underlying stream until the
735 /// newline delimiter (the 0xA byte) or EOF is found. Once found, all bytes
736 /// up to, and including, the delimiter (if found) will be appended to
737 /// `buf`.
738 ///
739 /// The returned future will resolve to the number of bytes read once the read
740 /// operation is completed.
741 ///
742 /// In the case of an error the buffer and the object will be discarded, with
743 /// the error yielded.
744 ///
745 /// # Errors
746 ///
747 /// This function has the same error semantics as [`read_until`] and will
748 /// also return an error if the read bytes are not valid UTF-8. If an I/O
749 /// error is encountered then `buf` may contain some bytes already read in
750 /// the event that all data read so far was valid UTF-8.
751 ///
752 /// [`read_until`]: AsyncBufReadExt::read_until
753 ///
754 /// # Examples
755 ///
756 /// ```
757 /// # futures::executor::block_on(async {
758 /// use futures::io::{AsyncBufReadExt, Cursor};
759 ///
760 /// let mut cursor = Cursor::new(b"foo\nbar");
761 /// let mut buf = String::new();
762 ///
763 /// // cursor is at 'f'
764 /// let num_bytes = cursor.read_line(&mut buf).await?;
765 /// assert_eq!(num_bytes, 4);
766 /// assert_eq!(buf, "foo\n");
767 /// buf.clear();
768 ///
769 /// // cursor is at 'b'
770 /// let num_bytes = cursor.read_line(&mut buf).await?;
771 /// assert_eq!(num_bytes, 3);
772 /// assert_eq!(buf, "bar");
773 /// buf.clear();
774 ///
775 /// // cursor is at EOF
776 /// let num_bytes = cursor.read_line(&mut buf).await?;
777 /// assert_eq!(num_bytes, 0);
778 /// assert_eq!(buf, "");
779 /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
780 /// ```
read_line<'a>(&'a mut self, buf: &'a mut String) -> ReadLine<'a, Self> where Self: Unpin,781 fn read_line<'a>(&'a mut self, buf: &'a mut String) -> ReadLine<'a, Self>
782 where
783 Self: Unpin,
784 {
785 assert_future::<Result<usize>, _>(ReadLine::new(self, buf))
786 }
787
788 /// Returns a stream over the lines of this reader.
789 /// This method is the async equivalent to [`BufRead::lines`](std::io::BufRead::lines).
790 ///
791 /// The stream returned from this function will yield instances of
792 /// [`io::Result`]`<`[`String`]`>`. Each string returned will *not* have a newline
793 /// byte (the 0xA byte) or CRLF (0xD, 0xA bytes) at the end.
794 ///
795 /// [`io::Result`]: std::io::Result
796 /// [`String`]: String
797 ///
798 /// # Errors
799 ///
800 /// Each line of the stream has the same error semantics as [`AsyncBufReadExt::read_line`].
801 ///
802 /// [`AsyncBufReadExt::read_line`]: AsyncBufReadExt::read_line
803 ///
804 /// # Examples
805 ///
806 /// ```
807 /// # futures::executor::block_on(async {
808 /// use futures::io::{AsyncBufReadExt, Cursor};
809 /// use futures::stream::StreamExt;
810 ///
811 /// let cursor = Cursor::new(b"lorem\nipsum\r\ndolor");
812 ///
813 /// let mut lines_stream = cursor.lines().map(|l| l.unwrap());
814 /// assert_eq!(lines_stream.next().await, Some(String::from("lorem")));
815 /// assert_eq!(lines_stream.next().await, Some(String::from("ipsum")));
816 /// assert_eq!(lines_stream.next().await, Some(String::from("dolor")));
817 /// assert_eq!(lines_stream.next().await, None);
818 /// # Ok::<(), Box<dyn std::error::Error>>(()) }).unwrap();
819 /// ```
lines(self) -> Lines<Self> where Self: Sized,820 fn lines(self) -> Lines<Self>
821 where
822 Self: Sized,
823 {
824 assert_stream::<Result<String>, _>(Lines::new(self))
825 }
826 }
827
828 impl<R: AsyncBufRead + ?Sized> AsyncBufReadExt for R {}
829
830 // Just a helper function to ensure the reader we're returning all have the
831 // right implementations.
assert_read<R>(reader: R) -> R where R: AsyncRead,832 pub(crate) fn assert_read<R>(reader: R) -> R
833 where
834 R: AsyncRead,
835 {
836 reader
837 }
838 // Just a helper function to ensure the writer we're returning all have the
839 // right implementations.
assert_write<W>(writer: W) -> W where W: AsyncWrite,840 pub(crate) fn assert_write<W>(writer: W) -> W
841 where
842 W: AsyncWrite,
843 {
844 writer
845 }
846