fn serve_connection(
        &mut self,
        connection: Connection,
        drain: drain::Watch,
    ) -> Box<dyn Future<Item = (), Error = ()> + Send + 'static> {
        let source = Source {
            remote: connection.remote_addr(),
            local: connection.local_addr().unwrap_or(self.listen_addr),
            orig_dst: connection.original_dst_addr(),
            tls_peer: connection.peer_identity(),
            _p: (),
        };

        let should_detect = connection.should_detect_protocol();
        let io = self.accept.accept(&source, connection);

        let accept_fut = if should_detect {
            Either::A(
                io.peek()
                    .map_err(|e| debug!("peek error: {}", e))
                    .map(|io| {
                        let proto = Protocol::detect(io.peeked());
                        (proto, io)
                    }),
            )
        } else {
            debug!("protocol detection disabled for {:?}", source.orig_dst);
            Either::B(future::ok((None, io)))
        };

        let connect = self.connect.clone();
        let log = self.log.clone().with_remote(source.remote);

        let mut http = self.http.clone();
        let mut make_http = self.make_http.clone();
        let log_clone = log.clone();
        let initial_stream_window_size = self.h2_settings.initial_stream_window_size;
        let initial_conn_window_size = self.h2_settings.initial_connection_window_size;
        let serve_fut = accept_fut.and_then(move |(proto, io)| match proto {
            None => {
                trace!("did not detect protocol; forwarding TCP");
                let fwd = tcp::forward(io, connect, source);
                Either::A(drain.watch(fwd, |_| {}))
            }

            Some(proto) => Either::B({
                make_http
                    .make_service(source)
                    .map_err(|never| match never {})
                    .and_then(move |http_svc| match proto {
                        Protocol::Http1 => Either::A({
                            trace!("detected HTTP/1");
                            // Enable support for HTTP upgrades (CONNECT and websockets).
                            let svc = upgrade::Service::new(
                                http_svc,
                                drain.clone(),
                                log_clone.executor(),
                            );
                            let conn = http
                                .http1_only(true)
                                .serve_connection(io, HyperServerSvc::new(svc))
                                .with_upgrades();
                            drain
                                .watch(conn, |conn| conn.graceful_shutdown())
                                .map(|_| ())
                                .map_err(|e| debug!("http1 server error: {:?}", e))
                        }),

                        Protocol::Http2 => Either::B({
                            trace!("detected HTTP/2");
                            let conn = http
                                .with_executor(log_clone.executor())
                                .http2_only(true)
                                .http2_initial_stream_window_size(initial_stream_window_size)
                                .http2_initial_connection_window_size(initial_conn_window_size)
                                .serve_connection(io, HyperServerSvc::new(http_svc));
                            drain
                                .watch(conn, |conn| conn.graceful_shutdown())
                                .map(|_| ())
                                .map_err(|e| debug!("http2 server error: {:?}", e))
                        }),
                    })
            }),
        });

        Box::new(serve_fut)
    }
/// Handle a new connection.
///
/// This will peek on the connection for the first bytes to determine
/// what protocol the connection is speaking. From there, the connection
/// will be mapped into respective services, and spawned into an
/// executor.
Handle a new connection.
This will peek on the connection for the first bytes to determine
what protocol the connection is speaking. From there, the connection
will be mapped into respective services, and spawned into an
executor.
Handle a new connection . This will peek on the connection for the first bytes to determine what protocol the connection is speaking . From there the connection will be mapped into respective services and spawned into an executor .
fn serve_connection(
        &mut self,
        connection: Connection,
        drain: drain::Watch,
    ) -> Box<dyn Future<Item = (), Error = ()> + Send + 'static> {
        let source = Source {
            remote: connection.remote_addr(),
            local: connection.local_addr().unwrap_or(self.listen_addr),
            orig_dst: connection.original_dst_addr(),
            tls_peer: connection.peer_identity(),
            _p: (),
        };
        let should_detect = connection.should_detect_protocol();
        let io = self.accept.accept(&source, connection);
        let accept_fut = if should_detect {
            Either::A(
                io.peek()
                    .map_err(|e| debug!("peek error: {}", e))
                    .map(|io| {
                        let proto = Protocol::detect(io.peeked());
                        (proto, io)
                    }),
            )
        } else {
            debug!("protocol detection disabled for {:?}", source.orig_dst);
            Either::B(future::ok((None, io)))
        };
        let connect = self.connect.clone();
        let log = self.log.clone().with_remote(source.remote);
        let mut http = self.http.clone();
        let mut make_http = self.make_http.clone();
        let log_clone = log.clone();
        let initial_stream_window_size = self.h2_settings.initial_stream_window_size;
        let initial_conn_window_size = self.h2_settings.initial_connection_window_size;
        let serve_fut = accept_fut.and_then(move |(proto, io)| match proto {
            None => {
                trace!("did not detect protocol; forwarding TCP");
                let fwd = tcp::forward(io, connect, source);
                Either::A(drain.watch(fwd, |_| {}))
            }
            Some(proto) => Either::B({
                make_http
                    .make_service(source)
                    .map_err(|never| match never {})
                    .and_then(move |http_svc| match proto {
                        Protocol::Http1 => Either::A({
                            trace!("detected HTTP/1");
                            // Enable support for HTTP upgrades (CONNECT and websockets).
                            let svc = upgrade::Service::new(
                                http_svc,
                                drain.clone(),
                                log_clone.executor(),
                            );
                            let conn = http
                                .http1_only(true)
                                .serve_connection(io, HyperServerSvc::new(svc))
                                .with_upgrades();
                            drain
                                .watch(conn, |conn| conn.graceful_shutdown())
                                .map(|_| ())
                                .map_err(|e| debug!("http1 server error: {:?}", e))
                        }),
                        Protocol::Http2 => Either::B({
                            trace!("detected HTTP/2");
                            let conn = http
                                .with_executor(log_clone.executor())
                                .http2_only(true)
                                .http2_initial_stream_window_size(initial_stream_window_size)
                                .http2_initial_connection_window_size(initial_conn_window_size)
                                .serve_connection(io, HyperServerSvc::new(http_svc));
                            drain
                                .watch(conn, |conn| conn.graceful_shutdown())
                                .map(|_| ())
                                .map_err(|e| debug!("http2 server error: {:?}", e))
                        }),
                    })
            }),
        });
        Box::new(serve_fut)
    }
