Method poll_next

opensuse-proxy-cache/src/main.rs:863–941

Required method

(mut self: Pin<&mut Self>, ctx: &mut Context<'_>)

    // Required method
    fn poll_next(mut self: Pin<&mut Self>, ctx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        // We have hit the buffer full limit, so drain some first. This only happens if the
        // buffer is misaligned as we proceed through the download, and we managed to
        // build up a ton of data that didn't align on frame boundaries.
        if self.as_mut().project().buffer.len() >= BUFFER_NET_LIMIT {
            let buf_to_send = self.as_mut().project().buffer.split_to(BUFFER_MIN_XMIT);
            let buf = Bytes::from(buf_to_send);
            return Poll::Ready(Some(Ok(buf)));
        }

        // Buffer must not have more than BUFFER_NET_LIMIT in it now, so we can read a bit more.
        loop {
            match self.as_mut().project().dlos_reader.poll_next(ctx) {
                Poll::Ready(Some(Ok(buf))) => {
                    // Buffer the content. We don't want to spend too much time buffering, so we
                    // only buffer a small amount at a time.
                    self.as_mut().project().buffer.extend_from_slice(&buf);
                    // We want at least this much data in the buffer to proceed, this is divisible
                    // by frames. Leftover content will be dealt with next Poll.
                    if self.as_mut().project().buffer.len() >= BUFFER_NET_LIMIT {
                        let buf_to_send = self.as_mut().project().buffer.split_to(BUFFER_MIN_XMIT);

                        let buf = Bytes::from(buf_to_send);
                        break Poll::Ready(Some(Ok(buf)));
                    } else {
                        // Fill more!!!
                        continue;
                    }
                }
                // Error
                Poll::Ready(Some(Err(e))) => break Poll::Ready(Some(Err(e))),
                // Indicates termination of the stream. We are DONE!!!
                Poll::Ready(None) => {
                    if self.as_mut().project().buffer.is_empty() {
                        break Poll::Ready(None);
                    } else {
                        // Send out all remaining bytes, yolo.
                        let buf = self.as_mut().project().buffer.split().freeze();
                        break Poll::Ready(Some(Ok(buf)));
                    }
                }
                // Pending on more bytes from upstream.
                Poll::Pending => {
                    // TODO: Does this break things?
                    // WAS - break Poll::Pending
                    //
                    // Okay, we're pending on upstream bytes, but do we have anything to send? This way
                    // we don't block out the reader. This can happen if during a tight Poll::Ready
                    // loop, we ended up in a Pending from upstream, but we don't want to penalise our downstream.
                    //
                    // But at the same time, we want to ensure we have *some* buffer to send, so we aim
                    // to have at least min batch xmit bytes.
                    let buf_len = self.as_mut().project().buffer.len();
                    // if buf_len >= BUFFER_MIN_BATCH_XMIT {
                    if buf_len >= BUFFER_MIN_XMIT {
                        /*
                        let excess = buf_len % BUFFER_MIN_XMIT;
                        let to_send = buf_len - excess;