/// Creates and starts an `XNetEndpoint` to publish XNet `Streams`.
Creates and starts an `XNetEndpoint` to publish XNet `Streams`.
Creates and starts an ` XNetEndpoint ` to publish XNet ` Streams ` .
pub fn new(
        runtime_handle: runtime::Handle,
        certified_stream_store: Arc<dyn CertifiedStreamStore>,
        tls: Arc<dyn TlsHandshake + Send + Sync>,
        registry_client: Arc<dyn RegistryClient + Send + Sync>,
        config: XNetEndpointConfig,
        metrics: &MetricsRegistry,
        log: ReplicaLogger,
    ) -> Self {
        use crate::hyper::{tls_bind, ExecuteOnRuntime, TlsConnection};
        use hyper::service::{make_service_fn, service_fn};
        let metrics = Arc::new(XNetEndpointMetrics::new(metrics));
        // The bounded channel for queuing requests between the HTTP server and the
        // background worker.
        //
        // We use a background worker because building streams is CPU and memory
        // bound and can take a few dozens of milliseconds, which is too long to execute
        // on the event handling thread.
        //
        // We do not use [tokio::runtime::Handle::spawn_blocking] because it's designed
        // for I/O bound tasks and can spawn extra threads when needed, which is
        // not the best strategy for CPU-bound tasks.
        //
        // We use a crossbeam channel instead of [tokio::sync::mpsc] because we want the
        // receiver to be blocking, and [tokio::sync::mpsc::Receiver::blocking_recv] is
        // only available in tokio ≥ 0.3.
        let (request_sender, request_receiver) = crossbeam_channel::bounded(REQUEST_QUEUE_LENGTH);
        let make_service = make_service_fn({
            #[derive(Clone)]
            struct Context {
                log: ReplicaLogger,
                request_sender: crossbeam_channel::Sender<WorkerMessage>,
                metrics: Arc<XNetEndpointMetrics>,
            }
            let ctx = Context {
                log: log.clone(),
                metrics: Arc::clone(&metrics),
                request_sender: request_sender.clone(),
            };
            fn ok<T>(t: T) -> Result<T, Infallible> {
                Ok(t)
            }
            move |tls_conn: &TlsConnection| {
                let ctx = ctx.clone();
                debug!(
                    ctx.log,
                    "Serving XNet streams to peer {:?}",
                    tls_conn.peer()
                );
                async move {
                    let ctx = ctx.clone();
                    ok(service_fn({
                        move |request: Request<Body>| {
                            let ctx = ctx.clone();
                            async move {
                                let (response_sender, response_receiver) = oneshot::channel();
                                let task = WorkerMessage::HandleRequest {
                                    request,
                                    response_sender,
                                };
                                // NOTE: we must use non-blocking send here, otherwise we might
                                // delay the event thread.
                                if ctx.request_sender.try_send(task).is_err() {
                                    ctx.metrics
                                        .request_duration
                                        .with_label_values(&[
                                            RESOURCE_UNKNOWN,
                                            StatusCode::SERVICE_UNAVAILABLE.as_str(),
                                        ])
                                        .observe(0.0);
                                    return ok(Response::builder()
                                        .status(StatusCode::SERVICE_UNAVAILABLE)
                                        .body(Body::from("Queue full"))
                                        .unwrap());
                                }
                                ok(response_receiver.await.unwrap_or_else(|e| {
                                    panic!("XNet Endpoint Handler shut down unexpectedly: {}", e)
                                }))
                            }
                        }
                    }))
                }
            }
        });
        let (address, server) = {
            let _guard = runtime_handle.enter();
            let (addr, builder) =
                tls_bind(&config.address, tls, registry_client).unwrap_or_else(|e| {
                    panic!(
                        "failed to bind XNet socket, address {:?}: {}",
                        config.address, e
                    )
                });
            (
                addr,
                builder
                    .executor(ExecuteOnRuntime(runtime_handle.clone()))
                    .serve(make_service),
            )
        };
        info!(log, "XNet Endpoint listening on {}", address);
        let shutdown_notify = Arc::new(Notify::new());
        let shutdown = server.with_graceful_shutdown({
            let shutdown_notify = Arc::clone(&shutdown_notify);
            async move { shutdown_notify.notified().await }
        });
        runtime_handle.spawn({
            let log = log.clone();
            async move {
                if let Err(e) = shutdown.await {
                    warn!(log, "XNet http server failed: {}", e);
                }
            }
        });
        // Spawn a request handler. We pass the certified stream store, which is
        // currently realized by the state manager.
        let handler_log = log.clone();
        let base_url = Url::parse(&format!("http://{}/", address)).unwrap();
        let handler_thread_handle = std::thread::Builder::new()
            .name("XNet Endpoint Handler".to_string())
            .spawn(move || {
                while let Ok(WorkerMessage::HandleRequest {
                    request,
                    response_sender,
                }) = request_receiver.recv()
                {
                    let response = handle_http_request(
                        request,
                        certified_stream_store.as_ref(),
                        &base_url,
                        &metrics,
                        &handler_log,
                    );
                    response_sender.send(response).unwrap_or_else(|res| {
                        info!(
                            handler_log,
                            "Failed to respond with {:?}",
                            res.into_parts().0
                        )
                    });
                }
                debug!(handler_log, "  ...XNet Endpoint Handler shut down");
            })
            .expect("Cannot spawn XNet Endpoint Handler thread");
        Self {
            server_address: address,
            shutdown_notify,
            request_sender,
            handler_thread_handle: Some(handler_thread_handle),
            log,
        }
    }
