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

base_http_client/
lib.rs
