1use std::marker::PhantomData;
2use std::pin::Pin;
3use std::task::Poll;
4use std::time::{Duration, Instant};
5
6use cartel_core::{Extract, Registrable, Reply, Slot};
7use dope::driver::token::Token;
8use dope::manifold::connector::Connector;
9use dope::manifold::connector::source::Dialer;
10use dope::manifold::env::Env;
11use dope_fiber::{Either, Fiber, TimerExt as _, poll_fn, race, wait_fn};
12use dope_net::Transport;
13use http::Method;
14use o3::buffer::{Lease, Pool};
15use o3::cell::RegionToken;
16use sark_core::http::Response;
17
18use crate::connector::error::Error;
19use crate::connector::redirect::RedirectState;
20use crate::connector::session::{Outcome, Port, Session};
21
22struct ExtractResponse;
23
24type HandleMarker<'a, S, E> = PhantomData<(&'a (), fn() -> (S, E))>;
25
26impl Extract<Outcome> for ExtractResponse {
27 type Output = Outcome;
28
29 fn extract(slot: &mut Slot<Outcome>) -> Option<Self::Output> {
30 if !slot.completed() {
31 return None;
32 }
33 if slot.take_overflow() {
34 return Some(Err(Error::CapacityOverflow));
35 }
36 Some(slot.pop().unwrap_or(Err(Error::Closed)))
37 }
38}
39
40pub struct HttpHandle<'a, 'd, const ID: u8, S, E> {
41 port: &'d Port<'d>,
42 marker: HandleMarker<'a, S, E>,
43}
44
45impl<S, E, const ID: u8> Copy for HttpHandle<'_, '_, ID, S, E> {}
46
47impl<S, E, const ID: u8> Clone for HttpHandle<'_, '_, ID, S, E> {
48 fn clone(&self) -> Self {
49 *self
50 }
51}
52
53impl<'a, 'd, const ID: u8, S, E> HttpHandle<'a, 'd, ID, S, E>
54where
55 S: Dialer<E::Transport> + 'd,
56 E: Env + 'd,
57 E::Transport: Transport<Addr: Clone>,
58{
59 pub fn from_port(port: &'d Port<'d>) -> Self {
60 Self {
61 port,
62 marker: PhantomData,
63 }
64 }
65
66 pub fn from_cell(conn: Pin<&Connector<'d, ID, Session<'d>, S, E>>) -> Self {
67 Self::from_port(conn.get_ref().session().port)
68 }
69
70 pub fn sleep(
71 &self,
72 duration: Duration,
73 ) -> impl Fiber<'d, Output = ()> + 'd + use<'d, ID, S, E> {
74 let timer: &'d _ = self.port.timer();
75 timer.sleep(duration)
76 }
77
78 pub fn connection_count(&self) -> usize {
79 self.port.shared.connection_count()
80 }
81
82 pub fn wait_active<'b>(&'b self) -> impl Fiber<'d, Output = Result<(), Error>> + 'b {
83 let handle = self;
84 wait_fn(move |cx, waiter| {
85 let shared = &handle.port.shared;
86 if shared.has_connection() {
87 return Poll::Ready(Ok(()));
88 }
89 if !shared.try_register_active(waiter, cx.as_ref()) {
90 return Poll::Ready(Err(Error::Backpressure));
91 }
92 if shared.has_connection() {
93 shared.wake();
94 return Poll::Ready(Ok(()));
95 }
96 Poll::Pending
97 })
98 }
99
100 pub fn host<'b>(&'b self) -> impl Fiber<'d, Output = String> + 'b {
101 let handle = self;
102 poll_fn(move |_cx| Poll::Ready(handle.port.shared.host.clone()))
103 }
104
105 pub fn get<'b>(
106 &'b self,
107 path: &'b str,
108 ) -> impl Fiber<'d, Output = Result<Response, Error>> + 'b {
109 self.send(Method::GET, path, &[])
110 }
111
112 pub fn send<'b>(
113 &'b self,
114 method: Method,
115 path: &'b str,
116 body: &'b [u8],
117 ) -> impl Fiber<'d, Output = Result<Response, Error>> + 'b {
118 self.send_with_headers(method, path, &[], body)
119 }
120
121 pub fn send_with_headers<'b>(
122 &'b self,
123 method: Method,
124 path: &'b str,
125 headers: &'b [(&'b str, &'b str)],
126 body: &'b [u8],
127 ) -> impl Fiber<'d, Output = Result<Response, Error>> + 'b {
128 let handle = *self;
129 let validation = <str as HeaderField>::validate_all(headers);
130 let max_redirects = handle.port.shared.max_redirects;
131 let origin = &handle.port.shared.origin;
132 dope_fiber::fiber!('d => async move {
133 validation?;
134 let mut method = method;
135 let mut body = body;
136 let mut response = handle
137 .dispatch_with_retry(&method, path, headers, body)
138 .await?;
139 if !response.status().is_redirection() {
140 return Ok(response);
141 }
142 let mut redirects = Box::new(RedirectState::new(max_redirects, origin, path)?);
143 loop {
144 let status = response.status().as_u16();
145 let location = response
146 .headers()
147 .get("location")
148 .and_then(|value| value.to_str().ok())
149 .ok_or_else(|| Error::Http("redirect without Location header".into()))?;
150 method = redirects.advance(status, location, &method)?;
151 if method == Method::GET {
152 body = &[];
153 }
154 response = handle
155 .dispatch_with_retry(&method, redirects.path_and_query(), headers, body)
156 .await?;
157 if !response.status().is_redirection() {
158 return Ok(response);
159 }
160 }
161 })
162 }
163
164 fn dispatch_with_retry<'b>(
165 self,
166 method: &'b Method,
167 path: &'b str,
168 headers: &'b [(&'b str, &'b str)],
169 body: &'b [u8],
170 ) -> impl Fiber<'d, Output = Result<Response, Error>> + 'b
171 where
172 Self: 'b,
173 {
174 let handle = self;
175 let retry = handle.port.shared.retry;
176 let timer: &'d _ = handle.port.timer();
177 dope_fiber::fiber!('d => async move {
178 let attempts = retry.attempts(method);
179 let mut attempt = 0;
180 loop {
181 match handle.dispatch_once(method, path, headers, body).await {
182 Ok(response) => return Ok(response),
183 Err(error)
184 if retry.should_retry(method, &error)
185 && attempt + 1 < attempts =>
186 {
187 attempt += 1;
188 let backoff = Duration::from_millis(25 * u64::from(attempt));
189 timer.sleep(backoff).await;
190 }
191 Err(error) => return Err(error),
192 }
193 }
194 })
195 }
196
197 fn dispatch_once<'b>(
198 self,
199 method: &'b Method,
200 path: &'b str,
201 headers: &'b [(&'b str, &'b str)],
202 body: &'b [u8],
203 ) -> impl Fiber<'d, Output = Result<Response, Error>> + 'b
204 where
205 Self: 'b,
206 {
207 let handle = self;
208 dope_fiber::fiber!('d => async move {
209 let shared = &handle.port.shared;
210 let request_timeout = shared.request_timeout;
211 let request = Encode::request(
212 handle.port.requests.as_ref(),
213 method,
214 path,
215 &shared.host,
216 headers,
217 body,
218 )?;
219 let mut request = Some(request);
220 let acquire = wait_fn(move |mut cx, waiter| {
221 let now = Instant::now();
222 let shared = &handle.port.shared;
223 let idle = shared.idle_timeout;
224 let chosen = shared.acquire(cx.as_mut().region_token(), now, idle, |token| {
225 handle.port.io.close(token)
226 });
227 match chosen {
228 Some(token) => {
229 let req = request.take().expect("dispatch enqueue polled twice");
230 Poll::Ready(
231 Enqueue::submit(handle, token, req, cx.as_mut().region_token())
232 .map(|reply| (token, reply)),
233 )
234 }
235 None => {
236 if !shared.try_register_active(waiter, cx.as_ref()) {
237 return Poll::Ready(Err(Error::Backpressure));
238 }
239 let chosen = shared.acquire(
240 cx.as_mut().region_token(),
241 now,
242 idle,
243 |token| handle.port.io.close(token),
244 );
245 match chosen {
246 Some(token) => {
247 shared.wake();
248 let req = request.take().expect("dispatch enqueue polled twice");
249 Poll::Ready(
250 Enqueue::submit(
251 handle,
252 token,
253 req,
254 cx.as_mut().region_token(),
255 )
256 .map(|reply| (token, reply)),
257 )
258 }
259 None => Poll::Pending,
260 }
261 }
262 }
263 });
264 let acquire_deadline = handle.sleep(request_timeout);
265 let (conn_id, reply) = match race(acquire, acquire_deadline).await {
266 Either::Left(result) => result?,
267 Either::Right(()) => return Err(Error::Timeout),
268 };
269
270 let reply_deadline = handle.sleep(request_timeout);
271 match race(reply, reply_deadline).await {
272 Either::Left(outcome) => outcome,
273 Either::Right(()) => {
274 handle.port.io.close(conn_id);
275 Err(Error::Timeout)
276 }
277 }
278 })
279 }
280}
281
282struct Enqueue;
283
284impl Enqueue {
285 fn submit<'a, 'd, const ID: u8, S, E>(
286 handle: HttpHandle<'a, 'd, ID, S, E>,
287 conn_id: Token,
288 request: Lease<'d>,
289 region: &mut RegionToken<'d>,
290 ) -> Result<Reply<'d, Outcome, ExtractResponse>, Error>
291 where
292 S: Dialer<E::Transport> + 'd,
293 E: Env + 'd,
294 E::Transport: Transport<Addr: Clone>,
295 {
296 let shared = &handle.port.shared;
297 if !handle.port.io.is_active(conn_id) {
298 shared.close_connection(region, conn_id);
299 return Err(Error::NotConnected);
300 }
301 let arena = shared.arena(conn_id).ok_or(Error::NotConnected)?;
302 if !arena.can_register(region) {
303 return Err(Error::Backpressure);
304 }
305 if handle.port.io.try_enqueue(conn_id, request).is_err() {
306 shared.make_available(region, conn_id);
307 return Err(Error::Backpressure);
308 }
309 let mut reply = Reply::new();
310 assert!(reply.try_attach(region, arena));
311 shared.submitted(region, conn_id, Instant::now());
312 Ok(reply)
313 }
314}
315
316struct Encode;
317
318impl Encode {
319 fn request<'d>(
320 pool: Pin<&'d Pool>,
321 method: &Method,
322 path: &str,
323 host: &str,
324 headers: &[(&str, &str)],
325 body: &[u8],
326 ) -> Result<Lease<'d>, Error> {
327 let mut buf = pool.try_acquire().ok_or(Error::Backpressure)?;
328 let initial: [&[u8]; 6] = [
329 method.as_str().as_bytes(),
330 b" ",
331 path.as_bytes(),
332 b" HTTP/1.1\r\nHost: ",
333 host.as_bytes(),
334 b"\r\nConnection: keep-alive\r\nAccept: \x2a/\x2a\r\n",
335 ];
336 let mut value = body.len();
337 let mut digits = [0; 20];
338 let mut cursor = digits.len();
339 if !body.is_empty() {
340 loop {
341 cursor -= 1;
342 digits[cursor] = b'0' + (value % 10) as u8;
343 value /= 10;
344 if value == 0 {
345 break;
346 }
347 }
348 }
349 let content_length: [&[u8]; 3] = if body.is_empty() {
350 [&[], &[], &[]]
351 } else {
352 [b"Content-Length: ", &digits[cursor..], b"\r\n"]
353 };
354 buf.try_extend_from_slices(initial)
355 .map_err(|_| Error::Backpressure)?;
356 for (name, value) in headers {
357 buf.try_extend_from_slices([name.as_bytes(), b": ", value.as_bytes(), b"\r\n"])
358 .map_err(|_| Error::Backpressure)?;
359 }
360 buf.try_extend_from_slices(content_length)
361 .map_err(|_| Error::Backpressure)?;
362 buf.try_extend_from_slices([b"\r\n", body])
363 .map_err(|_| Error::Backpressure)?;
364 Ok(buf)
365 }
366}
367
368trait HeaderField {
369 fn validate_all(headers: &[(&str, &str)]) -> Result<(), Error>;
370 fn is_valid_name(&self) -> bool;
371 fn is_valid_value(&self) -> bool;
372 fn is_reserved_name(&self) -> bool;
373}
374
375impl HeaderField for str {
376 fn validate_all(headers: &[(&str, &str)]) -> Result<(), Error> {
377 for (name, value) in headers {
378 if !name.is_valid_name() {
379 return Err(Error::Http("invalid request header name".into()));
380 }
381 if !value.is_valid_value() {
382 return Err(Error::Http("invalid request header value".into()));
383 }
384 if name.is_reserved_name() {
385 return Err(Error::Http("reserved request header".into()));
386 }
387 }
388 Ok(())
389 }
390
391 fn is_valid_name(&self) -> bool {
392 !self.is_empty()
393 && self.bytes().all(|b| {
394 matches!(
395 b,
396 b'!' | b'#'
397 | b'$'
398 | b'%'
399 | b'&'
400 | b'\''
401 | b'*'
402 | b'+'
403 | b'-'
404 | b'.'
405 | b'^'
406 | b'_'
407 | b'`'
408 | b'|'
409 | b'~'
410 | b'0'..=b'9'
411 | b'A'..=b'Z'
412 | b'a'..=b'z'
413 )
414 })
415 }
416
417 fn is_valid_value(&self) -> bool {
418 self.bytes()
419 .all(|b| b == b'\t' || (0x20..=0x7e).contains(&b))
420 }
421
422 fn is_reserved_name(&self) -> bool {
423 self.eq_ignore_ascii_case("host")
424 || self.eq_ignore_ascii_case("connection")
425 || self.eq_ignore_ascii_case("content-length")
426 }
427}