use std::cmp;
use bytes::Bytes;
use crate::error;
use crate::solicit::end_stream::EndStream;
use crate::solicit::header::Headers;
use crate::solicit::session::StreamState;
use crate::solicit::window_size::NonNegativeWindowSize;
use crate::solicit::window_size::WindowSize;
use super::types::Types;
use super::stream_queue::StreamQueue;
use super::window_size;
use crate::common::stream_handler::StreamHandlerInternal;
use crate::data_or_headers::DataOrHeaders;
use crate::data_or_headers_with_flag::DataOrHeadersWithFlag;
use crate::ErrorCode;
pub enum HttpStreamCommand {
Headers(Headers, EndStream),
Data(Bytes, EndStream),
Rst(ErrorCode),
}
impl HttpStreamCommand {
pub fn from(part: DataOrHeadersWithFlag) -> HttpStreamCommand {
let end_stream = match part.last {
true => EndStream::Yes,
false => EndStream::No,
};
match part.content {
DataOrHeaders::Data(data) => HttpStreamCommand::Data(data, end_stream),
DataOrHeaders::Headers(headers) => HttpStreamCommand::Headers(headers, end_stream),
}
}
}
#[must_use]
pub struct DroppedData {
pub size: usize,
}
#[derive(Debug, PartialEq, Eq, Clone)]
pub struct HttpStreamStateSnapshot {
pub state: StreamState,
pub out_window_size: i32,
pub in_window_size: i32,
pub pump_out_window_size: isize,
pub queued_out_data_size: usize,
pub out_data_size: usize,
}
#[derive(Eq, PartialEq, Copy, Clone, Debug)]
pub enum InMessageStage {
Initial,
AfterInitialHeaders,
AfterTrailingHeaders,
}
pub(crate) struct HttpStreamCommon<T: Types> {
pub specific: T::HttpStreamSpecific,
pub state: StreamState,
pub out_window_size: WindowSize,
pub in_window_size: NonNegativeWindowSize,
pub outgoing: StreamQueue,
pub peer_tx: Option<T::StreamHandlerHolder>,
pub pump_out_window: window_size::StreamOutWindowSender,
pub in_rem_content_length: Option<u64>,
pub in_message_stage: InMessageStage,
}
impl<T: Types> HttpStreamCommon<T> {
pub fn new(
in_window_size: u32,
out_window_size: u32,
pump_out_window: window_size::StreamOutWindowSender,
in_rem_content_length: Option<u64>,
in_message_stage: InMessageStage,
specific: T::HttpStreamSpecific,
) -> HttpStreamCommon<T> {
HttpStreamCommon {
specific,
state: StreamState::Open,
in_window_size: NonNegativeWindowSize::new(in_window_size as i32),
out_window_size: WindowSize::new(out_window_size as i32),
outgoing: StreamQueue::new(),
peer_tx: None,
pump_out_window,
in_rem_content_length,
in_message_stage,
}
}
pub fn snapshot(&self) -> HttpStreamStateSnapshot {
HttpStreamStateSnapshot {
state: self.state,
out_window_size: self.out_window_size.size(),
in_window_size: self.in_window_size.size(),
pump_out_window_size: self.pump_out_window.get(),
queued_out_data_size: self.outgoing.data_size(),
out_data_size: self.outgoing.data_size(),
}
}
pub fn close_local(&mut self) {
trace!("close local");
self.state = match self.state {
StreamState::Closed | StreamState::HalfClosedRemote => StreamState::Closed,
_ => StreamState::HalfClosedLocal,
};
}
pub fn close_remote(&mut self) {
trace!("close remote");
self.state = match self.state {
StreamState::Closed | StreamState::HalfClosedLocal => StreamState::Closed,
_ => StreamState::HalfClosedRemote,
};
}
pub fn conn_died(mut self, error: error::Error) {
if let Some(handler) = self.peer_tx.take() {
drop(handler.error(error));
}
}
pub fn is_writable(&self) -> bool {
match self.outgoing.front() {
Some(front) => match front {
DataOrHeaders::Headers(..) => true,
DataOrHeaders::Data(data) => data.len() == 0 || self.out_window_size.size() > 0,
},
None => {
if let Some(_error_code) = self.outgoing.end() {
if !self.state.is_closed_local() {
return true;
}
};
false
}
}
}
#[cfg(debug_assertions)]
pub fn pop_outg(&mut self, conn_out_window_size: &mut WindowSize) -> Option<HttpStreamCommand> {
let writable = self.is_writable();
let conn_out_window_size_before = conn_out_window_size.size();
let command = self.pop_outg_impl(conn_out_window_size);
if command.is_some() {
assert!(writable);
} else {
assert!(
!writable || conn_out_window_size_before == 0,
"popped nothing but writable: {}, conn out window size before: {}, {:?}",
writable,
conn_out_window_size_before,
self.snapshot()
);
}
command
}
#[cfg(not(debug_assertions))]
pub fn pop_outg(&mut self, conn_out_window_size: &mut WindowSize) -> Option<HttpStreamCommand> {
self.pop_outg_impl(conn_out_window_size)
}
fn pop_outg_impl(
&mut self,
conn_out_window_size: &mut WindowSize,
) -> Option<HttpStreamCommand> {
if self.outgoing.is_empty() {
return if let Some(error_code) = self.outgoing.end() {
if self.state.is_closed_local() {
None
} else {
self.close_local();
Some(match error_code {
ErrorCode::NoError => HttpStreamCommand::Data(Bytes::new(), EndStream::Yes),
error_code => HttpStreamCommand::Rst(error_code),
})
}
} else {
None
};
}
let pop_headers = if let &DataOrHeaders::Headers(..) = self.outgoing.front().unwrap() {
true
} else {
false
};
if pop_headers {
let r = self.outgoing.pop_front().unwrap();
let last = self.outgoing.end() == Some(ErrorCode::NoError);
if last {
self.close_local();
}
return Some(HttpStreamCommand::from(DataOrHeadersWithFlag {
content: r,
last: last,
}));
}
if self.out_window_size.size() <= 0 || conn_out_window_size.size() <= 0 {
return None;
}
let mut data = if let Some(DataOrHeaders::Data(data)) = self.outgoing.pop_front() {
data
} else {
unreachable!()
};
let max_window = cmp::min(self.out_window_size.size(), conn_out_window_size.size());
if data.len() as usize > max_window as usize {
trace!("truncating data of len {} to {}", data.len(), max_window);
let size = max_window as usize;
let rem = data.split_off(size);
self.outgoing.push_front(DataOrHeaders::Data(rem));
};
self.out_window_size
.try_decrease_to_non_negative(data.len() as i32)
.unwrap();
conn_out_window_size
.try_decrease_to_non_negative(data.len() as i32)
.unwrap();
let last = self.outgoing.end() == Some(ErrorCode::NoError);
if last {
self.close_local();
}
Some(HttpStreamCommand::from(DataOrHeadersWithFlag {
content: DataOrHeaders::Data(data),
last: last,
}))
}
pub fn data_recvd(&mut self, data: Bytes, last: bool) {
if let Some(ref mut response_handler) = self.peer_tx {
drop(response_handler.data_frame(data, last));
}
}
pub fn rst_recvd(&mut self, error_code: ErrorCode) -> DroppedData {
if let Some(response_handler) = self.peer_tx.take() {
drop(response_handler.rst(error_code));
}
DroppedData {
size: self.outgoing.data_size(),
}
}
pub fn goaway_recvd(&mut self, _raw_error_code: u32) {
if let Some(response_handler) = self.peer_tx.take() {
drop(response_handler.error(error::Error::GoawayReceived));
}
}
}
pub(crate) trait HttpStreamDataSpecific: Send + 'static {}
pub(crate) trait HttpStreamData {
type Types: Types;
}