fn serve_connection ( & mut self , connection : Connection , drain : drain :: Watch , ) -> Box < dyn Future < Item = ( ) , Error = ( ) > + Send + ' static > { let source = Source { remote : connection . remote_addr ( ) , local : connection . local_addr ( ) . unwrap_or ( self . listen_addr ) , orig_dst : connection . original_dst_addr ( ) , tls_peer : connection . peer_identity ( ) , _p : ( ) , } ; let should_detect = connection . should_detect_protocol ( ) ; let io = self . accept . accept ( & source , connection ) ; let accept_fut = if should_detect { Either :: A ( io . peek ( ) . map_err ( | e | debug ! ( " " e ) ) . map ( | io | { let proto = Protocol :: detect ( io . peeked ( ) ) ; ( proto , io ) } ) , ) } else { debug ! ( " " source orig_dst ) ; Either :: B ( future :: ok ( ( None , io ) ) ) } ; let connect = self . connect . clone ( ) ; let log = self . log . clone ( ) . with_remote ( source . remote ) ; let mut http = self . http . clone ( ) ; let mut make_http = self . make_http . clone ( ) ; let log_clone = log . clone ( ) ; let initial_stream_window_size = self . h2_settings . initial_stream_window_size ; let initial_conn_window_size = self . h2_settings . initial_connection_window_size ; let serve_fut = accept_fut . and_then ( move | ( proto , io ) | match proto { None => { trace ! ( " " ) ; let fwd = tcp :: forward ( io , connect , source ) ; Either :: A ( drain . watch ( fwd , | _ | { } ) ) } Some ( proto ) => Either :: B ( { make_http . make_service ( source ) . map_err ( | never | match never { } ) . and_then ( move | http_svc | match proto { Protocol :: Http1 => Either :: A ( { trace ! ( " " ) ; // Enable support for HTTP upgrades (CONNECT and websockets). let svc = upgrade :: Service :: new ( http_svc , drain . clone ( ) , log_clone . executor ( ) , ) ; let conn = http . http1_only ( true ) . serve_connection ( io , HyperServerSvc :: new ( svc ) ) . with_upgrades ( ) ; drain . watch ( conn , | conn | conn . graceful_shutdown ( ) ) . map ( | _ | ( ) ) . map_err ( | e | debug ! ( " " e ) ) } ) , Protocol :: Http2 => Either :: B ( { trace ! ( " " ) ; let conn = http . with_executor ( log_clone . executor ( ) ) . http2_only ( true ) . http2_initial_stream_window_size ( initial_stream_window_size ) . http2_initial_connection_window_size ( initial_conn_window_size ) . serve_connection ( io , HyperServerSvc :: new ( http_svc ) ) ; drain . watch ( conn , | conn | conn . graceful_shutdown ( ) ) . map ( | _ | ( ) ) . map_err ( | e | debug ! ( " " e ) ) } ) , } ) } ) , } ) ; Box :: new ( serve_fut ) }
Handle a new connection.
Handle a new connection .
&mut self
connection
Connection
drain
drain::Watch
&mut self
connection
Connection
drain
drain::Watch