Line data    Source code

       1              : use http::header::CONTENT_TYPE;
       2              : use http::{Method, Request, Response};
       3              : 
       4              : use crate::grpc_proxy_globals::{
       5              :     call_grpc_unary, create_grpc_channel, delete_grpc_channel, delete_grpc_server_stream,
       6              :     read_grpc_server_stream, start_grpc_server_stream,
       7              : };
       8              : use crate::grpc_proxy_routes::{parse_grpc_proxy_path, GrpcProxyRoute};
       9              : use crate::http_proxy_globals::{delete_http_server_stream, read_http_server_stream, start_http_server_stream};
      10              : use crate::http_proxy_routes::{parse_http_proxy_path, HttpProxyRoute};
      11              : 
      12            4 : fn not_found(message: String) -> Response<Vec<u8>> {
      13            4 :     Response::builder()
      14            4 :         .status(404)
      15            4 :         .header(CONTENT_TYPE, mime::TEXT_PLAIN.essence_str())
      16            4 :         .body(message.into_bytes())
      17            4 :         .unwrap()
      18            4 : }
      19              : 
      20          105 : pub async fn route_proxy_request(mut request: Request<Vec<u8>>) -> Response<Vec<u8>> {
      21          105 :     if request.method() == Method::GET && request.uri().path() == "/health" {
      22            1 :         return Response::builder()
      23            1 :             .status(200)
      24            1 :             .header(CONTENT_TYPE, mime::APPLICATION_JSON.essence_str())
      25            1 :             .body(br#"{"status":"ok"}"#.to_vec())
      26            1 :             .unwrap();
      27          104 :     }
      28              : 
      29          104 :     if let Some(route) = parse_http_proxy_path(request.uri().path()) {
      30           51 :         return match (request.method().clone(), route) {
      31           24 :             (Method::POST, HttpProxyRoute::Streams {}) => start_http_server_stream(std::mem::take(&mut request)).await,
      32           20 :             (Method::GET, HttpProxyRoute::Stream { stream_id }) => {
      33           20 :                 read_http_server_stream(stream_id, std::mem::take(&mut request)).await
      34              :             }
      35            5 :             (Method::DELETE, HttpProxyRoute::Stream { stream_id }) => {
      36            5 :                 delete_http_server_stream(stream_id, std::mem::take(&mut request)).await
      37              :             }
      38            2 :             _ => not_found(format!("cannot find handler for http proxy route={:?}, method={:?}", request.uri().path(), request.method())),
      39              :         };
      40           53 :     }
      41              : 
      42           53 :     if let Some(route) = parse_grpc_proxy_path(request.uri().path()) {
      43           51 :         return match (request.method().clone(), route) {
      44           19 :             (Method::POST, GrpcProxyRoute::Channels) => create_grpc_channel(std::mem::take(&mut request)).await,
      45            1 :             (Method::DELETE, GrpcProxyRoute::Channel { channel_id }) => delete_grpc_channel(channel_id).await,
      46            7 :             (Method::POST, GrpcProxyRoute::ChannelUnary { channel_id }) => {
      47            7 :                 call_grpc_unary(channel_id, std::mem::take(&mut request)).await
      48              :             }
      49            8 :             (Method::POST, GrpcProxyRoute::ChannelStreams { channel_id }) => {
      50            8 :                 start_grpc_server_stream(channel_id, std::mem::take(&mut request)).await
      51              :             }
      52           12 :             (Method::GET, GrpcProxyRoute::ChannelStream { channel_id, stream_id }) => {
      53           12 :                 read_grpc_server_stream(channel_id, stream_id, std::mem::take(&mut request)).await
      54              :             }
      55            2 :             (Method::DELETE, GrpcProxyRoute::ChannelStream { channel_id, stream_id }) => {
      56            2 :                 delete_grpc_server_stream(channel_id, stream_id, std::mem::take(&mut request)).await
      57              :             }
      58            2 :             _ => not_found(format!("cannot find handler for grpc proxy route={:?}, method={:?}", request.uri().path(), request.method())),
      59              :         };
      60            2 :     }
      61              : 
      62            2 :     Response::builder()
      63            2 :         .status(400)
      64            2 :         .header(CONTENT_TYPE, mime::TEXT_PLAIN.essence_str())
      65            2 :         .body(b"cannot find route for request path".to_vec())
      66            2 :         .unwrap()
      67          105 : }