pdk_classy/
client.rs

1// Copyright 2023 Salesforce, Inc. All rights reserved.
2use http::uri::Scheme;
3use std::collections::hash_map::Entry;
4use std::collections::HashMap;
5use std::fmt::{Debug, Display, Formatter};
6use std::str::FromStr;
7use std::{
8    convert::Infallible,
9    fmt,
10    future::Future,
11    marker::PhantomData,
12    rc::Rc,
13    task::{Poll, Waker},
14    time::Duration,
15};
16
17use crate::proxy_wasm::types::{Bytes, Status};
18use serde::de::StdError;
19
20use crate::http_constants::{
21    DEFAULT_TIMEOUT, HEADER_AUTHORITY, HEADER_METHOD, HEADER_PATH, HEADER_SCHEME, HEADER_STATUS,
22    METHOD_DELETE, METHOD_GET, METHOD_OPTIONS, METHOD_POST, METHOD_PUT, USER_AGENT_HEADER,
23};
24use crate::user_agent::UserAgent;
25use crate::{
26    event::EventKind,
27    extract::{Extract, FromContext},
28    host::Host,
29    reactor::root::{BoxedExtractor, RootReactor},
30    types::{Cid, RequestId},
31};
32
33#[derive(Clone, Debug, PartialEq, Eq)]
34pub struct HttpCallResponse {
35    pub request_id: RequestId,
36    pub num_headers: usize,
37    pub body_size: usize,
38    pub num_trailers: usize,
39}
40
41/// An asynchronous HTTP client to make Requests with.
42pub struct HttpClient {
43    reactor: Rc<RootReactor>,
44    host: Rc<dyn Host>,
45    user_agent: Rc<UserAgent>,
46}
47
48/// The Errors that may occur when processing a Request.
49#[derive(thiserror::Error, Debug, Clone)]
50pub enum HttpClientError {
51    /// Proxy status problem.
52    #[error("Proxy status problem: {0:?}")]
53    Status(Status),
54
55    /// Request awaited on create context event.
56    #[error("Request awaited on create context event")]
57    AwaitedOnCreateContext,
58}
59
60impl HttpClient {
61    pub(crate) fn new(
62        reactor: Rc<RootReactor>,
63        host: Rc<dyn Host>,
64        user_agent: Rc<UserAgent>,
65    ) -> Self {
66        Self {
67            reactor,
68            host,
69            user_agent,
70        }
71    }
72
73    /// Creates a request that will forward the whole response to the caller.
74    ///
75    /// Note: If you want to avoid reading body buffers in certain situations see the `extract_with` method.
76    pub fn request<'a>(
77        &'a self,
78        service: &'a Service,
79    ) -> RequestBuilder<'a, DefaultResponseExtractor> {
80        RequestBuilder::new(self, service, DefaultResponseExtractor)
81    }
82}
83
84impl<C> FromContext<C> for HttpClient
85where
86    Rc<dyn Host>: FromContext<C, Error = Infallible>,
87    Rc<RootReactor>: FromContext<C, Error = Infallible>,
88{
89    type Error = Infallible;
90
91    fn from_context(context: &C) -> Result<Self, Self::Error> {
92        let reactor = context.extract()?;
93        let host = context.extract()?;
94        let agent = context.extract()?;
95        Ok(Self::new(reactor, host, agent))
96    }
97}
98
99pub struct Request<T> {
100    reactor: Rc<RootReactor>,
101    request_id: RequestId,
102    cid_and_waker: Option<(Cid, Waker)>,
103    error: Option<HttpClientError>,
104    _response_type: PhantomData<T>,
105}
106
107/// An accessor for response parts.
108pub trait ResponseBuffers {
109    /// Returns the status code from a Response.
110    fn status_code(&self) -> u32;
111
112    /// Returns a header value by name if exists.
113    /// Known Limitations: The header value will be converted to an utf-8 String
114    /// If the bytes correspond to a non utf-8 string they will be parsed as an iso_8859_1 encoding.
115    fn header(&self, name: &str) -> Option<String>;
116
117    /// Returns a [`Vec`] containing all pairs of header names and their values.
118    /// Known Limitations: The header values will be converted to utf-8 Strings
119    /// If the bytes correspond to a non utf-8 string they will be parsed as an iso_8859_1 encoding.
120    fn headers(&self) -> Vec<(String, String)>;
121
122    /// Returns the body in binary format.
123    fn body(&self, start: usize, max_size: usize) -> Option<Bytes>;
124
125    /// Returns a [`Vec`] containing all pairs of trailer names and their values.
126    fn trailers(&self) -> Vec<(String, String)>;
127}
128
129impl ResponseBuffers for Rc<dyn Host> {
130    fn status_code(&self) -> u32 {
131        self.header(HEADER_STATUS)
132            .and_then(|status| status.parse::<u32>().ok())
133            .unwrap_or_default()
134    }
135
136    fn header(&self, name: &str) -> Option<String> {
137        self.get_http_call_response_header(name)
138    }
139
140    fn headers(&self) -> Vec<(String, String)> {
141        self.get_http_call_response_headers()
142    }
143
144    fn body(&self, start: usize, max_size: usize) -> Option<Bytes> {
145        self.get_http_call_response_body(start, max_size)
146    }
147
148    fn trailers(&self) -> Vec<(String, String)> {
149        self.get_http_call_response_trailers()
150    }
151}
152
153/// A low-level trait for extracting a Response and convert it to a
154/// [`ResponseExtractor::Output`] type.
155pub trait ResponseExtractor {
156    /// The output type
157    type Output;
158
159    /// Extracts the Response from their low-level components
160    ///
161    fn extract(self, event: &HttpCallResponse, buffers: &dyn ResponseBuffers) -> Self::Output;
162}
163
164pub struct FnResponseExtractor<F> {
165    function: F,
166}
167
168impl<F, T> ResponseExtractor for FnResponseExtractor<F>
169where
170    F: FnOnce(&HttpCallResponse, &dyn ResponseBuffers) -> T,
171{
172    type Output = T;
173
174    fn extract(self, event: &HttpCallResponse, buffers: &dyn ResponseBuffers) -> Self::Output {
175        (self.function)(event, buffers)
176    }
177}
178
179impl<F, T> FnResponseExtractor<F>
180where
181    F: FnOnce(&HttpCallResponse, &dyn ResponseBuffers) -> T,
182{
183    pub fn from_fn(function: F) -> FnResponseExtractor<F>
184    where
185        F: FnOnce(&HttpCallResponse, &dyn ResponseBuffers) -> T,
186    {
187        FnResponseExtractor { function }
188    }
189}
190
191pub struct RequestBuilder<'a, E> {
192    client: &'a HttpClient,
193    extractor: E,
194    service: &'a Service,
195    path: Option<&'a str>,
196    headers: Option<Vec<(&'a str, &'a str)>>,
197    body: Option<&'a [u8]>,
198    trailers: Option<Vec<(&'a str, &'a str)>>,
199    timeout: Option<Duration>,
200}
201
202impl<'a, E> RequestBuilder<'a, E>
203where
204    E: ResponseExtractor + 'static,
205    E::Output: 'static,
206{
207    fn new(client: &'a HttpClient, service: &'a Service, extractor: E) -> Self {
208        Self {
209            client,
210            extractor,
211            service,
212            path: None,
213            headers: None,
214            body: None,
215            trailers: None,
216            timeout: None,
217        }
218    }
219
220    pub fn extractor<T>(self, extractor: T) -> RequestBuilder<'a, T>
221    where
222        T: ResponseExtractor,
223    {
224        RequestBuilder {
225            client: self.client,
226            extractor,
227            service: self.service,
228            path: self.path,
229            headers: self.headers,
230            body: self.body,
231            trailers: self.trailers,
232            timeout: self.timeout,
233        }
234    }
235
236    pub fn extract_with<F, T>(self, function: F) -> RequestBuilder<'a, FnResponseExtractor<F>>
237    where
238        F: FnOnce(&HttpCallResponse, &dyn ResponseBuffers) -> T,
239    {
240        self.extractor(FnResponseExtractor::from_fn(function))
241    }
242
243    pub fn path(mut self, path: &'a str) -> Self {
244        self.path = Some(path);
245        self
246    }
247
248    pub fn headers(mut self, headers: Vec<(&'a str, &'a str)>) -> Self {
249        self.headers = Some(headers);
250        self
251    }
252
253    pub fn body(mut self, body: &'a [u8]) -> Self {
254        self.body = Some(body);
255        self
256    }
257
258    pub fn trailers(mut self, trailers: Vec<(&'a str, &'a str)>) -> Self {
259        self.trailers = Some(trailers);
260        self
261    }
262
263    pub fn timeout(mut self, timeout: Duration) -> Self {
264        self.timeout = Some(timeout);
265        self
266    }
267
268    pub fn post(self) -> Request<E::Output> {
269        self.send(METHOD_POST)
270    }
271    pub fn put(self) -> Request<E::Output> {
272        self.send(METHOD_PUT)
273    }
274    pub fn get(self) -> Request<E::Output> {
275        self.send(METHOD_GET)
276    }
277    pub fn options(self) -> Request<E::Output> {
278        self.send(METHOD_OPTIONS)
279    }
280    pub fn delete(self) -> Request<E::Output> {
281        self.send(METHOD_DELETE)
282    }
283
284    #[must_use]
285    pub fn send(mut self, method: &str) -> Request<E::Output> {
286        let mut headers = self.headers.take().unwrap_or_default();
287
288        headers.push((HEADER_PATH, self.path.unwrap_or(self.service.uri().path())));
289        headers.push((HEADER_AUTHORITY, self.service.uri().authority()));
290        headers.push((HEADER_METHOD, method));
291        headers.push((USER_AGENT_HEADER, self.client.user_agent.value()));
292        headers.push((HEADER_SCHEME, self.service.uri().scheme()));
293
294        let body = self.body.take();
295        let trailers = self.trailers.take().unwrap_or_default();
296        let timeout = self.timeout.take().unwrap_or(DEFAULT_TIMEOUT);
297
298        match self.client.host.dispatch_http_call(
299            self.service.cluster_name(),
300            headers,
301            body,
302            trailers,
303            timeout,
304        ) {
305            Ok(request_id) => {
306                let request_id: RequestId = request_id.into();
307                let extractor = boxed_extractor(self.client.host.clone(), self.extractor);
308                self.client.reactor.insert_extractor(request_id, extractor);
309                Request::new(self.client.reactor.clone(), request_id)
310            }
311            Err(err) => Request::error(self.client.reactor.clone(), HttpClientError::Status(err)),
312        }
313    }
314}
315
316impl<E: ResponseExtractor> ResponseExtractor for RequestBuilder<'_, E> {
317    type Output = E::Output;
318
319    fn extract(self, event: &HttpCallResponse, buffers: &dyn ResponseBuffers) -> Self::Output {
320        self.extractor.extract(event, buffers)
321    }
322}
323
324fn boxed_extractor<E>(buffers: Rc<dyn Host>, extractor: E) -> BoxedExtractor
325where
326    E: ResponseExtractor + 'static,
327    E::Output: 'static,
328{
329    Box::new(move |event| Box::new(extractor.extract(event, &buffers)))
330}
331
332pub struct EmptyResponseExtractor;
333
334impl ResponseExtractor for EmptyResponseExtractor {
335    type Output = ();
336
337    fn extract(self, _event: &HttpCallResponse, _buffers: &dyn ResponseBuffers) -> Self::Output {}
338}
339
340impl<T> Request<T> {
341    fn new(reactor: Rc<RootReactor>, request_id: RequestId) -> Self {
342        Request {
343            reactor,
344            request_id,
345            error: None,
346            cid_and_waker: None,
347            _response_type: PhantomData,
348        }
349    }
350
351    fn error(reactor: Rc<RootReactor>, error: HttpClientError) -> Self {
352        Request {
353            reactor,
354            request_id: RequestId::from(0),
355            error: Some(error),
356            cid_and_waker: None,
357            _response_type: PhantomData,
358        }
359    }
360
361    pub fn id(&self) -> RequestId {
362        self.request_id
363    }
364}
365
366impl<T> Drop for Request<T> {
367    fn drop(&mut self) {
368        if self.error.is_none() {
369            let reactor = self.reactor.as_ref();
370
371            // Ensure that all related objects were removed
372            reactor.remove_extractor(self.request_id);
373            reactor.remove_response(self.request_id);
374            reactor.remove_client(self.request_id);
375        }
376    }
377}
378
379impl<T: Unpin + 'static> Future for Request<T> {
380    type Output = Result<T, HttpClientError>;
381
382    fn poll(
383        mut self: std::pin::Pin<&mut Self>,
384        cx: &mut std::task::Context<'_>,
385    ) -> Poll<Self::Output> {
386        if let Some(error) = self.error.clone() {
387            return Poll::Ready(Err(error));
388        }
389
390        if self.reactor.current_event() == Some(EventKind::Start) {
391            return Poll::Ready(Err(HttpClientError::AwaitedOnCreateContext));
392        }
393
394        if let Some((_event, content)) = self.reactor.remove_response(self.request_id) {
395            // It should be safe to unwrap here
396            let content = content.expect("response content should have been extracted");
397
398            // It should be safe to unwrap here
399            let content = content.downcast().expect("downcasting");
400
401            Poll::Ready(Ok(*content))
402        } else {
403            let this = &mut *self.as_mut();
404            match this.cid_and_waker.as_ref() {
405                None => {
406                    let cid = this.reactor.active_cid();
407
408                    // Register the waker in the reactor.
409                    this.reactor
410                        .insert_client(this.request_id, cx.waker().clone());
411                    this.reactor.set_paused(cid, true);
412                    this.cid_and_waker = Some((cid, cx.waker().clone()));
413                }
414                Some((cid, waker)) if !waker.will_wake(cx.waker()) => {
415                    // Deregister the waker from the reactor to remove the old waker.
416                    let _ = this
417                        .reactor
418                        .remove_client(this.request_id)
419                        // It should be safe to unwrap here
420                        .expect("stored extractor");
421
422                    // Register the waker in the reactor with the new waker.
423                    this.reactor
424                        .insert_client(this.request_id, cx.waker().clone());
425                    this.cid_and_waker = Some((*cid, cx.waker().clone()));
426                }
427                Some(_) => {}
428            }
429            Poll::Pending
430        }
431    }
432}
433
434pub struct DefaultResponseExtractor;
435
436impl ResponseExtractor for DefaultResponseExtractor {
437    type Output = HttpClientResponse;
438
439    fn extract(self, event: &HttpCallResponse, buffers: &dyn ResponseBuffers) -> Self::Output {
440        let mut map = HashMap::new();
441        for (k, v) in buffers.headers().into_iter() {
442            match map.entry(k) {
443                Entry::Vacant(e) => {
444                    e.insert(v);
445                }
446                Entry::Occupied(mut e) => {
447                    e.insert(format!("{},{}", e.get(), v));
448                }
449            }
450        }
451
452        let body = buffers.body(0, event.body_size).unwrap_or_default();
453
454        HttpClientResponse::new(map, body)
455    }
456}
457
458/// Response from the [`HttpClient`].
459#[derive(Debug)]
460pub struct HttpClientResponse {
461    headers: HashMap<String, String>,
462    body: Bytes,
463}
464
465impl HttpClientResponse {
466    pub fn new(headers: HashMap<String, String>, body: Bytes) -> Self {
467        Self { headers, body }
468    }
469
470    /// Returns the status code.
471    pub fn status_code(&self) -> u32 {
472        self.header(HEADER_STATUS)
473            .and_then(|status| status.parse::<u32>().ok())
474            .unwrap_or_default()
475    }
476
477    /// Returns a list of headers grouped by name and value.
478    pub fn headers(&self) -> &HashMap<String, String> {
479        &self.headers
480    }
481
482    /// Returns a header by name if exists.
483    pub fn header(&self, header: &str) -> Option<&String> {
484        self.headers.get(header)
485    }
486
487    /// Returns the body in binary format.
488    pub fn body(&self) -> &[u8] {
489        self.body.as_slice()
490    }
491
492    /// Returns the body in [`String`] format.
493    pub fn as_utf8_lossy(&self) -> String {
494        String::from_utf8_lossy(&self.body).to_string()
495    }
496}
497
498/// Represents an invalid URI error.
499pub struct InvalidUri(InvalidUriKind);
500
501enum InvalidUriKind {
502    Delegate(http::uri::InvalidUri),
503    MissingAuthority,
504    InvalidSchema,
505}
506
507impl Display for InvalidUri {
508    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
509        match &self.0 {
510            InvalidUriKind::Delegate(d) => Display::fmt(d, f),
511            InvalidUriKind::MissingAuthority => Display::fmt("authority missing", f),
512            InvalidUriKind::InvalidSchema => Display::fmt("scheme not supported", f),
513        }
514    }
515}
516
517impl Debug for InvalidUri {
518    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
519        Display::fmt(&self, f)
520    }
521}
522
523impl StdError for InvalidUri {}
524
525/// Represents a URI.
526#[derive(Clone, Debug)]
527pub struct Uri {
528    delegate: http::Uri,
529}
530
531impl Uri {
532    /// Returns the URI path.
533    pub fn path(&self) -> &str {
534        self.delegate
535            .path_and_query()
536            .map(|path_and_query| path_and_query.as_str())
537            .unwrap_or_else(|| self.delegate.path())
538    }
539
540    /// Returns the URI schema.
541    pub fn scheme(&self) -> &str {
542        // The unwrap should never take effect since we don't allow construction without scheme
543        self.delegate.scheme_str().unwrap_or_default()
544    }
545
546    /// Returns the URI authority.
547    pub fn authority(&self) -> &str {
548        // The unwrap should never take effect since we don't allow construction without authority
549        self.delegate
550            .authority()
551            .map(|authority| authority.as_str())
552            .unwrap_or_default()
553    }
554}
555
556impl Display for Uri {
557    fn fmt(&self, f: &mut Formatter) -> fmt::Result {
558        f.write_fmt(format_args!("{}", self.delegate.to_string().as_str()))
559    }
560}
561
562impl FromStr for Uri {
563    type Err = InvalidUri;
564
565    fn from_str(s: &str) -> Result<Self, Self::Err> {
566        match s.parse::<http::Uri>() {
567            Ok(delegate) => {
568                if delegate.authority().is_none() {
569                    return Err(InvalidUri(InvalidUriKind::MissingAuthority));
570                }
571
572                if delegate
573                    .scheme()
574                    .map(|s| {
575                        !s.eq(&Scheme::HTTP)
576                            && !s.eq(&Scheme::HTTPS)
577                            && !s.as_str().eq_ignore_ascii_case("h2")
578                    })
579                    .unwrap_or(true)
580                {
581                    return Err(InvalidUri(InvalidUriKind::InvalidSchema));
582                }
583
584                Ok(Self { delegate })
585            }
586            Err(e) => Err(InvalidUri(InvalidUriKind::Delegate(e))),
587        }
588    }
589}
590
591#[derive(Clone, Debug)]
592pub struct Service {
593    cluster_name: String,
594    uri: Uri,
595}
596
597impl Service {
598    pub fn from<'a>(name: &'a str, namespace: &'a str, uri: Uri) -> Service {
599        let cluster_name = format!("{}.{}.svc", name, namespace);
600        Service { cluster_name, uri }
601    }
602
603    pub fn new(cluster_name: &str, uri: Uri) -> Service {
604        Service {
605            cluster_name: cluster_name.to_string(),
606            uri,
607        }
608    }
609
610    pub fn cluster_name(&self) -> &str {
611        self.cluster_name.as_str()
612    }
613
614    pub fn uri(&self) -> &Uri {
615        &self.uri
616    }
617}
618
619#[cfg(test)]
620mod test {
621    use super::Uri;
622
623    #[test]
624    fn successfully_parse_http() {
625        assert!("http://some.com/foo?some=val".parse::<Uri>().is_ok());
626    }
627
628    #[test]
629    fn successfully_parse_https() {
630        assert!("https://some.com/foo".parse::<Uri>().is_ok());
631    }
632
633    #[test]
634    fn successfully_parse_h2() {
635        assert!("h2://some.com/foo".parse::<Uri>().is_ok());
636    }
637
638    #[test]
639    fn error_invalid_scheme() {
640        assert!("ftp://some.com/foo".parse::<Uri>().is_err());
641    }
642
643    #[test]
644    fn error_on_missing_scheme() {
645        assert!("some.com/foo".parse::<Uri>().is_err());
646    }
647
648    #[test]
649    fn error_on_missing_host() {
650        assert!("/foo".parse::<Uri>().is_err());
651    }
652}
pdk_classy/client.rs

pdk_classy/
client.rs
