tuxmain/rust-rpxy
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix: fix message handler (changed this inside Arc)

Browse source
This commit is contained in:
Jun Kurihara 2023-08-09 11:31:38 +09:00
commit 3d60175c11
No known key found for this signature in database
GPG key ID: 6D3FEE70E498C15B
7 changed files with 32 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

1
TODO.md
View file

@ -2,6 +2,7 @@
- [Done in 0.6.0] ~~Fix strategy for `h2c` requests on forwarded requests upstream. This needs to update forwarder definition. Also, maybe forwarder would have a cache corresponding to the following task.~~ - [Done in 0.6.0] ~~Fix strategy for `h2c` requests on forwarded requests upstream. This needs to update forwarder definition. Also, maybe forwarder would have a cache corresponding to the following task.~~
- [Try in v0.6.0] **Cache option for the response with `Cache-Control: public` header directive ([#55](https://github.com/junkurihara/rust-rpxy/issues/55))** - [Try in v0.6.0] **Cache option for the response with `Cache-Control: public` header directive ([#55](https://github.com/junkurihara/rust-rpxy/issues/55))**
- Fix dynamic reloading of configuration file
- Improvement of path matcher - Improvement of path matcher
- More flexible option for rewriting path - More flexible option for rewriting path
- Refactoring - Refactoring

2
rpxy-bin/Cargo.toml
View file

@ -46,7 +46,7 @@ tracing-subscriber = { version = "0.3.17", features = ["env-filter"] }
[target.'cfg(not(target_env = "msvc"))'.dependencies] [target.'cfg(not(target_env = "msvc"))'.dependencies]
tikv-jemallocator = "0.5.0" tikv-jemallocator = "0.5.4"
[dev-dependencies] [dev-dependencies]

2
rpxy-lib/src/handler/handler_main.rs
View file

@ -50,7 +50,7 @@ where
/// Handle incoming request message from a client /// Handle incoming request message from a client
pub async fn handle_request( pub async fn handle_request(
self, &self,
mut req: Request<Body>, mut req: Request<Body>,
client_addr: SocketAddr, // アクセス制御用 client_addr: SocketAddr, // アクセス制御用
listen_addr: SocketAddr, listen_addr: SocketAddr,

6
rpxy-lib/src/lib.rs
View file

@ -69,10 +69,12 @@ where
}); });
// TODO: HTTP2 only client is needed for http2 cleartext case // TODO: HTTP2 only client is needed for http2 cleartext case
let msg_handler = HttpMessageHandlerBuilder::default() let msg_handler = Arc::new(
HttpMessageHandlerBuilder::default()
.forwarder(Arc::new(Forwarder::new().await)) .forwarder(Arc::new(Forwarder::new().await))
.globals(globals.clone()) .globals(globals.clone())
.build()?; .build()?,
);
let addresses = globals.proxy_config.listen_sockets.clone(); let addresses = globals.proxy_config.listen_sockets.clone();
let futures = select_all(addresses.into_iter().map(|addr| { let futures = select_all(addresses.into_iter().map(|addr| {

4
rpxy-lib/src/proxy/proxy_h3.rs
View file

@ -15,7 +15,7 @@ where
U: CryptoSource + Clone + Sync + Send + 'static, U: CryptoSource + Clone + Sync + Send + 'static,
{ {
pub(super) async fn connection_serve_h3<C>( pub(super) async fn connection_serve_h3<C>(
self, &self,
quic_connection: C, quic_connection: C,
tls_server_name: ServerNameBytesExp, tls_server_name: ServerNameBytesExp,
client_addr: SocketAddr, client_addr: SocketAddr,
@ -79,7 +79,7 @@ where
} }
async fn stream_serve_h3<S>( async fn stream_serve_h3<S>(
self, &self,
req: Request<()>, req: Request<()>,
stream: RequestStream<S, Bytes>, stream: RequestStream<S, Bytes>,
client_addr: SocketAddr, client_addr: SocketAddr,

23
rpxy-lib/src/proxy/proxy_main.rs
View file

@ -40,7 +40,7 @@ where
{ {
pub listening_on: SocketAddr, pub listening_on: SocketAddr,
pub tls_enabled: bool, // TCP待受がTLSかどうか pub tls_enabled: bool, // TCP待受がTLSかどうか
pub msg_handler: HttpMessageHandler<T, U>, pub msg_handler: Arc<HttpMessageHandler<T, U>>,
pub globals: Arc<Globals<U>>, pub globals: Arc<Globals<U>>,
} }
@ -49,6 +49,21 @@ where
T: Connect + Clone + Sync + Send + 'static, T: Connect + Clone + Sync + Send + 'static,
U: CryptoSource + Clone + Sync + Send, U: CryptoSource + Clone + Sync + Send,
{ {
/// Wrapper function to handle request
async fn serve(
handler: Arc<HttpMessageHandler<T, U>>,
req: Request<Body>,
client_addr: SocketAddr,
listen_addr: SocketAddr,
tls_enabled: bool,
tls_server_name: Option<ServerNameBytesExp>,
) -> Result<hyper::Response<Body>> {
handler
.handle_request(req, client_addr, listen_addr, tls_enabled, tls_server_name)
.await
}
/// Serves requests from clients
pub(super) fn client_serve<I>( pub(super) fn client_serve<I>(
self, self,
stream: I, stream: I,
@ -72,7 +87,8 @@ where
.serve_connection( .serve_connection(
stream, stream,
service_fn(move |req: Request<Body>| { service_fn(move |req: Request<Body>| {
self.msg_handler.clone().handle_request( Self::serve(
self.msg_handler.clone(),
req, req,
peer_addr, peer_addr,
self.listening_on, self.listening_on,
@ -91,11 +107,11 @@ where
}); });
} }
/// Start without TLS (HTTP cleartext)
async fn start_without_tls(self, server: Http<LocalExecutor>) -> Result<()> { async fn start_without_tls(self, server: Http<LocalExecutor>) -> Result<()> {
let listener_service = async { let listener_service = async {
let tcp_socket = bind_tcp_socket(&self.listening_on)?; let tcp_socket = bind_tcp_socket(&self.listening_on)?;
let tcp_listener = tcp_socket.listen(self.globals.proxy_config.tcp_listen_backlog)?; let tcp_listener = tcp_socket.listen(self.globals.proxy_config.tcp_listen_backlog)?;
// let tcp_listener = TcpListener::bind(&self.listening_on).await?;
info!("Start TCP proxy serving with HTTP request for configured host names"); info!("Start TCP proxy serving with HTTP request for configured host names");
while let Ok((stream, _client_addr)) = tcp_listener.accept().await { while let Ok((stream, _client_addr)) = tcp_listener.accept().await {
self.clone().client_serve(stream, server.clone(), _client_addr, None); self.clone().client_serve(stream, server.clone(), _client_addr, None);
@ -106,6 +122,7 @@ where
Ok(()) Ok(())
} }
/// Entrypoint for HTTP/1.1 and HTTP/2 servers
pub async fn start(self) -> Result<()> { pub async fn start(self) -> Result<()> {
let mut server = Http::new(); let mut server = Http::new();
server.http1_keep_alive(self.globals.proxy_config.keepalive); server.http1_keep_alive(self.globals.proxy_config.keepalive);

2
rpxy-lib/src/proxy/socket.rs
View file

@ -32,7 +32,7 @@ pub(super) fn bind_udp_socket(listening_on: &SocketAddr) -> Result<UdpSocket> {
} else { } else {
Socket::new(Domain::IPV4, Type::DGRAM, Some(Protocol::UDP)) Socket::new(Domain::IPV4, Type::DGRAM, Some(Protocol::UDP))
}?; }?;
// socket.set_reuse_address(true)?; // This isn't necessary socket.set_reuse_address(true)?; // This isn't necessary?
socket.set_reuse_port(true)?; socket.set_reuse_port(true)?;
if let Err(e) = socket.bind(&(*listening_on).into()) { if let Err(e) = socket.bind(&(*listening_on).into()) {