base_http_client/
lib.rs

1// Released under the MIT License.
2// Copyright, 2026, by Samuel Williams.
3
4//! Endpoint-scoped clients with explicit, supervised protocol progress.
5#![doc = include_str!("../readme.md")]
6#![forbid(unsafe_code)]
7
8mod body;
9mod connection;
10mod driver;
11mod error;
12mod outgoing;
13mod reuse;
14mod sink;
15mod writable;
16pub use writable::{PendingResponse, RequestBodyWriter, StreamingResponseBody, WritableRequest};
17
18pub use body::ResponseBody;
19pub use connection::Connection;
20pub use driver::Driver;
21pub use error::Error;
22
23use base_http::{Authority, InterimHandler, Method, Request, Response, Scheme, body::ReadableBody};
24use base_pool::MultiplexPool;
25use base_scheduler::{
26    Cancellation, SocketClose, SocketConnect, SocketOpen, SocketSetOption, SocketStreamReading,
27    SocketStreamWriting, TcpNoDelay,
28};
29use driver::{CancelOnDrop, Operation, Pending, Queue, checked};
30use std::{
31    convert::Infallible,
32    net::SocketAddr,
33    sync::{Arc, atomic::Ordering},
34};
35
36/// A resolved HTTP endpoint. Its HTTP policy is independent of TLS identity
37/// and the address used to establish the underlying transport.
38#[derive(Clone, Debug)]
39pub struct Endpoint {
40    pub address: SocketAddr,
41    pub authority: Authority,
42    #[cfg(feature = "rustls")]
43    pub(crate) tls: Option<Tls>,
44}
45#[cfg(feature = "rustls")]
46#[derive(Clone, Debug)]
47pub(crate) struct Tls {
48    config: Arc<rustls::ClientConfig>,
49    server_name: String,
50}
51impl Endpoint {
52    pub fn new(address: SocketAddr) -> Self {
53        Self {
54            address,
55            authority: address.to_string().into(),
56            #[cfg(feature = "rustls")]
57            tls: None,
58        }
59    }
60    /// Enable HTTPS with explicit certificate trust and server identity. The
61    /// HTTP implementation supplies the ALPN offers when connecting.
62    #[cfg(feature = "rustls")]
63    pub fn with_tls(
64        mut self,
65        config: Arc<rustls::ClientConfig>,
66        server_name: impl Into<String>,
67    ) -> Self {
68        self.tls = Some(Tls {
69            config,
70            server_name: server_name.into(),
71        });
72        self
73    }
74    pub fn scheme(&self) -> &'static str {
75        #[cfg(feature = "rustls")]
76        if self.tls.is_some() {
77            return "https";
78        }
79        "http"
80    }
81    // The scheme constant, so requests need not convert the text.
82    pub(crate) fn request_scheme(&self) -> Scheme {
83        #[cfg(feature = "rustls")]
84        if self.tls.is_some() {
85            return Scheme::HTTPS;
86        }
87        Scheme::HTTP
88    }
89}
90
91#[derive(Clone, Debug)]
92pub struct Options {
93    pub pool: base_pool::Options,
94    pub http1: base_http1::Options,
95    /// HTTP engine selection, shared by plain and encrypted establishment.
96    pub protocols: base_http_protocol::HttpProtocols,
97    /// Bound accepted calls, including queued, uploading and unread responses.
98    pub max_requests: usize,
99    pub response_buffer_bytes: usize,
100    pub response_buffer_frames: usize,
101    /// Disable Nagle's algorithm on each TCP connection, so small writes such
102    /// as streamed frames are not held back waiting for acknowledgement. This
103    /// is a best-effort latency hint: a socket that rejects it remains usable.
104    pub tcp_nodelay: bool,
105}
106impl Default for Options {
107    fn default() -> Self {
108        Self {
109            pool: base_pool::Options::default(),
110            http1: base_http1::Options::default(),
111            protocols: base_http_protocol::HttpProtocols::default(),
112            max_requests: 1024,
113            response_buffer_bytes: 64 * 1024,
114            response_buffer_frames: 16,
115            tcp_nodelay: true,
116        }
117    }
118}
119impl Options {
120    pub fn validate(&self) -> Result<(), Error> {
121        self.pool.validate()?;
122        self.http1.validate()?;
123        if self.max_requests == 0
124            || self.response_buffer_frames == 0
125            || self.response_buffer_bytes < self.http1.read_size.max(self.http1.max_head_bytes)
126        {
127            return Err(Error::InvalidOptions);
128        }
129        Ok(())
130    }
131}
132
133/// Durable remote-server interface. Clones share connection admission and
134/// supervision. `call` submits pull bodies; `open` lends direct HTTP/1 directions.
135/// Neither operation chooses a process-global scheduler.
136pub struct Client<'scope> {
137    endpoint: Endpoint,
138    queue: Arc<Queue<'scope>>,
139    pool: MultiplexPool<Connection<'scope>>,
140    context: Arc<driver::Context<'scope>>,
141}
142impl<'scope> Client<'scope> {
143    /// Construct without dialing. Poll the returned driver concurrently with
144    /// calls. Bodies may borrow for the driver's scope; spawning requires owned
145    /// `Send + 'static` captures, as with any scheduler task.
146    pub fn new<S>(
147        scheduler: &'scope S,
148        endpoint: Endpoint,
149        options: Options,
150    ) -> Result<(Self, Driver<'scope>), Error>
151    where
152        S: SocketOpen
153            + SocketConnect
154            + SocketClose
155            + SocketSetOption<TcpNoDelay>
156            + SocketStreamReading
157            + SocketStreamWriting
158            + 'scope,
159        S::Socket: 'static,
160    {
161        options.validate()?;
162        let driver::ClientParts {
163            queue,
164            pool,
165            driver,
166            context,
167        } = driver::make(scheduler, endpoint.clone(), options);
168        Ok((
169            Self {
170                endpoint,
171                queue,
172                pool,
173                context,
174            },
175            driver,
176        ))
177    }
178    pub fn endpoint(&self) -> &Endpoint {
179        &self.endpoint
180    }
181    pub fn pool_stats(&self) -> base_pool::Stats {
182        self.pool.stats()
183    }
184    /// Reject new submissions, finish accepted calls, then retire connections.
185    /// Cancel the driver's lifetime signal to force active calls down.
186    pub fn begin_close(&self) {
187        self.queue.begin_close();
188    }
189
190    /// Return final response metadata promptly. The driver continues upload
191    /// independently of response consumption. No automatic retries occur.
192    pub async fn call<B>(
193        &self,
194        request: Request<B>,
195        cancellation: &Cancellation,
196    ) -> Result<Response<ResponseBody>, Error>
197    where
198        B: ReadableBody + 'scope,
199        B::Data: Into<Vec<u8>>,
200    {
201        self.call_with_interim(
202            request,
203            |_: base_http::ResponseMetadata| async { Ok::<_, Infallible>(()) },
204            cancellation,
205        )
206        .await
207    }
208    /// Observe informational responses in order, without accumulating a log.
209    pub async fn call_with_interim<B, I>(
210        &self,
211        mut request: Request<B>,
212        interim: I,
213        cancellation: &Cancellation,
214    ) -> Result<Response<ResponseBody>, Error>
215    where
216        B: ReadableBody + 'scope,
217        B::Data: Into<Vec<u8>>,
218        I: InterimHandler + 'scope,
219    {
220        if cancellation.is_cancelled() {
221            request.body.close(None);
222            return Err(Error::Cancelled);
223        }
224        if request.method == Method::CONNECT
225            || request.protocol.is_some()
226            || request.fields.contains("upgrade")
227        {
228            request.body.close(None);
229            return Err(Error::Unsupported("managed client tunnels and upgrades"));
230        }
231        if request
232            .scheme
233            .as_deref()
234            .is_some_and(|scheme| scheme != self.endpoint.scheme())
235        {
236            request.body.close(None);
237            return Err(Error::Unsupported(
238                "request scheme does not match the endpoint",
239            ));
240        }
241        request
242            .scheme
243            .get_or_insert_with(|| self.endpoint.request_scheme());
244        request
245            .authority
246            .get_or_insert_with(|| self.endpoint.authority.clone());
247        let local = Cancellation::new();
248        let mut guard = CancelOnDrop(local.clone(), true);
249        let (reply, response) = futures_channel::oneshot::channel();
250        let pending = match Pending::new(self.queue.clone()) {
251            Ok(pending) => pending,
252            Err(error) => {
253                request.body.close(None);
254                return Err(error);
255            }
256        };
257        let operation = Operation {
258            request: request.map_body(|body| outgoing::OutgoingBody::new(body, local.clone())),
259            interim: Box::new(interim),
260            reply: Some(reply),
261            cancellation: local,
262            lifetime: cancellation.clone(),
263            _pending: Some(pending),
264        };
265        // A queue closed since admission drops the operation and its reply sender, so the response reports `Closed`, as for calls queued when the driver stops.
266        let _ = self.queue.push(operation);
267        let response = checked(
268            async { response.await.map_err(|_| Error::Closed)? },
269            cancellation,
270        )
271        .await?;
272        guard.1 = false;
273        Ok(response)
274    }
275}
276impl Clone for Client<'_> {
277    fn clone(&self) -> Self {
278        self.queue.handles.fetch_add(1, Ordering::Relaxed);
279        Self {
280            endpoint: self.endpoint.clone(),
281            queue: self.queue.clone(),
282            pool: self.pool.clone(),
283            context: self.context.clone(),
284        }
285    }
286}
287impl Drop for Client<'_> {
288    fn drop(&mut self) {
289        if self.queue.handles.fetch_sub(1, Ordering::AcqRel) == 1 {
290            self.queue.begin_close();
291        }
292    }
293}
base_http_client/lib.rs

base_http_client/
lib.rs
