use bytes::{Buf, BufMut, BytesMut}; use std::pin::Pin; use std::task::{Context, Poll}; use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; use tokio_util::io::poll_write_buf;
use std::io::{self, Cursor};
// A macro to get around a method needing to borrow &mut self
macro_rules! limited_write_buf {
($self:expr) => {{ let limit = $self.max_frame_size() + frame::HEADER_LEN;
$self.buf.get_mut().limit(limit)
}};
}
/// Initialize the connection with this amount of write buffer. /// /// The minimum MAX_FRAME_SIZE is 16kb, so always be able to send a HEADERS /// frame that big. const DEFAULT_BUFFER_CAPACITY: usize = 16 * 1_024;
/// Chain payloads bigger than this when vectored I/O is enabled. The remote /// will never advertise a max frame size less than this (well, the spec says /// the max frame size can't be less than 16kb, so not even close). const CHAIN_THRESHOLD: usize = 256;
/// Chain payloads bigger than this when vectored I/O is **not** enabled. /// A larger value in this scenario will reduce the number of small and /// fragmented data being sent, and hereby improve the throughput. const CHAIN_THRESHOLD_WITHOUT_VECTORED_IO: usize = 1024;
/// Returns `Ready` when `send` is able to accept a frame /// /// Calling this function may result in the current contents of the buffer /// to be flushed to `T`. pubfn poll_ready(&mutself, cx: &mut Context) -> Poll<io::Result<()>> { if !self.encoder.has_capacity() { // Try flushing
ready!(self.flush(cx))?;
if !self.encoder.has_capacity() { return Poll::Pending;
}
}
Poll::Ready(Ok(()))
}
/// Buffer a frame. /// /// `poll_ready` must be called first to ensure that a frame may be /// accepted. pubfn buffer(&mutself, item: Frame<B>) -> Result<(), UserError> { self.encoder.buffer(item)
}
/// Flush buffered data to the wire pubfn flush(&mutself, cx: &mut Context) -> Poll<io::Result<()>> { let span = tracing::trace_span!("FramedWrite::flush"); let _e = span.enter();
// The data frame has been written, so unset it matchself.next.take() {
Some(Next::Data(frame)) => { self.last_data_frame = Some(frame);
debug_assert!(self.is_empty());
ControlFlow::Break
}
Some(Next::Continuation(frame)) => { // Buffer the continuation frame, then try to write again letmut buf = limited_write_buf!(self); iflet Some(continuation) = frame.encode(&mut buf) { self.next = Some(Next::Continuation(continuation));
}
ControlFlow::Continue
}
None => ControlFlow::Break,
}
}
fn buffer(&mutself, item: Frame<B>) -> Result<(), UserError> { // Ensure that we have enough capacity to accept the write.
assert!(self.has_capacity()); let span = tracing::trace_span!("FramedWrite::buffer", frame = ?item); let _e = span.enter();
tracing::debug!(frame = ?item, "send");
match item {
Frame::Data(mut v) => { // Ensure that the payload is not greater than the max frame. let len = v.payload().remaining();
if len > self.max_frame_size() { return Err(PayloadTooBig);
}
if len >= self.chain_threshold { let head = v.head();
// Encode the frame head to the buffer
head.encode(len, self.buf.get_mut());
impl<T, B> FramedWrite<T, B> { /// Returns the max frame size that can be sent pubfn max_frame_size(&self) -> usize { self.encoder.max_frame_size()
}
/// Set the peer's max frame size. pubfn set_max_frame_size(&mutself, val: usize) {
assert!(val <= frame::MAX_MAX_FRAME_SIZE as usize); self.encoder.max_frame_size = val as FrameSize;
}
/// Set the peer's header table size. pubfn set_header_table_size(&mutself, val: usize) { self.encoder.hpack.update_max_size(val);
}
/// Retrieve the last data frame that has been sent pubfn take_last_data_frame(&mutself) -> Option<frame::Data<B>> { self.encoder.last_data_frame.take()
}
pubfn get_mut(&mutself) -> &mut T {
&mutself.inner
}
}
Die Informationen auf dieser Webseite wurden
nach bestem Wissen sorgfältig zusammengestellt. Es wird jedoch weder Vollständigkeit, noch Richtigkeit,
noch Qualität der bereit gestellten Informationen zugesichert.
Bemerkung:
Die farbliche Syntaxdarstellung und die Messung sind noch experimentell.