use std::pin::Pin;
use std::task::{Context, Poll};
use bytes::{BufMut, Bytes, BytesMut};
use http_body::{Body, Frame};
use http_body_util::Full;
use tokio::fs::File;
use tokio::io::{AsyncRead, ReadBuf};
use crate::error::StaticError;
pub(crate) const FILE_CHUNK_SIZE: usize = 65_536;
pub enum ResponseBody {
Buffered(Full<Bytes>),
Streamed(FileBody),
}
impl Body for ResponseBody {
type Data = Bytes;
type Error = StaticError;
fn poll_frame(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
match self.get_mut() {
ResponseBody::Buffered(body) => match Pin::new(body).poll_frame(cx) {
Poll::Ready(Some(Ok(frame))) => Poll::Ready(Some(Ok(frame))),
Poll::Ready(Some(Err(never))) => match never {},
Poll::Ready(None) => Poll::Ready(None),
Poll::Pending => Poll::Pending,
},
ResponseBody::Streamed(body) => Pin::new(body).poll_frame(cx),
}
}
}
pub struct FileBody {
file: File,
buf: BytesMut,
}
impl FileBody {
pub(crate) fn new(file: File) -> Self {
FileBody {
file,
buf: BytesMut::new(),
}
}
}
impl Body for FileBody {
type Data = Bytes;
type Error = StaticError;
fn poll_frame(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
let this = self.get_mut();
if this.buf.capacity() - this.buf.len() < FILE_CHUNK_SIZE {
this.buf.reserve(FILE_CHUNK_SIZE);
}
let mut read_buf = ReadBuf::uninit(this.buf.spare_capacity_mut());
let file = Pin::new(&mut this.file);
match file.poll_read(cx, &mut read_buf) {
Poll::Ready(Ok(())) => {
let n = read_buf.filled().len();
if n == 0 {
return Poll::Ready(None);
}
unsafe { this.buf.advance_mut(n) };
let chunk = this.buf.split_to(n).freeze();
Poll::Ready(Some(Ok(Frame::data(chunk))))
}
Poll::Ready(Err(e)) => Poll::Ready(Some(Err(StaticError::Io(e)))),
Poll::Pending => Poll::Pending,
}
}
}