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