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.
136///
137/// `'scope` is the scheduler borrow, and it also bounds the request bodies and interim observers that calls queue for the driver. The client and its driver share that queue, so neither can outlive the scheduler. The type is invariant over `'scope`: shortening it would let a call queue a body that borrows shorter-lived data for a driver that keeps running after the data is gone.
138///
139/// To keep a client for longer, give the scheduler a longer life. Create it at the application root and store the client in a struct generic over `'scope`, or use a scheduler that lives for `'static`, whose `Client<'static>` can be stored anywhere.
140pub struct Client<'scope> {
141    endpoint: Endpoint,
142    queue: Arc<Queue<'scope>>,
143    pool: MultiplexPool<Connection<'scope>>,
144    context: Arc<driver::Context<'scope>>,
145}
146impl<'scope> Client<'scope> {
147    /// Construct without dialing. Poll the returned driver concurrently with
148    /// calls. Bodies may borrow for the driver's scope; spawning requires owned
149    /// `Send + 'static` captures, as with any scheduler task.
150    pub fn new<S>(
151        scheduler: &'scope S,
152        endpoint: Endpoint,
153        options: Options,
154    ) -> Result<(Self, Driver<'scope>), Error>
155    where
156        S: SocketOpen
157            + SocketConnect
158            + SocketClose
159            + SocketSetOption<TcpNoDelay>
160            + SocketStreamReading
161            + SocketStreamWriting
162            + 'scope,
163        S::Socket: 'static,
164    {
165        options.validate()?;
166        let driver::ClientParts {
167            queue,
168            pool,
169            driver,
170            context,
171        } = driver::make(scheduler, endpoint.clone(), options);
172        Ok((
173            Self {
174                endpoint,
175                queue,
176                pool,
177                context,
178            },
179            driver,
180        ))
181    }
182    pub fn endpoint(&self) -> &Endpoint {
183        &self.endpoint
184    }
185    pub fn pool_stats(&self) -> base_pool::Stats {
186        self.pool.stats()
187    }
188    /// Reject new submissions, finish accepted calls, then retire connections.
189    /// Cancel the driver's lifetime signal to force active calls down.
190    pub fn begin_close(&self) {
191        self.queue.begin_close();
192    }
193
194    /// Return final response metadata promptly. The driver continues upload
195    /// independently of response consumption. A reused idle connection is
196    /// checked before any request bytes are sent, and one the server closed is
197    /// replaced. No automatic retries occur.
198    pub async fn call<B>(
199        &self,
200        request: Request<B>,
201        cancellation: &Cancellation,
202    ) -> Result<Response<ResponseBody>, Error>
203    where
204        B: ReadableBody + 'scope,
205        B::Data: Into<Vec<u8>>,
206    {
207        self.call_with_interim(
208            request,
209            |_: base_http::ResponseMetadata| async { Ok::<_, Infallible>(()) },
210            cancellation,
211        )
212        .await
213    }
214    /// Observe informational responses in order, without accumulating a log.
215    pub async fn call_with_interim<B, I>(
216        &self,
217        mut request: Request<B>,
218        interim: I,
219        cancellation: &Cancellation,
220    ) -> Result<Response<ResponseBody>, Error>
221    where
222        B: ReadableBody + 'scope,
223        B::Data: Into<Vec<u8>>,
224        I: InterimHandler + 'scope,
225    {
226        if cancellation.is_cancelled() {
227            request.body.close(None);
228            return Err(Error::Cancelled);
229        }
230        if request.method == Method::CONNECT
231            || request.protocol.is_some()
232            || request.fields.contains("upgrade")
233        {
234            request.body.close(None);
235            return Err(Error::Unsupported("managed client tunnels and upgrades"));
236        }
237        if request
238            .scheme
239            .as_deref()
240            .is_some_and(|scheme| scheme != self.endpoint.scheme())
241        {
242            request.body.close(None);
243            return Err(Error::Unsupported(
244                "request scheme does not match the endpoint",
245            ));
246        }
247        request
248            .scheme
249            .get_or_insert_with(|| self.endpoint.request_scheme());
250        request
251            .authority
252            .get_or_insert_with(|| self.endpoint.authority.clone());
253        let local = Cancellation::new();
254        let mut guard = CancelOnDrop(local.clone(), true);
255        let (reply, response) = futures_channel::oneshot::channel();
256        let pending = match Pending::new(self.queue.clone()) {
257            Ok(pending) => pending,
258            Err(error) => {
259                request.body.close(None);
260                return Err(error);
261            }
262        };
263        let operation = Operation {
264            request: request.map_body(|body| outgoing::OutgoingBody::new(body, local.clone())),
265            interim: Box::new(interim),
266            reply: Some(reply),
267            cancellation: local,
268            lifetime: cancellation.clone(),
269            _pending: Some(pending),
270        };
271        // 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.
272        let _ = self.queue.push(operation);
273        let response = checked(
274            async { response.await.map_err(|_| Error::Closed)? },
275            cancellation,
276        )
277        .await?;
278        guard.1 = false;
279        Ok(response)
280    }
281}
282impl Clone for Client<'_> {
283    fn clone(&self) -> Self {
284        self.queue.handles.fetch_add(1, Ordering::Relaxed);
285        Self {
286            endpoint: self.endpoint.clone(),
287            queue: self.queue.clone(),
288            pool: self.pool.clone(),
289            context: self.context.clone(),
290        }
291    }
292}
293impl Drop for Client<'_> {
294    fn drop(&mut self) {
295        if self.queue.handles.fetch_sub(1, Ordering::AcqRel) == 1 {
296            self.queue.begin_close();
297        }
298    }
299}
base_http_client/lib.rs

base_http_client/
lib.rs
