Line data    Source code

       1              : use pin_project_lite::pin_project;
       2              : use std::io::Read;
       3              : use std::pin::Pin;
       4              : use std::{io, task};
       5              : use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
       6              : 
       7              : pin_project! {
       8              :     /// This stream tracks all writes and calls user provided
       9              :     /// callback when the underlying stream is flushed.
      10              :     pub struct MeasuredStream<S, R, W> {
      11              :         #[pin]
      12              :         stream: S,
      13              :         write_count: usize,
      14              :         inc_read_count: R,
      15              :         inc_write_count: W,
      16              :     }
      17              : }
      18              : 
      19              : impl<S, R, W> MeasuredStream<S, R, W> {
      20         3396 :     pub fn new(stream: S, inc_read_count: R, inc_write_count: W) -> Self {
      21         3396 :         Self {
      22         3396 :             stream,
      23         3396 :             write_count: 0,
      24         3396 :             inc_read_count,
      25         3396 :             inc_write_count,
      26         3396 :         }
      27         3396 :     }
      28              : }
      29              : 
      30              : impl<S: AsyncRead + Unpin, R: FnMut(usize), W> AsyncRead for MeasuredStream<S, R, W> {
      31      8717941 :     fn poll_read(
      32      8717941 :         self: Pin<&mut Self>,
      33      8717941 :         context: &mut task::Context<'_>,
      34      8717941 :         buf: &mut ReadBuf<'_>,
      35      8717941 :     ) -> task::Poll<io::Result<()>> {
      36      8717941 :         let this = self.project();
      37      8717941 :         let filled = buf.filled().len();
      38      8717941 :         this.stream.poll_read(context, buf).map_ok(|()| {
      39      2672980 :             let cnt = buf.filled().len() - filled;
      40      2672980 :             // Increment the read count.
      41      2672980 :             (this.inc_read_count)(cnt);
      42      8717941 :         })
      43      8717941 :     }
      44              : }
      45              : 
      46              : impl<S: AsyncWrite + Unpin, R, W: FnMut(usize)> AsyncWrite for MeasuredStream<S, R, W> {
      47      2319344 :     fn poll_write(
      48      2319344 :         self: Pin<&mut Self>,
      49      2319344 :         context: &mut task::Context<'_>,
      50      2319344 :         buf: &[u8],
      51      2319344 :     ) -> task::Poll<io::Result<usize>> {
      52      2319344 :         let this = self.project();
      53      2319344 :         this.stream.poll_write(context, buf).map_ok(|cnt| {
      54      2296706 :             // Increment the write count.
      55      2296706 :             *this.write_count += cnt;
      56      2296706 :             cnt
      57      2319344 :         })
      58      2319344 :     }
      59              : 
      60      2297674 :     fn poll_flush(
      61      2297674 :         self: Pin<&mut Self>,
      62      2297674 :         context: &mut task::Context<'_>,
      63      2297674 :     ) -> task::Poll<io::Result<()>> {
      64      2297674 :         let this = self.project();
      65      2297674 :         this.stream.poll_flush(context).map_ok(|()| {
      66      2297674 :             // Call the user provided callback and reset the write count.
      67      2297674 :             (this.inc_write_count)(*this.write_count);
      68      2297674 :             *this.write_count = 0;
      69      2297674 :         })
      70      2297674 :     }
      71              : 
      72         2709 :     fn poll_shutdown(
      73         2709 :         self: Pin<&mut Self>,
      74         2709 :         context: &mut task::Context<'_>,
      75         2709 :     ) -> task::Poll<io::Result<()>> {
      76         2709 :         self.project().stream.poll_shutdown(context)
      77         2709 :     }
      78              : }
      79              : 
      80              : /// Wrapper for a reader that counts bytes read.
      81              : ///
      82              : /// Similar to MeasuredStream but it's one way and it's sync
      83              : pub struct MeasuredReader<R: Read> {
      84              :     inner: R,
      85              :     byte_count: usize,
      86              : }
      87              : 
      88              : impl<R: Read> MeasuredReader<R> {
      89          571 :     pub fn new(reader: R) -> Self {
      90          571 :         Self {
      91          571 :             inner: reader,
      92          571 :             byte_count: 0,
      93          571 :         }
      94          571 :     }
      95              : 
      96          571 :     pub fn get_byte_count(&self) -> usize {
      97          571 :         self.byte_count
      98          571 :     }
      99              : }
     100              : 
     101              : impl<R: Read> Read for MeasuredReader<R> {
     102       859372 :     fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
     103       859372 :         let result = self.inner.read(buf);
     104       859372 :         if let Ok(n_bytes) = result {
     105       859372 :             self.byte_count += n_bytes
     106            0 :         }
     107       859372 :         result
     108       859372 :     }
     109              : }