combine/stream/
buf_reader.rs

1use std::io::{self, BufRead, Read};
2
3#[cfg(any(
4    feature = "futures-03",
5    feature = "tokio-02",
6    feature = "tokio-03",
7    feature = "tokio"
8))]
9use std::pin::Pin;
10
11#[cfg(any(feature = "futures-03", feature = "tokio-02", feature = "tokio-03"))]
12use std::mem::MaybeUninit;
13
14#[cfg(feature = "futures-core-03")]
15use std::task::{Context, Poll};
16
17#[cfg(feature = "futures-03")]
18use std::future::Future;
19
20use bytes::{Buf, BufMut, BytesMut};
21
22#[cfg(feature = "pin-project-lite")]
23use pin_project_lite::pin_project;
24
25#[cfg(feature = "tokio-03")]
26use tokio_03_dep::io::AsyncBufRead as _;
27
28#[cfg(feature = "tokio")]
29use tokio_dep::io::AsyncBufRead as _;
30
31#[cfg(feature = "futures-core-03")]
32use futures_core_03::ready;
33
34#[cfg(feature = "pin-project-lite")]
35pin_project! {
36    /// `BufReader` used by `Decoder` when it is constructed with [`Decoder::new_bufferless`][]
37    ///
38    /// [`Decoder::new_bufferless`]: ../decoder/struct.Decoder.html#method.new_bufferless
39    #[derive(Debug)]
40    pub struct BufReader<R> {
41        #[pin]
42        inner: R,
43        buf: BytesMut
44    }
45}
46
47#[cfg(not(feature = "pin-project-lite"))]
48/// `BufReader` used by `Decoder` when it is constructed with [`Decoder::new_bufferless`][]
49///
50/// [`Decoder::new_bufferless`]: ../decoder/struct.Decoder.html#method.new_bufferless
51#[derive(Debug)]
52pub struct BufReader<R> {
53    inner: R,
54    buf: BytesMut,
55}
56
57impl<R> BufReader<R> {
58    /// Creates a new `BufReader` with a default buffer capacity. The default is currently 8 KB,
59    /// but may change in the future.
60    pub fn new(inner: R) -> Self {
61        Self::with_capacity(8096, inner)
62    }
63
64    /// Creates a new `BufReader` with the specified buffer capacity.
65    pub fn with_capacity(capacity: usize, inner: R) -> Self {
66        let buf = BytesMut::with_capacity(capacity);
67
68        Self { inner, buf }
69    }
70
71    /// Gets a reference to the underlying reader.
72    ///
73    /// It is inadvisable to directly read from the underlying reader.
74    pub fn get_ref(&self) -> &R {
75        &self.inner
76    }
77
78    /// Gets a mutable reference to the underlying reader.
79    ///
80    /// It is inadvisable to directly read from the underlying reader.
81    pub fn get_mut(&mut self) -> &mut R {
82        &mut self.inner
83    }
84
85    #[cfg(feature = "pin-project-lite")]
86    /// Gets a pinned mutable reference to the underlying reader.
87    ///
88    /// It is inadvisable to directly read from the underlying reader.
89    pub fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut R> {
90        self.project().inner
91    }
92
93    /// Consumes this `BufWriter`, returning the underlying reader.
94    ///
95    /// Note that any leftover data in the internal buffer is lost.
96    pub fn into_inner(self) -> R {
97        self.inner
98    }
99
100    /// Returns a reference to the internally buffered data.
101    ///
102    /// Unlike `fill_buf`, this will not attempt to fill the buffer if it is empty.
103    pub fn buffer(&self) -> &[u8] {
104        &self.buf
105    }
106
107    /// Invalidates all data in the internal buffer.
108    #[inline]
109    #[cfg(any(feature = "tokio-02", feature = "tokio-03", feature = "tokio"))]
110    fn discard_buffer(self: Pin<&mut Self>) {
111        let me = self.project();
112        me.buf.clear();
113    }
114}
115
116mod sealed {
117    pub trait Sealed {}
118}
119
120#[doc(hidden)]
121pub trait CombineBuffer<R>: sealed::Sealed {
122    fn buffer<'a>(&'a self, read: &'a R) -> &'a [u8];
123
124    fn advance(&mut self, read: &mut R, len: usize);
125
126    #[cfg(feature = "pin-project-lite")]
127    fn advance_pin(&mut self, read: Pin<&mut R>, len: usize);
128}
129
130#[doc(hidden)]
131pub trait CombineSyncRead<R>: CombineBuffer<R> {
132    fn extend_buf_sync(&mut self, read: &mut R) -> io::Result<usize>;
133}
134
135#[cfg(any(feature = "tokio-02", feature = "tokio-03", feature = "tokio"))]
136#[doc(hidden)]
137pub trait CombineRead<R, T: ?Sized>: CombineBuffer<R> {
138    fn poll_extend_buf(
139        &mut self,
140        cx: &mut Context<'_>,
141        read: Pin<&mut R>,
142    ) -> Poll<io::Result<usize>>;
143}
144
145#[cfg(feature = "futures-03")]
146#[doc(hidden)]
147pub trait CombineAsyncRead<R>: CombineBuffer<R> {
148    fn poll_extend_buf(
149        &mut self,
150        cx: &mut Context<'_>,
151        read: Pin<&mut R>,
152    ) -> Poll<io::Result<usize>>;
153
154    fn extend_buf<'a>(&'a mut self, read: Pin<&'a mut R>) -> ExtendBuf<'a, Self, R>
155    where
156        Self: Sized;
157}
158
159#[cfg(feature = "futures-03")]
160pin_project_lite::pin_project! {
161    #[doc(hidden)]
162    pub struct ExtendBuf<'a, C, R> {
163        buffer: &'a mut C,
164        read: Pin<&'a mut R>
165    }
166}
167
168#[cfg(feature = "futures-03")]
169impl<'a, C, R> Future for ExtendBuf<'a, C, R>
170where
171    C: CombineAsyncRead<R>,
172{
173    type Output = io::Result<usize>;
174
175    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
176        let me = self.project();
177        me.buffer.poll_extend_buf(cx, me.read.as_mut())
178    }
179}
180
181/// Marker used by `Decoder` for an internal buffer
182#[derive(Default)]
183pub struct Buffer(pub(crate) BytesMut);
184
185impl sealed::Sealed for Buffer {}
186
187impl<R> CombineBuffer<R> for Buffer {
188    fn buffer<'a>(&'a self, _read: &'a R) -> &'a [u8] {
189        &self.0
190    }
191
192    fn advance(&mut self, _read: &mut R, len: usize) {
193        self.0.advance(len);
194    }
195
196    #[cfg(feature = "pin-project-lite")]
197    fn advance_pin(&mut self, _read: Pin<&mut R>, len: usize) {
198        self.0.advance(len);
199    }
200}
201
202impl<R> CombineSyncRead<R> for Buffer
203where
204    R: Read,
205{
206    fn extend_buf_sync(&mut self, read: &mut R) -> io::Result<usize> {
207        extend_buf_sync(&mut self.0, read)
208    }
209}
210
211#[cfg(feature = "futures-03")]
212impl<R> CombineAsyncRead<R> for Buffer
213where
214    R: futures_io_03::AsyncRead,
215{
216    fn poll_extend_buf(
217        &mut self,
218        cx: &mut Context<'_>,
219        read: Pin<&mut R>,
220    ) -> Poll<io::Result<usize>> {
221        poll_extend_buf(&mut self.0, cx, read)
222    }
223
224    fn extend_buf<'a>(&'a mut self, read: Pin<&'a mut R>) -> ExtendBuf<'a, Self, R> {
225        if !self.0.has_remaining_mut() {
226            self.0.reserve(8 * 1024);
227        }
228        // Copy of tokio's read_buf method (but it has to force initialize the buffer)
229        let bs = self.0.chunk_mut();
230
231        for i in 0..bs.len() {
232            bs.write_byte(i, 0);
233        }
234        ExtendBuf { buffer: self, read }
235    }
236}
237
238#[cfg(feature = "tokio-02")]
239impl<R> CombineRead<R, dyn tokio_02_dep::io::AsyncRead> for Buffer
240where
241    R: tokio_02_dep::io::AsyncRead,
242{
243    fn poll_extend_buf(
244        &mut self,
245        cx: &mut Context<'_>,
246        read: Pin<&mut R>,
247    ) -> Poll<io::Result<usize>> {
248        if !self.0.has_remaining_mut() {
249            self.0.reserve(8 * 1024);
250        }
251        read.poll_read_buf(cx, &mut Bytes05(&mut self.0))
252    }
253}
254
255#[cfg(feature = "tokio-03")]
256fn tokio_03_to_read_buf(bs: &mut BytesMut) -> tokio_03_dep::io::ReadBuf<'_> {
257    let uninit = bs.chunk_mut();
258    unsafe {
259        tokio_03_dep::io::ReadBuf::uninit(std::slice::from_raw_parts_mut(
260            uninit.as_mut_ptr() as *mut MaybeUninit<u8>,
261            uninit.len(),
262        ))
263    }
264}
265
266#[cfg(feature = "tokio-03")]
267impl<R> CombineRead<R, dyn tokio_03_dep::io::AsyncRead> for Buffer
268where
269    R: tokio_03_dep::io::AsyncRead,
270{
271    fn poll_extend_buf(
272        &mut self,
273        cx: &mut Context<'_>,
274        read: Pin<&mut R>,
275    ) -> Poll<io::Result<usize>> {
276        tokio_03_read_buf(cx, read, &mut self.0)
277    }
278}
279
280#[cfg(feature = "tokio-03")]
281fn tokio_03_read_buf(
282    cx: &mut Context<'_>,
283    read: Pin<&mut impl tokio_03_dep::io::AsyncRead>,
284    bs: &mut bytes::BytesMut,
285) -> Poll<io::Result<usize>> {
286    if !bs.has_remaining_mut() {
287        bs.reserve(8 * 1024);
288    }
289
290    let mut buf = tokio_03_to_read_buf(bs);
291    ready!(read.poll_read(cx, &mut buf))?;
292    unsafe {
293        let n = buf.filled().len();
294        bs.advance_mut(n);
295        Poll::Ready(Ok(n))
296    }
297}
298
299#[cfg(feature = "tokio")]
300impl<R> CombineRead<R, dyn tokio_dep::io::AsyncRead> for Buffer
301where
302    R: tokio_dep::io::AsyncRead,
303{
304    fn poll_extend_buf(
305        &mut self,
306        cx: &mut Context<'_>,
307        read: Pin<&mut R>,
308    ) -> Poll<io::Result<usize>> {
309        tokio_read_buf(read, cx, &mut self.0)
310    }
311}
312
313#[cfg(feature = "tokio")]
314fn tokio_read_buf(
315    read: Pin<&mut impl tokio_dep::io::AsyncRead>,
316    cx: &mut Context<'_>,
317    bs: &mut bytes::BytesMut,
318) -> Poll<io::Result<usize>> {
319    if !bs.has_remaining_mut() {
320        bs.reserve(8 * 1024);
321    }
322
323    tokio_util::io::poll_read_buf(read, cx, bs)
324}
325
326/// Marker used by `Decoder` for an external buffer
327#[derive(Default)]
328pub struct Bufferless;
329
330impl sealed::Sealed for Bufferless {}
331
332impl<R> CombineBuffer<BufReader<R>> for Bufferless {
333    fn buffer<'a>(&'a self, read: &'a BufReader<R>) -> &'a [u8] {
334        &read.buf
335    }
336
337    fn advance(&mut self, read: &mut BufReader<R>, len: usize) {
338        read.buf.advance(len);
339    }
340
341    #[cfg(feature = "pin-project-lite")]
342    fn advance_pin(&mut self, read: Pin<&mut BufReader<R>>, len: usize) {
343        read.project().buf.advance(len);
344    }
345}
346
347impl<R> CombineSyncRead<BufReader<R>> for Bufferless
348where
349    R: Read,
350{
351    fn extend_buf_sync(&mut self, read: &mut BufReader<R>) -> io::Result<usize> {
352        extend_buf_sync(&mut read.buf, &mut read.inner)
353    }
354}
355
356fn extend_buf_sync<R>(buf: &mut BytesMut, read: &mut R) -> io::Result<usize>
357where
358    R: Read,
359{
360    let size = 8 * 1024;
361    if buf.capacity() == buf.len() {
362        buf.reserve(size);
363    }
364
365    // Copy of tokio's poll_read_buf method (but it has to force initialize the buffer)
366    let n = {
367        let bs = buf.spare_capacity_mut();
368
369        let initial_size = bs.len().min(size);
370        let bs = &mut bs[..initial_size];
371        // SAFETY: the above slicing operation guarantees `bs.len() == initial_size`
372        unsafe { bs.as_mut_ptr().cast::<u8>().write_bytes(0, initial_size) }
373
374        // Convert to `&mut [u8]`
375        // SAFETY: the entire buffer is preinitialized above
376        let bs = unsafe { &mut *(bs as *mut _ as *mut [u8]) };
377
378        let n = read.read(bs)?;
379        assert!(
380            n <= bs.len(),
381            "AsyncRead reported that it initialized more than the number of bytes in the buffer"
382        );
383        n
384    };
385
386    // SAFETY: the entire buffer has been preinitialized
387    unsafe { buf.advance_mut(n) };
388
389    Ok(n)
390}
391
392#[cfg(feature = "tokio-02")]
393struct Bytes05<'a>(&'a mut BytesMut);
394
395#[cfg(feature = "tokio-02")]
396impl bytes_05::BufMut for Bytes05<'_> {
397    fn remaining_mut(&self) -> usize {
398        self.0.remaining_mut()
399    }
400    unsafe fn advance_mut(&mut self, cnt: usize) {
401        self.0.advance_mut(cnt)
402    }
403    fn bytes_mut(&mut self) -> &mut [MaybeUninit<u8>] {
404        unsafe { &mut *(self.0.chunk_mut() as *mut _ as *mut [MaybeUninit<u8>]) }
405    }
406}
407
408#[cfg(feature = "tokio-02")]
409impl<R> CombineRead<BufReader<R>, dyn tokio_02_dep::io::AsyncRead> for Bufferless
410where
411    R: tokio_02_dep::io::AsyncRead,
412{
413    fn poll_extend_buf(
414        &mut self,
415        cx: &mut Context<'_>,
416        read: Pin<&mut BufReader<R>>,
417    ) -> Poll<io::Result<usize>> {
418        let me = read.project();
419
420        if !me.buf.has_remaining_mut() {
421            me.buf.reserve(8 * 1024);
422        }
423        tokio_02_dep::io::AsyncRead::poll_read_buf(me.inner, cx, &mut Bytes05(me.buf))
424    }
425}
426
427#[cfg(feature = "tokio-03")]
428impl<R> CombineRead<BufReader<R>, dyn tokio_03_dep::io::AsyncRead> for Bufferless
429where
430    R: tokio_03_dep::io::AsyncRead,
431{
432    fn poll_extend_buf(
433        &mut self,
434        cx: &mut Context<'_>,
435        read: Pin<&mut BufReader<R>>,
436    ) -> Poll<io::Result<usize>> {
437        let me = read.project();
438
439        tokio_03_read_buf(cx, me.inner, me.buf)
440    }
441}
442
443#[cfg(feature = "tokio")]
444impl<R> CombineRead<BufReader<R>, dyn tokio_dep::io::AsyncRead> for Bufferless
445where
446    R: tokio_dep::io::AsyncRead,
447{
448    fn poll_extend_buf(
449        &mut self,
450        cx: &mut Context<'_>,
451        read: Pin<&mut BufReader<R>>,
452    ) -> Poll<io::Result<usize>> {
453        let me = read.project();
454
455        tokio_read_buf(me.inner, cx, me.buf)
456    }
457}
458
459#[cfg(feature = "futures-03")]
460impl<R> CombineAsyncRead<BufReader<R>> for Bufferless
461where
462    R: futures_io_03::AsyncRead,
463{
464    fn poll_extend_buf(
465        &mut self,
466        cx: &mut Context<'_>,
467        read: Pin<&mut BufReader<R>>,
468    ) -> Poll<io::Result<usize>> {
469        let me = read.project();
470
471        poll_extend_buf(me.buf, cx, me.inner)
472    }
473
474    fn extend_buf<'a>(
475        &'a mut self,
476        mut read: Pin<&'a mut BufReader<R>>,
477    ) -> ExtendBuf<'a, Self, BufReader<R>> {
478        let me = read.as_mut().project();
479
480        if !me.buf.has_remaining_mut() {
481            me.buf.reserve(8 * 1024);
482        }
483        // Copy of tokio's read_buf method (but it has to force initialize the buffer)
484        let bs = me.buf.chunk_mut();
485
486        for i in 0..bs.len() {
487            bs.write_byte(i, 0);
488        }
489        ExtendBuf { buffer: self, read }
490    }
491}
492
493#[cfg(feature = "futures-03")]
494fn poll_extend_buf<R>(
495    buf: &mut BytesMut,
496    cx: &mut Context<'_>,
497    read: Pin<&mut R>,
498) -> Poll<io::Result<usize>>
499where
500    R: futures_io_03::AsyncRead,
501{
502    // Copy of tokio's read_buf method (but it has to force initialize the buffer)
503    let n = {
504        let bs = buf.chunk_mut();
505        // preinit the buffer
506        for i in 0..bs.len() {
507            bs.write_byte(i, 0);
508        }
509
510        // Convert to `&mut [u8]`
511        // SAFETY: preinitialize the buffer
512        let bs = unsafe { &mut *(bs as *mut _ as *mut [u8]) };
513
514        let n = ready!(read.poll_read(cx, bs))?;
515        assert!(
516            n <= bs.len(),
517            "AsyncRead reported that it initialized more than the number of bytes in the buffer"
518        );
519        n
520    };
521    // SAFETY: the buffer was preinitialized
522    unsafe { buf.advance_mut(n) };
523    Poll::Ready(Ok(n))
524}
525
526#[cfg(feature = "tokio-02")]
527impl<R: tokio_02_dep::io::AsyncRead> tokio_02_dep::io::AsyncRead for BufReader<R> {
528    fn poll_read(
529        mut self: Pin<&mut Self>,
530        cx: &mut Context<'_>,
531        buf: &mut [u8],
532    ) -> Poll<io::Result<usize>> {
533        use tokio_02_dep::io::AsyncBufRead;
534
535        // If we don't have any buffered data and we're doing a massive read
536        // (larger than our internal buffer), bypass our internal buffer
537        // entirely.
538        if !self.buf.has_remaining_mut() && buf.len() >= self.buf.len() {
539            let res = ready!(self.as_mut().get_pin_mut().poll_read(cx, buf));
540            self.discard_buffer();
541            return Poll::Ready(res);
542        }
543        let mut rem = ready!(self.as_mut().poll_fill_buf(cx))?;
544        let nread = rem.read(buf)?;
545        self.consume(nread);
546        Poll::Ready(Ok(nread))
547    }
548
549    // we can't skip unconditionally because of the large buffer case in read.
550    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [MaybeUninit<u8>]) -> bool {
551        self.inner.prepare_uninitialized_buffer(buf)
552    }
553}
554
555#[cfg(feature = "tokio-02")]
556impl<R: tokio_02_dep::io::AsyncRead> tokio_02_dep::io::AsyncBufRead for BufReader<R> {
557    fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
558        let me = self.project();
559
560        // If we've reached the end of our internal buffer then we need to fetch
561        // some more data from the underlying reader.
562        // Branch using `>=` instead of the more correct `==`
563        // to tell the compiler that the pos..cap slice is always valid.
564
565        if me.buf.is_empty() {
566            ready!(me.inner.poll_read_buf(cx, &mut Bytes05(me.buf)))?;
567        }
568        Poll::Ready(Ok(&me.buf[..]))
569    }
570
571    fn consume(self: Pin<&mut Self>, amt: usize) {
572        let me = self.project();
573        me.buf.advance(amt);
574    }
575}
576
577#[cfg(feature = "tokio-02")]
578impl<R: tokio_02_dep::io::AsyncRead + tokio_02_dep::io::AsyncWrite> tokio_02_dep::io::AsyncWrite
579    for BufReader<R>
580{
581    fn poll_write(
582        self: Pin<&mut Self>,
583        cx: &mut Context<'_>,
584        buf: &[u8],
585    ) -> Poll<io::Result<usize>> {
586        self.get_pin_mut().poll_write(cx, buf)
587    }
588
589    fn poll_write_buf<B: bytes_05::Buf>(
590        self: Pin<&mut Self>,
591        cx: &mut Context<'_>,
592        buf: &mut B,
593    ) -> Poll<io::Result<usize>> {
594        self.get_pin_mut().poll_write_buf(cx, buf)
595    }
596
597    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
598        self.get_pin_mut().poll_flush(cx)
599    }
600
601    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
602        self.get_pin_mut().poll_shutdown(cx)
603    }
604}
605
606#[cfg(feature = "tokio-03")]
607impl<R: tokio_03_dep::io::AsyncRead> tokio_03_dep::io::AsyncRead for BufReader<R> {
608    fn poll_read(
609        mut self: Pin<&mut Self>,
610        cx: &mut Context<'_>,
611        buf: &mut tokio_03_dep::io::ReadBuf<'_>,
612    ) -> Poll<io::Result<()>> {
613        // If we don't have any buffered data and we're doing a massive read
614        // (larger than our internal buffer), bypass our internal buffer
615        // entirely.
616        if !self.buf.has_remaining_mut() && buf.remaining() >= self.buf.len() {
617            let res = ready!(self.as_mut().get_pin_mut().poll_read(cx, buf));
618            self.discard_buffer();
619            return Poll::Ready(res);
620        }
621        let rem = ready!(self.as_mut().poll_fill_buf(cx))?;
622        let amt = std::cmp::min(rem.len(), buf.remaining());
623        buf.put_slice(&rem[..amt]);
624        self.consume(amt);
625        Poll::Ready(Ok(()))
626    }
627}
628
629#[cfg(feature = "tokio-03")]
630impl<R: tokio_03_dep::io::AsyncRead> tokio_03_dep::io::AsyncBufRead for BufReader<R> {
631    fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
632        let me = self.project();
633
634        // If we've reached the end of our internal buffer then we need to fetch
635        // some more data from the underlying reader.
636        if me.buf.is_empty() {
637            ready!(tokio_03_read_buf(cx, me.inner, me.buf))?;
638        }
639        Poll::Ready(Ok(&me.buf[..]))
640    }
641
642    fn consume(self: Pin<&mut Self>, amt: usize) {
643        let me = self.project();
644        me.buf.advance(amt);
645    }
646}
647
648#[cfg(feature = "tokio-03")]
649impl<R: tokio_03_dep::io::AsyncRead + tokio_03_dep::io::AsyncWrite> tokio_03_dep::io::AsyncWrite
650    for BufReader<R>
651{
652    fn poll_write(
653        self: Pin<&mut Self>,
654        cx: &mut Context<'_>,
655        buf: &[u8],
656    ) -> Poll<io::Result<usize>> {
657        self.get_pin_mut().poll_write(cx, buf)
658    }
659
660    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
661        self.get_pin_mut().poll_flush(cx)
662    }
663
664    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
665        self.get_pin_mut().poll_shutdown(cx)
666    }
667}
668
669#[cfg(feature = "tokio")]
670impl<R: tokio_dep::io::AsyncRead> tokio_dep::io::AsyncRead for BufReader<R> {
671    fn poll_read(
672        mut self: Pin<&mut Self>,
673        cx: &mut Context<'_>,
674        buf: &mut tokio_dep::io::ReadBuf<'_>,
675    ) -> Poll<io::Result<()>> {
676        // If we don't have any buffered data and we're doing a massive read
677        // (larger than our internal buffer), bypass our internal buffer
678        // entirely.
679        if !self.buf.has_remaining_mut() && buf.remaining() >= self.buf.len() {
680            let res = ready!(self.as_mut().get_pin_mut().poll_read(cx, buf));
681            self.discard_buffer();
682            return Poll::Ready(res);
683        }
684        let rem = ready!(self.as_mut().poll_fill_buf(cx))?;
685        let amt = std::cmp::min(rem.len(), buf.remaining());
686        buf.put_slice(&rem[..amt]);
687        self.consume(amt);
688        Poll::Ready(Ok(()))
689    }
690}
691
692#[cfg(feature = "tokio")]
693impl<R: tokio_dep::io::AsyncRead> tokio_dep::io::AsyncBufRead for BufReader<R> {
694    fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
695        let me = self.project();
696
697        // If we've reached the end of our internal buffer then we need to fetch
698        // some more data from the underlying reader.
699        if me.buf.is_empty() {
700            ready!(tokio_read_buf(me.inner, cx, me.buf))?;
701        }
702        Poll::Ready(Ok(&me.buf[..]))
703    }
704
705    fn consume(self: Pin<&mut Self>, amt: usize) {
706        let me = self.project();
707        me.buf.advance(amt);
708    }
709}
710
711#[cfg(feature = "tokio")]
712impl<R: tokio_dep::io::AsyncRead + tokio_dep::io::AsyncWrite> tokio_dep::io::AsyncWrite
713    for BufReader<R>
714{
715    fn poll_write(
716        self: Pin<&mut Self>,
717        cx: &mut Context<'_>,
718        buf: &[u8],
719    ) -> Poll<io::Result<usize>> {
720        self.get_pin_mut().poll_write(cx, buf)
721    }
722
723    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
724        self.get_pin_mut().poll_flush(cx)
725    }
726
727    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
728        self.get_pin_mut().poll_shutdown(cx)
729    }
730}
731
732impl<R: Read> Read for BufReader<R> {
733    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
734        // If we don't have any buffered data and we're doing a massive read
735        // (larger than our internal buffer), bypass our internal buffer
736        // entirely.
737        if !self.buf.has_remaining_mut() && buf.len() >= self.buf.len() {
738            let res = self.read(buf);
739            self.buf.clear();
740            return res;
741        }
742        let nread = {
743            let mut rem = self.fill_buf()?;
744            rem.read(buf)?
745        };
746        self.consume(nread);
747        Ok(nread)
748    }
749}
750
751impl<R: Read> BufRead for BufReader<R> {
752    fn fill_buf(&mut self) -> io::Result<&[u8]> {
753        // If we've reached the end of our internal buffer then we need to fetch
754        // some more data from the underlying reader.
755        // Branch using `>=` instead of the more correct `==`
756        // to tell the compiler that the pos..cap slice is always valid.
757
758        if self.buf.is_empty() {
759            Bufferless.extend_buf_sync(self)?;
760        }
761        Ok(&self.buf[..])
762    }
763
764    fn consume(&mut self, amt: usize) {
765        self.buf.advance(amt);
766    }
767}
768
769#[cfg(test)]
770#[cfg(feature = "tokio-02")]
771mod tests {
772    use super::{BufReader, Bufferless, CombineRead};
773
774    use std::{io, pin::Pin};
775
776    use {
777        bytes_05::BytesMut,
778        tokio_02_dep::{
779            self as tokio,
780            io::{AsyncRead, AsyncReadExt},
781        },
782    };
783
784    impl<R: AsyncRead> BufReader<R> {
785        async fn extend_buf_tokio_02(mut self: Pin<&mut Self>) -> io::Result<usize> {
786            crate::future_ext::poll_fn(|cx| Bufferless.poll_extend_buf(cx, self.as_mut())).await
787        }
788    }
789
790    #[tokio::test]
791    async fn buf_reader() {
792        let mut read = BufReader::with_capacity(3, &[1u8, 2, 3, 4, 5, 6, 7, 8, 9, 0][..]);
793
794        let mut buf = [0u8; 3];
795        read.read(&mut buf).await.unwrap();
796        assert_eq!(buf, [1, 2, 3]);
797
798        let mut buf = [0u8; 3];
799        read.read(&mut buf).await.unwrap();
800        assert_eq!(buf, [4, 5, 6]);
801
802        let mut buf = [0u8; 3];
803        read.read(&mut buf).await.unwrap();
804        assert_eq!(buf, [7, 8, 9]);
805
806        let mut buf = [1u8; 3];
807        read.read(&mut buf).await.unwrap();
808        assert_eq!(buf, [0, 1, 1]);
809    }
810
811    #[tokio::test]
812    async fn buf_reader_buf() {
813        let mut read = BufReader::with_capacity(3, &[1u8, 2, 3, 4, 5, 6, 7, 8, 9, 0][..]);
814
815        let mut buf = BytesMut::with_capacity(3);
816        read.read_buf(&mut buf).await.unwrap();
817        assert_eq!(&buf[..], [1, 2, 3]);
818
819        read.read_buf(&mut buf).await.unwrap();
820        assert_eq!(&buf[..], [1, 2, 3, 4, 5, 6, 7, 8, 9, 0]);
821    }
822
823    #[tokio::test]
824    async fn buf_reader_extend_buf() {
825        let read = BufReader::with_capacity(3, &[1u8, 2, 3, 4, 5, 6, 7, 8, 9, 0][..]);
826        futures_03_dep::pin_mut!(read);
827
828        assert_eq!(read.as_mut().extend_buf_tokio_02().await.unwrap(), 3);
829        assert_eq!(read.buffer(), [1, 2, 3]);
830
831        assert_eq!(read.as_mut().extend_buf_tokio_02().await.unwrap(), 7);
832        assert_eq!(read.buffer(), [1, 2, 3, 4, 5, 6, 7, 8, 9, 0]);
833    }
834}
835
836#[cfg(test)]
837#[cfg(feature = "tokio")]
838mod tests_tokio_1 {
839    use super::{BufReader, Bufferless, CombineRead};
840
841    use std::{io, pin::Pin};
842
843    use {
844        bytes::BytesMut,
845        tokio_dep::{
846            self as tokio,
847            io::{AsyncRead, AsyncReadExt},
848        },
849    };
850
851    impl<R: AsyncRead> BufReader<R> {
852        async fn extend_buf_tokio(mut self: Pin<&mut Self>) -> io::Result<usize> {
853            crate::future_ext::poll_fn(|cx| Bufferless.poll_extend_buf(cx, self.as_mut())).await
854        }
855    }
856
857    #[tokio::test]
858    async fn buf_reader() {
859        let mut read = BufReader::with_capacity(3, &[1u8, 2, 3, 4, 5, 6, 7, 8, 9, 0][..]);
860
861        let mut buf = [0u8; 3];
862        read.read(&mut buf).await.unwrap();
863        assert_eq!(buf, [1, 2, 3]);
864
865        let mut buf = [0u8; 3];
866        read.read(&mut buf).await.unwrap();
867        assert_eq!(buf, [4, 5, 6]);
868
869        let mut buf = [0u8; 3];
870        read.read(&mut buf).await.unwrap();
871        assert_eq!(buf, [7, 8, 9]);
872
873        let mut buf = [1u8; 3];
874        read.read(&mut buf).await.unwrap();
875        assert_eq!(buf, [0, 1, 1]);
876    }
877
878    #[tokio::test]
879    async fn buf_reader_buf() {
880        let mut read = BufReader::with_capacity(3, &[1u8, 2, 3, 4, 5, 6, 7, 8, 9, 0][..]);
881
882        let mut buf = BytesMut::with_capacity(3);
883        read.read_buf(&mut buf).await.unwrap();
884        assert_eq!(&buf[..], [1, 2, 3]);
885
886        read.read_buf(&mut buf).await.unwrap();
887        assert_eq!(&buf[..], [1, 2, 3, 4, 5, 6, 7, 8, 9, 0]);
888    }
889
890    #[tokio::test]
891    async fn buf_reader_extend_buf() {
892        let read = BufReader::with_capacity(3, &[1u8, 2, 3, 4, 5, 6, 7, 8, 9, 0][..]);
893        futures_03_dep::pin_mut!(read);
894
895        assert_eq!(read.as_mut().extend_buf_tokio().await.unwrap(), 3);
896        assert_eq!(read.buffer(), [1, 2, 3]);
897
898        assert_eq!(read.as_mut().extend_buf_tokio().await.unwrap(), 7);
899        assert_eq!(read.buffer(), [1, 2, 3, 4, 5, 6, 7, 8, 9, 0]);
900    }
901}
902
903#[cfg(test)]
904mod tests_sync {
905    use super::{BufReader, Bufferless, CombineSyncRead};
906
907    use std::io::Read;
908
909    #[test]
910    #[allow(clippy::unused_io_amount)]
911    fn buf_reader() {
912        let mut read = BufReader::with_capacity(3, &[1u8, 2, 3, 4, 5, 6, 7, 8, 9, 0][..]);
913
914        let mut buf = [0u8; 3];
915        read.read(&mut buf).unwrap();
916        assert_eq!(buf, [1, 2, 3]);
917
918        let mut buf = [0u8; 3];
919        read.read(&mut buf).unwrap();
920        assert_eq!(buf, [4, 5, 6]);
921
922        let mut buf = [0u8; 3];
923        read.read(&mut buf).unwrap();
924        assert_eq!(buf, [7, 8, 9]);
925
926        let mut buf = [1u8; 3];
927        read.read(&mut buf).unwrap();
928        assert_eq!(buf, [0, 1, 1]);
929    }
930
931    #[test]
932    fn buf_reader_extend_buf() {
933        let mut read = BufReader::with_capacity(3, &[1u8, 2, 3, 4, 5, 6, 7, 8, 9, 0][..]);
934
935        assert_eq!(Bufferless.extend_buf_sync(&mut read).unwrap(), 3);
936        assert_eq!(read.buffer(), [1, 2, 3]);
937
938        assert_eq!(Bufferless.extend_buf_sync(&mut read).unwrap(), 7);
939        assert_eq!(read.buffer(), [1, 2, 3, 4, 5, 6, 7, 8, 9, 0]);
940    }
941}
combine/stream/buf_reader.rs

combine/stream/
buf_reader.rs
