base_http1/
client.rs

1// Released under the MIT License.
2// Copyright, 2026, by Samuel Williams.
3
4use crate::{
5    Error, Exchange, IncomingBody, Options, RequestReader, ResponseWriter, input, output, wire,
6};
7use base_endpoint::stream::{ReadableStream, WritableStream};
8use base_http::{
9    InterimHandler, InterimResponses, Method, Protocol, Request, RequestMetadata, Response,
10    ResponseMetadata, Status, Version,
11    body::{BodyFrame, ReadableBody},
12};
13use base_scheduler::Cancellation;
14use std::io;
15
16/// Sequential client output. Drive this lane concurrently with `ResponseReader`
17/// so an upstream response can arrive before request production completes.
18/// No dialing, pooling, detached task or Expect/Continue wait policy is hidden.
19pub struct RequestWriter<W> {
20    output: ResponseWriter<W>,
21    body: Option<output::BodyEncoder>,
22    persistent: bool,
23    buffer: Vec<u8>,
24}
25
26impl<W: WritableStream> RequestWriter<W> {
27    pub fn new(writer: W) -> Self {
28        Self {
29            output: ResponseWriter::new(writer),
30            body: None,
31            persistent: false,
32            buffer: Vec::new(),
33        }
34    }
35    /// Whether this output lane permits another request. This does not check
36    /// completion of the independent response lane.
37    pub fn is_reusable(&self) -> bool {
38        self.output.is_usable()
39    }
40
41    /// Send one request. `true` permits another output message; the owner must
42    /// also finish the matching response before reusing the HTTP/1 connection.
43    /// The body is closed on return, including error and cancellation.
44    pub async fn write_request<B: ReadableBody>(
45        &mut self,
46        request: &mut Request<B>,
47        cancellation: &Cancellation,
48    ) -> Result<bool, Error> {
49        let result = self.write_inner(request, cancellation).await;
50        request.body.close(None);
51        result
52    }
53
54    async fn write_inner<B: ReadableBody>(
55        &mut self,
56        request: &mut Request<B>,
57        cancellation: &Cancellation,
58    ) -> Result<bool, Error> {
59        cancellation.check()?;
60        let (framing, persistent) = request_plan(
61            &mut self.buffer,
62            &request.metadata,
63            request.body.length(),
64            request.body.may_have_trailers(),
65        )?;
66        self.output.begin()?;
67        self.output.write_all(&self.buffer, cancellation).await?;
68        self.output.flush(cancellation).await?;
69        self.output
70            .write_body(&mut request.body, framing, false, cancellation)
71            .await?;
72        self.output.completed(persistent);
73        Ok(persistent)
74    }
75
76    /// Begin imperative output. A missing content length uses chunked HTTP/1.1
77    /// framing; declared fixed lengths are checked at every write and at EOF.
78    /// After failure or cancellation this lane must be retired.
79    pub async fn begin_request(
80        &mut self,
81        metadata: &RequestMetadata,
82        cancellation: &Cancellation,
83    ) -> Result<(), Error> {
84        cancellation.check()?;
85        let (framing, persistent) =
86            request_plan(&mut self.buffer, metadata, metadata.content_length, false)?;
87        self.output.begin()?;
88        self.output.write_all(&self.buffer, cancellation).await?;
89        self.output.flush(cancellation).await?;
90        // Activate the body only once the whole head is sent, so frames cannot follow a failed or dropped head.
91        self.persistent = persistent;
92        self.body = Some(output::BodyEncoder::new(framing, false));
93        Ok(())
94    }
95
96    /// Directly encode and flush one frame. Taking the encoder before awaiting
97    /// makes a dropped, partially completed write permanently non-resumable.
98    pub async fn write_frame<D: AsRef<[u8]>>(
99        &mut self,
100        frame: BodyFrame<D>,
101        cancellation: &Cancellation,
102    ) -> Result<(), Error> {
103        cancellation.check()?;
104        let mut body = self
105            .body
106            .take()
107            .ok_or(Error::Invalid("request output is not active"))?;
108        let end = frame.is_end();
109        let flush = !matches!(&frame, BodyFrame::End) || body.end_requires_output();
110        body.write_frame(&mut self.output, frame, cancellation)
111            .await?;
112        if flush {
113            self.output.flush(cancellation).await?;
114        }
115        if end {
116            self.output.completed(self.persistent);
117        } else {
118            self.body = Some(body);
119        }
120        Ok(())
121    }
122
123    /// Flush pending transport output. A failure retires the lane; accepted
124    /// bytes must not be replayed after an interrupted flush.
125    pub async fn flush(&mut self, cancellation: &Cancellation) -> Result<(), Error> {
126        let result = self.output.flush(cancellation).await;
127        if result.is_err() {
128            self.body = None;
129            self.output.completed(false);
130        }
131        result
132    }
133
134    pub async fn close(&mut self) -> io::Result<()> {
135        self.output.close().await
136    }
137}
138
139fn request_plan(
140    bytes: &mut Vec<u8>,
141    head: &RequestMetadata,
142    length: Option<u64>,
143    may_have_trailers: bool,
144) -> Result<(output::Framing, bool), Error> {
145    let version = head.version.unwrap_or(Version::Http11);
146    let label = match version {
147        Version::Http10 => "HTTP/1.0",
148        Version::Http11 => "HTTP/1.1",
149        _ => return Err(Error::Invalid("unsupported request version")),
150    };
151    let fields = &head.fields;
152    let host = wire::singleton(fields, "host")?;
153    if let Some(host) = host {
154        wire::authority(
155            std::str::from_utf8(host).map_err(|_| Error::Invalid("invalid host"))?,
156            false,
157        )?;
158    }
159    let add_host = if let Some(authority) = &head.authority {
160        wire::authority(authority, false)?;
161        if let Some(host) = host {
162            if !host.eq_ignore_ascii_case(authority.as_bytes()) {
163                return Err(Error::Invalid("authority differs from Host"));
164            }
165            None
166        } else {
167            Some(authority.as_str())
168        }
169    } else {
170        None
171    };
172    if version == Version::Http11 && host.is_none() && add_host.is_none() {
173        return Err(Error::Invalid("missing host"));
174    }
175    let upgrade = if let Some(protocol) = &head.protocol {
176        if version != Version::Http11 {
177            return Err(Error::Invalid("upgrade requires HTTP/1.1"));
178        }
179        if let Some(offer) = wire::singleton(fields, "upgrade")? {
180            if !offer.eq_ignore_ascii_case(protocol.as_str().as_bytes()) {
181                return Err(Error::Invalid("conflicting upgrade protocol"));
182            }
183            None
184        } else {
185            Some(protocol.as_str())
186        }
187    } else {
188        None
189    };
190    wire::connection_options(fields)?;
191    let add_upgrade = head.protocol.is_some() && !wire::has_token(fields, "connection", b"upgrade");
192    if version == Version::Http11
193        && (add_upgrade || wire::has_token(fields, "connection", b"upgrade"))
194        && !fields.contains("upgrade")
195        && upgrade.is_none()
196    {
197        return Err(Error::Invalid("missing upgrade protocol"));
198    }
199    let connection = head
200        .persistent
201        .map(|p| if p { "keep-alive" } else { "close" })
202        .filter(|value| !wire::has_token(fields, "connection", value.as_bytes()));
203    let target = if head.method == Method::CONNECT {
204        if head.path.is_some() {
205            return Err(Error::Invalid("CONNECT cannot have a path"));
206        }
207        head.authority
208            .as_deref()
209            .ok_or(Error::Invalid("CONNECT requires authority"))?
210    } else {
211        head.path.as_deref().unwrap_or("/")
212    };
213    wire::request_target(&head.method, target)?;
214    let (raw_length, raw_chunked) = wire::framing(fields)?;
215    let declared = wire::declared_length(head.content_length, raw_length)?;
216    if declared.is_some() && raw_chunked {
217        return Err(Error::Invalid("content-length with transfer-encoding"));
218    }
219    if raw_chunked && version != Version::Http11 {
220        return Err(Error::Invalid("HTTP/1.0 transfer-encoding"));
221    }
222    if let (Some(length), Some(declared)) = (length, declared) {
223        if length != declared {
224            return Err(Error::LengthMismatch);
225        }
226    }
227    let chunked = raw_chunked || may_have_trailers || fields.contains("trailer");
228    let framing = if chunked {
229        if version != Version::Http11 {
230            return Err(Error::Invalid("trailers require HTTP/1.1 chunked framing"));
231        }
232        output::Framing::Chunked(declared.or(length))
233    } else if let Some(length) = declared.or(length) {
234        output::Framing::Fixed(length)
235    } else if version == Version::Http11 {
236        output::Framing::Chunked(None)
237    } else {
238        return Err(Error::Invalid("HTTP/1.0 request requires a body length"));
239    };
240    let chunked = matches!(framing, output::Framing::Chunked(_));
241    let persistent = head.persistent != Some(false)
242        && !wire::has_token(fields, "connection", b"close")
243        && (version == Version::Http11
244            || head.persistent == Some(true)
245            || wire::has_token(fields, "connection", b"keep-alive"));
246    wire::output_fields(fields.pairs())?;
247    // Validate first, then encode directly into the connection's reusable buffer.
248    bytes.clear();
249    use std::io::Write;
250    // Writing into a `Vec` cannot fail.
251    write!(bytes, "{} {target} {label}\r\n", head.method.as_str()).expect("writing to a Vec");
252    for field in fields.pairs() {
253        if chunked && field.name().eq_ignore_ascii_case("content-length") {
254            continue;
255        }
256        bytes.extend_from_slice(field.name().as_bytes());
257        bytes.extend_from_slice(b": ");
258        bytes.extend_from_slice(field.value());
259        bytes.extend_from_slice(b"\r\n");
260    }
261    for (name, value) in [
262        ("Host", add_host),
263        ("Upgrade", upgrade),
264        ("Connection", add_upgrade.then_some("upgrade")),
265        ("Connection", connection),
266        (
267            "Transfer-Encoding",
268            (chunked && !raw_chunked).then_some("chunked"),
269        ),
270    ] {
271        if let Some(value) = value {
272            write!(bytes, "{name}: {value}\r\n").expect("writing to a Vec");
273        }
274    }
275    if let output::Framing::Fixed(length) = framing {
276        if raw_length.is_none()
277            && (declared.is_some()
278                || length != 0
279                || matches!(head.method.as_str(), "POST" | "PUT" | "PATCH"))
280        {
281            write!(bytes, "Content-Length: {length}\r\n").expect("writing to a Vec");
282        }
283    }
284    bytes.extend_from_slice(b"\r\n");
285    Ok((framing, persistent))
286}
287
288/// Sequential client input sharing the native request body's chunk decoder.
289/// The matching request head supplies HEAD/CONNECT semantics. This reader does
290/// not queue outstanding requests; the owner pairs each response in wire order.
291/// Switch responses expose raw input only; coordinating the outgoing switch
292/// and managed bidirectional client bodies is not implemented here.
293pub struct ResponseReader<R> {
294    input: RequestReader<R>,
295    persistent: bool,
296}
297
298impl<R: ReadableStream> ResponseReader<R> {
299    pub fn new(reader: R, options: Options) -> Result<Self, Error> {
300        Ok(Self {
301            input: RequestReader::new(reader, options)?,
302            persistent: true,
303        })
304    }
305    pub fn is_reusable(&self) -> bool {
306        self.persistent && self.input.is_reusable()
307    }
308
309    /// Deliver interim heads in order, then return the final response. `101`
310    /// and successful CONNECT yield raw input and permanently end HTTP reuse.
311    /// The connection owner supervises the independent request-writing lane.
312    pub async fn read_response<I: InterimHandler>(
313        &mut self,
314        request: &RequestMetadata,
315        interim: &mut InterimResponses<I>,
316        cancellation: &Cancellation,
317    ) -> Result<Response<IncomingBody<'_, R>>, Error> {
318        cancellation.check()?;
319        if !self.is_reusable() {
320            return Err(Error::Invalid("response input is not reusable"));
321        }
322        let max_fields = self.input.options.max_fields;
323        let head = loop {
324            self.input.poisoned = true;
325            let mut head = self
326                .input
327                .read_section(cancellation, "message head", |bytes| {
328                    wire::response(bytes, max_fields)
329                })
330                .await?
331                .ok_or(Error::Truncated)?;
332            if head.status.is_final() {
333                interim.finish();
334                break head;
335            }
336            if head.fields.contains("content-length")
337                || head.fields.contains("transfer-encoding")
338                || head.version != Some(Version::Http11)
339            {
340                return Err(Error::Invalid("invalid interim framing"));
341            }
342            head.fields = wire::application_fields(head.fields)?.0;
343            interim
344                .send_interim_response(head, cancellation)
345                .await
346                .map_err(|error| {
347                    if cancellation.is_cancelled() {
348                        Error::Cancelled
349                    } else {
350                        Error::Body(Box::new(error))
351                    }
352                })?;
353        };
354        self.response_body(request, head)
355    }
356
357    fn response_body(
358        &mut self,
359        request: &RequestMetadata,
360        mut head: ResponseMetadata,
361    ) -> Result<Response<IncomingBody<'_, R>>, Error> {
362        let connect = request.method == Method::CONNECT && head.status.is_success();
363        let upgrade = head.status == Status::SWITCHING_PROTOCOLS;
364        let tunnel = connect || upgrade;
365        // CONNECT framing fields are ignored by recipients (RFC 9112 6.3).
366        let (length, chunked) = if connect {
367            (None, false)
368        } else {
369            wire::framing(&head.fields)?
370        };
371        if chunked && head.version != Some(Version::Http11) {
372            return Err(Error::Invalid("HTTP/1.0 transfer-encoding"));
373        }
374        if upgrade {
375            if length.is_some()
376                || chunked
377                || !wire::has_token(&head.fields, "connection", b"upgrade")
378            {
379                return Err(Error::Invalid("invalid upgrade response"));
380            }
381            let selected = wire::singleton(&head.fields, "upgrade")?
382                .ok_or(Error::Invalid("missing selected protocol"))?;
383            let selected = std::str::from_utf8(selected)
384                .map_err(|_| Error::Invalid("invalid selected protocol"))?;
385            let offered = request
386                .protocol
387                .as_ref()
388                .is_some_and(|p| p.as_str().eq_ignore_ascii_case(selected))
389                || (wire::has_token(&request.fields, "connection", b"upgrade")
390                    && wire::has_token(&request.fields, "upgrade", selected.as_bytes()));
391            if !offered {
392                return Err(Error::Invalid("upgrade protocol was not offered"));
393            }
394            head.protocol = Some(
395                Protocol::new(selected)
396                    .map_err(|_| Error::Invalid("unsupported selected protocol"))?,
397            );
398        }
399        let absent = !tunnel
400            && (request.method == Method::HEAD
401                || head.status == Status::NO_CONTENT
402                || head.status == Status::NOT_MODIFIED);
403        if head.status == Status::NO_CONTENT && !connect && (length.is_some() || chunked) {
404            return Err(Error::Invalid("framing fields on 204"));
405        }
406        let raw = tunnel || (!absent && !chunked && length.is_none());
407        let framing = if absent || raw {
408            input::Framing::Fixed(0)
409        } else if chunked {
410            input::Framing::CHUNKED
411        } else {
412            input::Framing::Fixed(length.unwrap())
413        };
414        let body_length = if raw {
415            None
416        } else if absent && head.status == Status::NO_CONTENT {
417            Some(0)
418        } else {
419            length
420        };
421        let response_persistent = !wire::has_token(&head.fields, "connection", b"close")
422            && (head.version == Some(Version::Http11)
423                || wire::has_token(&head.fields, "connection", b"keep-alive"));
424        head.content_length = length;
425        let (fields, trailer_options) = wire::application_fields(head.fields)?;
426        head.fields = fields;
427        let exchange = Exchange::new(request, trailer_options);
428        let request_persistent = exchange.is_persistent();
429        self.persistent = request_persistent && response_persistent && !raw;
430        if tunnel {
431            exchange.switch();
432        }
433        self.input.previous = Some(exchange.clone());
434        self.input.poisoned = false;
435        let forbid_content = head.status == Status::RESET_CONTENT && !tunnel;
436        let body = IncomingBody::new(
437            &mut self.input,
438            exchange,
439            framing,
440            body_length,
441            raw,
442            forbid_content,
443        );
444        Ok(Response::from_parts(head, body))
445    }
446
447    pub async fn close(&mut self) -> io::Result<()> {
448        self.input.close().await
449    }
450}
base_http1/client.rs

base_http1/
client.rs