pub fn new ( runtime_handle : runtime :: Handle , certified_stream_store : Arc < dyn CertifiedStreamStore > , tls : Arc < dyn TlsHandshake + Send + Sync > , registry_client : Arc < dyn RegistryClient + Send + Sync > , config : XNetEndpointConfig , metrics : & MetricsRegistry , log : ReplicaLogger , ) -> Self { use crate :: hyper :: { tls_bind , ExecuteOnRuntime , TlsConnection } ; use hyper :: service :: { make_service_fn , service_fn } ; let metrics = Arc :: new ( XNetEndpointMetrics :: new ( metrics ) ) ; // The bounded channel for queuing requests between the HTTP server and the // background worker. // // We use a background worker because building streams is CPU and memory // bound and can take a few dozens of milliseconds, which is too long to execute // on the event handling thread. // // We do not use [tokio::runtime::Handle::spawn_blocking] because it's designed // for I/O bound tasks and can spawn extra threads when needed, which is // not the best strategy for CPU-bound tasks. // // We use a crossbeam channel instead of [tokio::sync::mpsc] because we want the // receiver to be blocking, and [tokio::sync::mpsc::Receiver::blocking_recv] is // only available in tokio ≥ 0.3. let ( request_sender , request_receiver ) = crossbeam_channel :: bounded ( REQUEST_QUEUE_LENGTH ) ; let make_service = make_service_fn ( { # [ derive ( Clone ) ] struct Context { log : ReplicaLogger , request_sender : crossbeam_channel :: Sender < WorkerMessage > , metrics : Arc < XNetEndpointMetrics > , } let ctx = Context { log : log . clone ( ) , metrics : Arc :: clone ( & metrics ) , request_sender : request_sender . clone ( ) , } ; fn ok < T > ( t : T ) -> Result < T , Infallible > { Ok ( t ) } move | tls_conn : & TlsConnection | { let ctx = ctx . clone ( ) ; debug ! ( ctx log " " tls_conn peer ( ) ) ; async move { let ctx = ctx . clone ( ) ; ok ( service_fn ( { move | request : Request < Body > | { let ctx = ctx . clone ( ) ; async move { let ( response_sender , response_receiver ) = oneshot :: channel ( ) ; let task = WorkerMessage :: HandleRequest { request , response_sender , } ; // NOTE: we must use non-blocking send here, otherwise we might // delay the event thread. if ctx . request_sender . try_send ( task ) . is_err ( ) { ctx . metrics . request_duration . with_label_values ( & [ RESOURCE_UNKNOWN , StatusCode :: SERVICE_UNAVAILABLE . as_str ( ) , ] ) . observe ( 0.0 ) ; return ok ( Response :: builder ( ) . status ( StatusCode :: SERVICE_UNAVAILABLE ) . body ( Body :: from ( " " ) ) . unwrap ( ) ) ; } ok ( response_receiver . await . unwrap_or_else ( | e | { panic ! ( " " e ) } ) ) } } } ) ) } } } ) ; let ( address , server ) = { let _guard = runtime_handle . enter ( ) ; let ( addr , builder ) = tls_bind ( & config . address , tls , registry_client ) . unwrap_or_else ( | e | { panic ! ( " " config address e ) } ) ; ( addr , builder . executor ( ExecuteOnRuntime ( runtime_handle . clone ( ) ) ) . serve ( make_service ) , ) } ; info ! ( log " " address ) ; let shutdown_notify = Arc :: new ( Notify :: new ( ) ) ; let shutdown = server . with_graceful_shutdown ( { let shutdown_notify = Arc :: clone ( & shutdown_notify ) ; async move { shutdown_notify . notified ( ) . await } } ) ; runtime_handle . spawn ( { let log = log . clone ( ) ; async move { if let Err ( e ) = shutdown . await { warn ! ( log " " e ) ; } } } ) ; // Spawn a request handler. We pass the certified stream store, which is // currently realized by the state manager. let handler_log = log . clone ( ) ; let base_url = Url :: parse ( & format ! ( " " address ) ) . unwrap ( ) ; let handler_thread_handle = std :: thread :: Builder :: new ( ) . name ( " " . to_string ( ) ) . spawn ( move | | { while let Ok ( WorkerMessage :: HandleRequest { request , response_sender , } ) = request_receiver . recv ( ) { let response = handle_http_request ( request , certified_stream_store . as_ref ( ) , & base_url , & metrics , & handler_log , ) ; response_sender . send ( response ) . unwrap_or_else ( | res | { info ! ( handler_log " " res into_parts ( ) 0 ) } ) ; } debug ! ( handler_log " " ) ; } ) . expect ( " " ) ; Self { server_address : address , shutdown_notify , request_sender , handler_thread_handle : Some ( handler_thread_handle ) , log , } }
Creates and starts an `XNetEndpoint` to publish XNet `Streams`.
Creates and starts an ` XNetEndpoint ` to publish XNet ` Streams ` .
runtime_handle
runtime::Handle
certified_stream_store
Arc<dyn CertifiedStreamStore>
tls
Arc<dyn TlsHandshake + Send + Sync>
registry_client
Arc<dyn RegistryClient + Send + Sync>
config
XNetEndpointConfig
metrics
&MetricsRegistry
log
ReplicaLogger