gway 0.1.5

proxy
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123

use std::sync::Arc;

use http::{Request, Response, StatusCode, header, response::Builder};
use http_body::Body;
use http_body_util::{BodyExt, Full, combinators::BoxBody};
use hyper::body::Bytes;
use sub_host::sub_host;

use crate::{Error, IntoError, Result, Route, req_host, route::Protocol::H1};

pub static mut N: usize = 0;

pub async fn proxy<B>(req: Request<B>, route: Arc<Route>) -> Response<BoxBody<Bytes, hyper::Error>>
where
  B: Body<Data = Bytes> + Send + 'static,
  B::Error: IntoError + Send + Sync + 'static,
{
  let host = req_host(&req).to_owned();
  let path = req
    .uri()
    .path_and_query()
    .map(|x| x.as_str())
    .unwrap_or("")
    .to_owned();
  match _proxy(&host, &path, req, route).await {
    Ok(res) => {
      let status = res.status();
      log::info!("{status} {host} {path}");
      res
    }
    Err(err) => {
      let err = err.to_string();
      log::warn!("Error: {host} {path} {err}");
      response(|b| b.status(500), err).unwrap_or_default()
    }
  }
}

fn response(
  build: impl Fn(Builder) -> Builder,
  body: impl Into<Bytes>,
) -> Result<Response<BoxBody<Bytes, hyper::Error>>> {
  Ok(
    build(Builder::new()).body(
      Full::new(body.into())
        .map_err(|never| match never {})
        .boxed(),
    )?,
  )
}

pub async fn _proxy<B>(
  host: &str,
  path_and_query: &str,
  req: Request<B>,
  route: Arc<Route>,
) -> Result<Response<BoxBody<Bytes, hyper::Error>>>
where
  B: Body<Data = Bytes> + Send + 'static,
  B::Error: IntoError + Send + Sync + 'static,
{
  if let Some(site_conf) = route.host_conf.get(host) {
    let site_conf = site_conf.value();
    let upstream = &site_conf.upstream;
    let protocol = &upstream.protocol;
    let upstream_addr_li = &upstream.addr_li;
    let len = upstream_addr_li.len();
    if len == 0 {
      return Err(Error::UpstreamNotFound);
    }
    let (mut parts, body) = req.into_parts();

    match protocol {
      H1 => {
        parts.version = http::Version::HTTP_11;
        parts.headers.insert(
          header::CONNECTION,
          header::HeaderValue::from_static("keep-alive"),
        );
      }
    }
    let body = body.collect().await.map_err(|e| e.into_error())?.to_bytes();
    let mut pos = unsafe {
      N = N.overflowing_add(1).0;
      N
    } % len;
    let mut retry = 0;
    loop {
      let upstream_addr = upstream_addr_li[pos];
      let req = Request::from_parts(parts.clone(), Full::new(body.clone()));
      let r = match protocol {
        H1 => pooled_fetch::http(upstream_addr, req).await,
      };
      match r {
        Ok(res) => {
          return Ok(res.map(|b| b.boxed()));
        }
        Err(err) => {
          log::warn!("Error: {host} {path_and_query} {upstream_addr} {}", err);
          retry += 1;
          if retry > upstream.max_retry {
            return Err(err.into());
          }
          pos = (pos + 1) % len;
        }
      }
    }
  } else {
    if let Some(host) = sub_host(host)
      && route.host_conf.get(&faststr::FastStr::new(&host)).is_some()
    {
      return response(
        |b| {
          let new_uri = format!("https://{}{}", host, path_and_query);
          b.status(StatusCode::MOVED_PERMANENTLY)
            .header(header::LOCATION, new_uri)
        },
        &b""[..],
      );
    }
    response(|b| b.status(404), &b"404: Not Found"[..])
  }
}