n3server/
lib.rs

1//! `N3` is a fast `quic/http3` reverse proxy implementation.
2
3#![cfg_attr(docsrs, feature(doc_cfg))]
4
5use std::{
6    io::Result,
7    net::{SocketAddr, ToSocketAddrs},
8};
9
10use futures::{AsyncWriteExt, io::copy};
11use n3_spawner::spawn;
12use n3io::net::TcpStream;
13use n3quic::{QuicConn, QuicConnExt, QuicServer};
14
15/// Reverse proxy server.
16pub struct N3 {
17    /// Redirection target for tcp stream.
18    redirect_to: SocketAddr,
19    /// the QUIC server configuration.
20    quic_server: QuicServer,
21}
22
23impl N3 {
24    /// Create a new `N3` configuration with `redirect_to` target.
25    pub fn new(redirect_to: SocketAddr) -> Self {
26        Self {
27            redirect_to,
28            quic_server: QuicServer::new(),
29        }
30    }
31
32    // Update `quic_server` config.
33    pub fn quic_server<F>(mut self, f: F) -> Self
34    where
35        F: FnOnce(QuicServer) -> QuicServer,
36    {
37        self.quic_server = f(self.quic_server);
38        self
39    }
40
41    /// Bind `n3` to `laddrs` and run it.
42    pub async fn bind<S>(self, laddrs: S) -> Result<()>
43    where
44        S: ToSocketAddrs,
45    {
46        let mut listener = self.quic_server.bind(laddrs).await?;
47
48        loop {
49            let conn = listener.accept().await?;
50
51            spawn(async move {
52                let trace_id = conn.quiche_conn(|conn| conn.trace_id().to_owned());
53
54                log::info!("redirect, id={}, to={}", trace_id, self.redirect_to);
55
56                if let Err(err) = Self::redirect_loop(conn, self.redirect_to, &trace_id).await {
57                    log::error!("pipe is broken, id={}, err={}", trace_id, err);
58                } else {
59                    log::info!("pipe is broken, id={}", trace_id);
60                }
61            })?;
62        }
63    }
64
65    async fn redirect_loop(conn: QuicConn, raddr: SocketAddr, trace_id: &str) -> Result<()> {
66        loop {
67            let inbound = conn.accept().await?;
68
69            let outbound = TcpStream::connect(raddr).await?;
70
71            let stream_id = inbound.id();
72
73            let laddr = outbound.mio_socket().local_addr()?;
74
75            log::info!(
76                "new pipe quic({},{}) => tcp({},{})",
77                trace_id,
78                stream_id,
79                laddr,
80                raddr
81            );
82
83            let (mut inbound_writer, inbound_reader) = inbound.split();
84            let (mut outbound_writer, outbound_reader) = outbound.split();
85
86            let trace_id_owned = trace_id.to_owned();
87
88            spawn(async move {
89                match copy(outbound_reader, &mut inbound_writer).await {
90                    Ok(len) => {
91                        log::info!(
92                            "stream(backward) is closed, quic({},{}) <== tcp({},{}), trans_size={}",
93                            trace_id_owned,
94                            stream_id,
95                            laddr,
96                            raddr,
97                            len
98                        );
99                    }
100                    Err(err) => {
101                        log::error!(
102                            "stream(backward) is broken, quic({},{}) <== tcp({},{}), err={}",
103                            trace_id_owned,
104                            stream_id,
105                            laddr,
106                            raddr,
107                            err
108                        );
109                    }
110                }
111
112                if let Err(err) = inbound_writer.close().await {
113                    log::trace!(
114                        "stream(backward) close writer, quic({},{}) ==> tcp({},{}), err={}",
115                        trace_id_owned,
116                        stream_id,
117                        laddr,
118                        raddr,
119                        err
120                    );
121                }
122            })?;
123
124            let trace_id_owned = trace_id.to_owned();
125
126            spawn(async move {
127                match copy(inbound_reader, &mut outbound_writer).await {
128                    Ok(len) => {
129                        log::info!(
130                            "stream(forward) is closed, quic({},{}) ==> tcp({},{}), trans_size={}",
131                            trace_id_owned,
132                            stream_id,
133                            laddr,
134                            raddr,
135                            len
136                        );
137                    }
138                    Err(err) => {
139                        log::error!(
140                            "stream(forward) is broken, quic({},{}) ==> tcp({},{}), err={}",
141                            trace_id_owned,
142                            stream_id,
143                            laddr,
144                            raddr,
145                            err
146                        );
147                    }
148                }
149
150                if let Err(err) = outbound_writer.close().await {
151                    log::error!(
152                        "stream(forward) close writer, quic({},{}) <== tcp({},{}), err={}",
153                        trace_id_owned,
154                        stream_id,
155                        laddr,
156                        raddr,
157                        err
158                    );
159                }
160            })?;
161        }
162    }
163}
n3server/lib.rs

n3server/
lib.rs
