use alloc::vec::Vec;
use crate::{
body::{Decoder, Encoder, Step, push_framing},
buffer::Buffer,
error::HttpError,
head::{
Asked, HeadKind, HeadScanner, Headers, ResponseHead, ResponseInfo, Scan,
interpret_response, switches_to_offered,
},
limits::{Limits, MAX_PIPELINE_DEPTH},
message::{OutBody, Request, check_trailers, push_last_chunk_with_trailers},
method::Method,
scan::find_byte,
server::Progress,
};
const REFUSED: HttpError = HttpError::InvalidState(
"the server answered with a final status instead of 100 Continue; \
the body must not be sent",
);
#[non_exhaustive]
#[derive(Debug)]
pub enum ClientEvent<'a> {
Response(ResponseHead<'a>),
Informational(ResponseHead<'a>),
Continue,
Body(&'a [u8]),
Trailers(Headers<'a>),
ResponseEnd,
NeedInput,
Idle,
Upgraded,
Closed,
}
#[non_exhaustive]
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)]
pub struct ClientLeniency {
pub obs_fold_in_responses: bool,
}
impl ClientLeniency {
#[must_use]
pub const fn new() -> Self {
Self {
obs_fold_in_responses: false,
}
}
#[must_use]
pub const fn with_obs_fold_in_responses(mut self, on: bool) -> Self {
self.obs_fold_in_responses = on;
self
}
}
#[derive(Clone, Copy, Debug)]
struct Unfold {
scan: usize,
line_start: usize,
past_status: bool,
done: bool,
}
impl Unfold {
const fn new() -> Self {
Self {
scan: 0,
line_start: 0,
past_status: false,
done: false,
}
}
fn run(&mut self, msg: &mut [u8], eof: bool) -> usize {
if self.done {
return msg.len();
}
loop {
let Some(off) = msg.get(self.scan..).and_then(|rest| find_byte(b'\n', rest)) else {
self.scan = msg.len();
return msg.len();
};
let lf = self.scan + off;
if !self.past_status {
self.past_status = true;
self.line_start = lf + 1;
self.scan = lf + 1;
continue;
}
let line = msg.get(self.line_start..lf).unwrap_or(&[]);
if line.is_empty() || line == b"\r" {
self.done = true;
return msg.len();
}
let crlf = lf > 0 && msg.get(lf - 1) == Some(&b'\r');
match msg.get(lf + 1) {
None if eof => {
self.scan = msg.len();
return msg.len();
}
None => {
self.scan = lf;
return lf;
}
Some(&b) if crlf && (b == b' ' || b == b'\t') => {
msg[lf - 1] = b' ';
msg[lf] = b' ';
self.scan = lf + 1;
}
Some(_) => {
self.line_start = lf + 1;
self.scan = lf + 1;
}
}
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum State {
Idle,
Head,
Interim,
Body,
Complete,
Upgraded,
Closed,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum ReqWrite {
Streaming(Encoder),
Sent,
}
pub struct ClientConn {
buf: Buffer,
limits: Limits,
scanner: HeadScanner,
trailers: HeadScanner,
decoder: Decoder,
info: ResponseInfo,
asked: Asked,
upgrade_asked: bool,
upgrade_offer: Vec<u8>,
state: State,
req: ReqWrite,
keep_alive: bool,
interim_bytes: usize,
interim_count: u16,
report_interim: bool,
report_trailers: bool,
waiting_continue: bool,
body_refused: bool,
req_chunked: bool,
idempotent: bool,
queue: [Queued; MAX_PIPELINE_DEPTH as usize],
queue_start: usize,
queued: usize,
leniency: ClientLeniency,
unfold: Unfold,
}
#[derive(Clone, Copy, Debug)]
struct Queued {
asked: Asked,
close: bool,
}
fn write_request(request: &Request, out: &mut Vec<u8>) -> Result<ReqWrite, HttpError> {
let method = request.method_kind;
let expects_body = matches!(method, Method::Post | Method::Put | Method::Patch);
let (mut encoder, declared) = match &request.body {
OutBody::Full(b) if b.is_empty() && !expects_body => (Encoder::Discard, None),
OutBody::Full(b) => (
Encoder::Length {
remaining: b.len() as u64,
},
Some(b.len() as u64),
),
OutBody::Length(n) => (Encoder::Length { remaining: *n }, Some(*n)),
OutBody::Chunked => (Encoder::Chunked, None),
};
out.extend_from_slice(request.method().as_bytes());
out.push(b' ');
out.extend_from_slice(request.target().as_bytes());
out.extend_from_slice(b" HTTP/1.1\r\n");
out.extend_from_slice(request.fields());
push_framing(out, encoder, declared);
if request.close {
out.extend_from_slice(b"connection: close\r\n");
}
out.extend_from_slice(b"\r\n");
match &request.body {
OutBody::Full(b) => {
encoder.write(b, out)?;
Ok(ReqWrite::Sent)
}
_ => Ok(ReqWrite::Streaming(encoder)),
}
}
const NO_QUEUED: Queued = Queued {
asked: Asked {
head: false,
connect: false,
},
close: false,
};
impl Default for ClientConn {
fn default() -> Self {
Self::new()
}
}
impl core::fmt::Debug for ClientConn {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
f.debug_struct("ClientConn")
.field("state", &self.state)
.field("request", &self.req)
.field("buffered", &self.buf.unread().len())
.field("limits", &self.limits)
.finish()
}
}
impl ClientConn {
#[must_use]
pub fn new() -> Self {
Self::with_limits(Limits::new())
}
#[must_use]
pub fn with_limits(limits: Limits) -> Self {
Self::from_buffer(Buffer::new(limits.buffer_capacity()), limits)
}
#[must_use]
pub fn with_buffer(storage: Vec<u8>, limits: Limits) -> Self {
Self::from_buffer(
Buffer::with_storage(storage, limits.buffer_capacity()),
limits,
)
}
fn from_buffer(buf: Buffer, limits: Limits) -> Self {
Self {
buf,
limits,
scanner: HeadScanner::new(),
trailers: HeadScanner::new(),
decoder: Decoder::new(),
info: ResponseInfo::EMPTY,
asked: Asked {
head: false,
connect: false,
},
upgrade_asked: false,
upgrade_offer: Vec::new(),
state: State::Idle,
req: ReqWrite::Sent,
keep_alive: true,
interim_bytes: 0,
interim_count: 0,
report_interim: false,
report_trailers: false,
waiting_continue: false,
body_refused: false,
req_chunked: false,
idempotent: true,
queue: [NO_QUEUED; MAX_PIPELINE_DEPTH as usize],
queue_start: 0,
queued: 0,
leniency: ClientLeniency::new(),
unfold: Unfold::new(),
}
}
#[must_use]
pub fn limits(&self) -> &Limits {
&self.limits
}
#[must_use = "bytes beyond the returned count were not taken"]
pub fn feed(&mut self, bytes: &[u8]) -> usize {
self.buf.feed(bytes)
}
pub fn read_buf(&mut self) -> &mut [u8] {
self.buf.spare()
}
pub fn commit(&mut self, n: usize) {
self.buf.commit(n);
}
pub fn feed_eof(&mut self) {
self.buf.set_eof();
}
pub fn send_request(&mut self, request: &Request, out: &mut Vec<u8>) -> Result<(), HttpError> {
if let Some(e) = request.error() {
return Err(e);
}
match self.state {
State::Idle => {}
State::Closed | State::Upgraded => {
return Err(HttpError::InvalidState(
"the connection is closed or upgraded",
));
}
_ => return self.send_pipelined(request, out),
}
if !self.buf.unread().is_empty() {
return Err(HttpError::UnexpectedData);
}
if !request.has_host {
return Err(HttpError::MissingHost);
}
if request.expect_continue && matches!(request.body, OutBody::Full(_)) {
return Err(HttpError::InvalidState(
"expect_continue needs a streamed body (content_length or chunked), \
sent after ClientEvent::Continue",
));
}
let method = request.method_kind;
self.req = write_request(request, out)?;
self.asked = Asked {
head: method == Method::Head,
connect: method == Method::Connect,
};
self.upgrade_asked = request.upgrade;
if request.upgrade {
self.upgrade_offer.clear();
for offer in request.upgrade_offers() {
if !self.upgrade_offer.is_empty() {
self.upgrade_offer.push(b',');
}
self.upgrade_offer.extend_from_slice(offer);
}
}
self.keep_alive = !request.close;
self.idempotent = method.is_idempotent();
self.req_chunked = matches!(request.body, OutBody::Chunked);
self.waiting_continue = request.expect_continue;
self.body_refused = false;
self.buf.unpin();
self.buf.pin_at_pos();
self.scanner.reset();
self.unfold = Unfold::new();
self.interim_bytes = 0;
self.interim_count = 0;
self.state = State::Head;
Ok(())
}
fn send_pipelined(&mut self, request: &Request, out: &mut Vec<u8>) -> Result<(), HttpError> {
if !self.can_send() {
return Err(HttpError::InvalidState(
if self.limits.pipeline_depth() <= 1 {
"a request is already outstanding (raise Limits::with_pipeline_depth to pipeline)"
} else {
"pipelining needs room in the pipeline depth, idempotent requests, \
every earlier body sent, and no close, upgrade, CONNECT, or expectation"
},
));
}
if !request.has_host {
return Err(HttpError::MissingHost);
}
let method = request.method_kind;
if !method.is_idempotent()
|| request.upgrade
|| request.expect_continue
|| !matches!(request.body, OutBody::Full(_))
{
return Err(HttpError::InvalidState(
"only idempotent requests with a full body and no upgrade or expectation \
can be pipelined",
));
}
let req = write_request(request, out)?;
debug_assert_eq!(req, ReqWrite::Sent);
let slot = (self.queue_start + self.queued) % self.queue.len();
self.queue[slot] = Queued {
asked: Asked {
head: method == Method::Head,
connect: false,
},
close: request.close,
};
self.queued += 1;
Ok(())
}
fn next_queued(&mut self) {
let next = self.queue[self.queue_start];
self.queue_start = (self.queue_start + 1) % self.queue.len();
self.queued -= 1;
self.asked = next.asked;
self.keep_alive = !next.close;
self.upgrade_asked = false;
self.idempotent = true;
self.req_chunked = false;
self.waiting_continue = false;
self.body_refused = false;
self.buf.unpin();
self.buf.pin_at_pos();
self.scanner.reset();
self.interim_bytes = 0;
self.interim_count = 0;
self.state = State::Head;
}
fn closing(&self) -> bool {
!self.keep_alive
|| (0..self.queued).any(|i| self.queue[(self.queue_start + i) % self.queue.len()].close)
}
#[must_use]
pub fn outstanding(&self) -> usize {
match self.state {
State::Idle => 0,
State::Head | State::Interim | State::Body => 1 + self.queued,
State::Complete | State::Closed | State::Upgraded => self.queued,
}
}
#[must_use]
pub fn can_send(&self) -> bool {
match self.state {
State::Idle => self.buf.unread().is_empty() && !self.buf.is_eof(),
State::Closed | State::Upgraded => false,
State::Head | State::Interim | State::Body | State::Complete => {
self.outstanding() < usize::from(self.limits.pipeline_depth())
&& self.idempotent
&& self.req == ReqWrite::Sent
&& !self.closing()
&& !self.upgrade_asked
&& !self.asked.connect
&& !self.waiting_continue
&& !self.body_refused
&& !self.buf.is_eof()
}
}
}
pub fn reset(&mut self) {
self.buf.reset();
self.scanner.reset();
self.trailers.reset();
self.decoder = Decoder::new();
self.info = ResponseInfo::EMPTY;
self.asked = NO_QUEUED.asked;
self.upgrade_asked = false;
self.state = State::Idle;
self.req = ReqWrite::Sent;
self.keep_alive = true;
self.interim_bytes = 0;
self.interim_count = 0;
self.waiting_continue = false;
self.body_refused = false;
self.req_chunked = false;
self.idempotent = true;
self.queue_start = 0;
self.queued = 0;
}
#[must_use]
pub fn into_buffer(mut self) -> Vec<u8> {
self.buf.take_storage()
}
pub fn shrink_idle(&mut self) -> bool {
if self.state != State::Idle {
return false;
}
self.buf.unpin();
let freed = self.buf.shrink();
if freed {
self.scanner.fields = Vec::new();
self.trailers.fields = Vec::new();
}
freed
}
pub fn send_body(&mut self, data: &[u8], out: &mut Vec<u8>) -> Result<(), HttpError> {
if self.body_refused {
return Err(REFUSED);
}
match &mut self.req {
ReqWrite::Streaming(encoder) => {
encoder.write(data, out)?;
self.waiting_continue = false;
Ok(())
}
ReqWrite::Sent => Err(HttpError::InvalidState("no streamed request in progress")),
}
}
pub fn end_body(&mut self, out: &mut Vec<u8>) -> Result<(), HttpError> {
self.end_body_with_trailers(&[], out)
}
pub fn end_body_with_trailers(
&mut self,
trailers: &[(&str, &[u8])],
out: &mut Vec<u8>,
) -> Result<(), HttpError> {
if self.body_refused {
return Err(REFUSED);
}
let ReqWrite::Streaming(encoder) = &mut self.req else {
return Err(HttpError::InvalidState("no streamed request in progress"));
};
if !trailers.is_empty() && !self.req_chunked {
return Err(HttpError::TrailerNotAllowed);
}
if *encoder == Encoder::Chunked {
push_last_chunk_with_trailers(out, trailers)?;
} else {
check_trailers(trailers)?;
encoder.finish(out)?;
}
self.req = ReqWrite::Sent;
self.waiting_continue = false;
Ok(())
}
pub fn poll(&mut self) -> Result<ClientEvent<'_>, HttpError> {
loop {
match self.state {
State::Idle => {
if !self.buf.unread().is_empty() {
return Err(self.fail(HttpError::UnexpectedData));
}
if self.buf.is_eof() {
self.state = State::Closed;
return Ok(ClientEvent::Closed);
}
return Ok(ClientEvent::Idle);
}
State::Head => return self.poll_head(),
State::Interim => {
self.next_head();
self.state = State::Head;
}
State::Body => return self.poll_body(),
State::Complete => {
if self.reusable() && self.queued > 0 {
self.next_queued();
} else if self.reusable() {
self.state = State::Idle;
} else {
self.state = State::Closed;
return Ok(ClientEvent::Closed);
}
}
State::Upgraded => return Ok(ClientEvent::Upgraded),
State::Closed => return Ok(ClientEvent::Closed),
}
}
}
fn poll_head(&mut self) -> Result<ClientEvent<'_>, HttpError> {
loop {
let start = self.buf.pin();
let visible = if self.leniency.obs_fold_in_responses {
let eof = self.buf.is_eof();
self.unfold.run(self.buf.tail_mut(start), eof)
} else {
self.buf.from(start).len()
};
let msg = self.buf.from(start);
let progress = self
.scanner
.advance(
msg.get(..visible).unwrap_or(msg),
&self.limits,
HeadKind::Response,
true,
)
.map_err(|e| self.fail(e))?;
let len = match progress {
Scan::Partial if self.buf.is_eof() => {
return Err(self.fail(HttpError::UnexpectedEof));
}
Scan::Partial => return Ok(ClientEvent::NeedInput),
Scan::Done(len) => len,
};
let info = interpret_response(
self.buf.from(start),
self.scanner.start_line(),
&self.scanner.fields,
&self.limits,
self.asked,
)
.map_err(|e| self.fail(e))?;
self.buf.consume(len);
self.buf.extend_pin_to_pos();
self.info = info;
let status = info.status;
let switched = if status == 101 {
let offered = self.upgrade_asked
&& info.upgrade_field
&& switches_to_offered(
self.buf.from(start),
&self.scanner.fields,
&self.upgrade_offer,
);
if !offered {
return Err(self.fail(HttpError::UnexpectedUpgrade));
}
true
} else {
self.asked.connect && (200..300).contains(&status)
};
if switched {
self.waiting_continue = false;
self.state = State::Upgraded;
} else if (100..200).contains(&status) {
self.interim_count = self.interim_count.saturating_add(1);
if self.interim_count > self.limits.max_interim() {
return Err(self.fail(HttpError::TooManyInterim));
}
self.interim_bytes = self.interim_bytes.saturating_add(len);
if self.interim_bytes > self.limits.max_header_bytes() {
return Err(self.fail(HttpError::HeadersTooLarge));
}
if status == 100 && self.waiting_continue {
self.waiting_continue = false;
self.next_head();
return Ok(ClientEvent::Continue);
}
if self.report_interim {
self.state = State::Interim;
return Ok(ClientEvent::Informational(self.view()));
}
self.next_head();
continue;
} else {
if self.waiting_continue {
self.waiting_continue = false;
self.body_refused = self.req != ReqWrite::Sent;
}
self.decoder.start(info.framing);
self.state = State::Body;
}
return Ok(ClientEvent::Response(self.view()));
}
}
fn next_head(&mut self) {
self.buf.unpin();
self.buf.pin_at_pos();
self.scanner.reset();
self.unfold = Unfold::new();
}
fn reusable(&self) -> bool {
self.keep_alive && self.info.keep_alive && self.req == ReqWrite::Sent && !self.body_refused
}
fn poll_body(&mut self) -> Result<ClientEvent<'_>, HttpError> {
let mut framing_bytes = 0;
let step = self.decoder.step(
self.buf.unread(),
self.buf.is_eof(),
&self.limits,
&mut self.trailers,
self.report_trailers,
&mut framing_bytes,
);
self.buf.consume(framing_bytes);
match step.map_err(|e| self.fail(e))? {
Step::Data(n) => {
let start = self.buf.pos();
self.buf.consume(n);
Ok(ClientEvent::Body(self.buf.range(start, start + n)))
}
Step::Trailers => Ok(ClientEvent::Trailers(Headers::new(
self.buf.unread(),
&self.trailers.fields,
))),
Step::Done => {
self.state = State::Complete;
Ok(ClientEvent::ResponseEnd)
}
Step::NeedMore => Ok(ClientEvent::NeedInput),
}
}
fn fail(&mut self, e: HttpError) -> HttpError {
self.state = State::Closed;
e
}
fn view(&self) -> ResponseHead<'_> {
ResponseHead::new(
self.buf.from(self.buf.pin()),
&self.scanner.fields,
&self.info,
)
}
#[must_use]
pub fn response(&self) -> Option<ResponseHead<'_>> {
match self.state {
State::Body | State::Complete | State::Upgraded => Some(self.view()),
_ => None,
}
}
#[must_use]
pub fn is_upgraded(&self) -> bool {
self.state == State::Upgraded
}
#[must_use]
pub fn remaining_input(&self) -> &[u8] {
self.buf.unread()
}
pub fn set_report_informational(&mut self, report: bool) {
self.report_interim = report;
}
pub fn set_report_trailers(&mut self, report: bool) {
self.report_trailers = report;
}
pub fn set_leniency(&mut self, leniency: ClientLeniency) {
self.leniency = leniency;
}
#[must_use]
pub fn is_waiting_for_continue(&self) -> bool {
self.waiting_continue
}
#[must_use]
pub fn is_reusable(&self) -> bool {
match self.state {
State::Idle => self.buf.unread().is_empty() && !self.buf.is_eof(),
State::Complete => self.reusable() && self.buf.unread().is_empty(),
_ => false,
}
}
#[must_use]
pub fn progress(&self) -> Progress {
if matches!(self.req, ReqWrite::Streaming(_))
&& !self.body_refused
&& !matches!(self.state, State::Closed | State::Upgraded)
{
return Progress::WritingBody;
}
match self.state {
State::Idle => Progress::Idle,
State::Head if self.buf.from(self.buf.pin()).is_empty() => Progress::AwaitingResponse,
State::Head | State::Interim => Progress::Head,
State::Body => Progress::Body,
State::Complete if self.reusable() => Progress::Idle,
State::Complete | State::Closed => Progress::Closed,
State::Upgraded => Progress::Upgraded,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::head::BodyFraming;
fn get() -> Request {
Request::get("/").header("host", "a")
}
#[test]
fn test_interim_responses_are_skipped() {
let mut conn = ClientConn::new();
conn.send_request(&get(), &mut Vec::new()).unwrap();
let _ = conn.feed(b"HTTP/1.1 100 Continue\r\n\r\nHTTP/1.1 103 Early Hints\r\nLink: </a>\r\n\r\nHTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok");
let ClientEvent::Response(r) = conn.poll().unwrap() else {
panic!("expected response");
};
assert_eq!(r.status(), 200);
assert!(matches!(conn.poll().unwrap(), ClientEvent::Body(b"ok")));
assert!(matches!(conn.poll().unwrap(), ClientEvent::ResponseEnd));
assert!(matches!(conn.poll().unwrap(), ClientEvent::Idle));
}
#[test]
fn test_unrequested_upgrade_and_unsolicited_data_are_errors() {
let mut conn = ClientConn::new();
conn.send_request(&get(), &mut Vec::new()).unwrap();
let _ = conn
.feed(b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: x\r\nConnection: upgrade\r\n\r\n");
assert_eq!(conn.poll().unwrap_err(), HttpError::UnexpectedUpgrade);
let mut conn = ClientConn::new();
let _ = conn.feed(b"HTTP/1.1 200 OK\r\n\r\n");
assert_eq!(conn.poll().unwrap_err(), HttpError::UnexpectedData);
}
#[test]
fn test_switching_protocols_must_name_an_offered_protocol() {
let answer = |offers: &[&str], upgrade: &str| {
let mut request = Request::get("/").header("host", "a");
for offer in offers {
request = request.upgrade(offer);
}
let mut conn = ClientConn::new();
conn.send_request(&request, &mut Vec::new()).unwrap();
let response = alloc::format!(
"HTTP/1.1 101 Switching Protocols\r\nConnection: upgrade\r\n{upgrade}\r\n"
);
let _ = conn.feed(response.as_bytes());
match conn.poll() {
Ok(ClientEvent::Response(r)) => {
assert_eq!(r.status(), 101);
matches!(conn.poll(), Ok(ClientEvent::Upgraded))
}
Err(e) => {
assert_eq!(e, HttpError::UnexpectedUpgrade);
false
}
other => panic!("unexpected {other:?}"),
}
};
assert!(answer(&["websocket"], "Upgrade: websocket\r\n"));
assert!(answer(&["websocket"], "Upgrade: WebSocket\r\n"));
assert!(answer(&["websocket/13"], "Upgrade: websocket\r\n"));
assert!(answer(&["websocket"], "Upgrade: websocket/13\r\n"));
assert!(answer(&["foo", "websocket"], "Upgrade: websocket\r\n"));
assert!(!answer(&["websocket"], "Upgrade: h2c\r\n"));
assert!(!answer(&["websocket"], "Upgrade: foo\r\n"));
assert!(!answer(&["websocket"], "Upgrade: websocket, foo\r\n"));
assert!(!answer(
&["websocket"],
"Upgrade: websocket\r\nUpgrade: foo\r\n"
));
assert!(!answer(&["websocket/13"], "Upgrade: websocket/14\r\n"));
assert!(!answer(&["websocket"], "Upgrade:\r\n"));
assert!(!answer(&["websocket"], ""));
}
#[test]
fn test_head_response_has_no_body_and_connection_is_reused() {
let mut conn = ClientConn::new();
conn.send_request(
&Request::new("HEAD", "/").header("host", "a"),
&mut Vec::new(),
)
.unwrap();
let _ = conn.feed(b"HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\n");
let ClientEvent::Response(r) = conn.poll().unwrap() else {
panic!("expected response");
};
assert_eq!(r.framing(), BodyFraming::None);
assert!(matches!(conn.poll().unwrap(), ClientEvent::ResponseEnd));
assert!(matches!(conn.poll().unwrap(), ClientEvent::Idle));
conn.send_request(&get(), &mut Vec::new()).unwrap();
assert!(matches!(
conn.send_request(&get(), &mut Vec::new()),
Err(HttpError::InvalidState(_))
));
}
#[test]
fn test_close_and_streaming_request_prevent_reuse() {
let mut conn = ClientConn::new();
conn.send_request(&get().close(), &mut Vec::new()).unwrap();
let _ = conn.feed(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n");
while !matches!(conn.poll().unwrap(), ClientEvent::ResponseEnd) {}
assert!(matches!(conn.poll().unwrap(), ClientEvent::Closed));
assert!(conn.send_request(&get(), &mut Vec::new()).is_err());
let mut conn = ClientConn::new();
let mut out = Vec::new();
conn.send_request(&Request::post("/").header("host", "a").chunked(), &mut out)
.unwrap();
let _ = conn.feed(b"HTTP/1.1 413 Content Too Large\r\nContent-Length: 0\r\n\r\n");
while !matches!(conn.poll().unwrap(), ClientEvent::ResponseEnd) {}
assert!(matches!(conn.poll().unwrap(), ClientEvent::Closed));
assert!(matches!(conn.send_body(b"x", &mut out), Ok(())));
assert!(matches!(conn.end_body(&mut out), Ok(())));
assert!(matches!(
conn.end_body(&mut out),
Err(HttpError::InvalidState(_))
));
}
#[test]
fn test_eof_before_response_is_an_error() {
let mut conn = ClientConn::new();
conn.send_request(&get(), &mut Vec::new()).unwrap();
conn.feed_eof();
assert_eq!(conn.poll().unwrap_err(), HttpError::UnexpectedEof);
}
}