pub async fn serve(
    common_state: CommonServerState,
    frontend_shutdown: CancellationToken,
    grpc_listener: tokio::net::TcpListener,
    http_listener: Option<AddrIncoming>,
    server_type: Arc<dyn ServerType>,
) -> Result<()> {
    let trace_header_parser = TraceHeaderParser::new()
        .with_jaeger_trace_context_header_name(
            &common_state
                .run_config()
                .tracing_config()
                .traces_jaeger_trace_context_header_name,
        )
        .with_jaeger_debug_name(
            &common_state
                .run_config()
                .tracing_config()
                .traces_jaeger_debug_name,
        );

    // Construct and start up gRPC server
    let grpc_server = rpc::serve(
        grpc_listener,
        Arc::clone(&server_type),
        trace_header_parser.clone(),
        frontend_shutdown.clone(),
    )
    .fuse();
    info!(?server_type, "gRPC server listening");

    let captured_server_type = Arc::clone(&server_type);
    let captured_shutdown = frontend_shutdown.clone();
    let http_server = async move {
        if let Some(http_listener) = http_listener {
            info!(server_type=?captured_server_type, "HTTP server listening");
            http::serve(
                http_listener,
                captured_server_type,
                captured_shutdown,
                trace_header_parser,
            )
            .await?
        } else {
            // don't resolve otherwise will cause server to shutdown
            captured_shutdown.cancelled().await
        }
        Ok(())
    }
    .fuse();

    // Purposefully use log not tokio-tracing to ensure correctly hooked up
    log::info!("InfluxDB IOx {:?} server ready", server_type);

    // Get IOx background worker join handle
    let server_handle = Arc::clone(&server_type).join().fuse();

    // Shutdown signal
    let signal = wait_for_signal().fuse();

    // There are two different select macros - tokio::select and futures::select
    //
    // tokio::select takes ownership of the passed future "moving" it into the
    // select block. This works well when not running select inside a loop, or
    // when using a future that can be dropped and recreated, often the case
    // with tokio's futures e.g. `channel.recv()`
    //
    // futures::select is more flexible as it doesn't take ownership of the provided
    // future. However, to safely provide this it imposes some additional
    // requirements
    //
    // All passed futures must implement FusedFuture - it is IB to poll a future
    // that has returned Poll::Ready(_). A FusedFuture has an is_terminated()
    // method that indicates if it is safe to poll - e.g. false if it has
    // returned Poll::Ready(_). futures::select uses this to implement its
    // functionality. futures::FutureExt adds a fuse() method that
    // wraps an arbitrary future and makes it a FusedFuture
    //
    // The additional requirement of futures::select is that if the future passed
    // outlives the select block, it must be Unpin or already Pinned

    // pin_mut constructs a Pin<&mut T> from a T by preventing moving the T
    // from the current stack frame and constructing a Pin<&mut T> to it
    pin_mut!(signal);
    pin_mut!(server_handle);
    pin_mut!(grpc_server);
    pin_mut!(http_server);

    // Return the first error encountered
    let mut res = Ok(());

    // Graceful shutdown can be triggered by sending SIGINT or SIGTERM to the
    // process, or by a background task exiting - most likely with an error
    //
    // Graceful shutdown should then proceed in the following order
    // 1. Stop accepting new HTTP and gRPC requests and drain existing connections
    // 2. Trigger shutdown of internal background workers loops
    //
    // This is important to ensure background tasks, such as polling the tracker
    // registry, don't exit before HTTP and gRPC requests dependent on them
    while !grpc_server.is_terminated() && !http_server.is_terminated() {
        futures::select! {
            _ = signal => info!(?server_type, "Shutdown requested"),
            _ = server_handle => {
                error!(?server_type, "server worker shutdown prematurely");
                res = res.and(Err(Error::LostServer));
            },
            result = grpc_server => match result {
                Ok(_) if frontend_shutdown.is_cancelled() => info!(?server_type, "gRPC server shutdown"),
                Ok(_) => {
                    error!(?server_type, "Early gRPC server exit");
                    res = res.and(Err(Error::LostRpc));
                }
                Err(error) => {
                    error!(%error, ?server_type, "gRPC server error");
                    res = res.and(Err(Error::ServingRpc{source: error}));
                }
            },
            result = http_server => match result {
                Ok(_) if frontend_shutdown.is_cancelled() => info!(?server_type, "HTTP server shutdown"),
                Ok(_) => {
                    error!(?server_type, "Early HTTP server exit");
                    res = res.and(Err(Error::LostHttp));
                }
                Err(error) => {
                    error!(%error, ?server_type, "HTTP server error");
                    res = res.and(Err(Error::ServingHttp{source: error}));
                }
            },
        }

        frontend_shutdown.cancel()
    }
    info!(?server_type, "frontend shutdown completed");

    server_type.shutdown();
    if !server_handle.is_terminated() {
        server_handle.await;
    }
    info!(?server_type, "backend shutdown completed");

    res
}
/// Instantiates the gRPC and optional HTTP listeners and returns a
/// Future that completes when these listeners, the Server, Databases,
/// etc... have all exited or the frontend_shutdown token is called.
Instantiates the gRPC and optional HTTP listeners and returns a
Future that completes when these listeners, the Server, Databases,
etc... have all exited or the frontend_shutdown token is called.
Instantiates the gRPC and optional HTTP listeners and returns a Future that completes when these listeners the Server Databases etc ... have all exited or the frontend_shutdown token is called .
pub async fn serve(
    common_state: CommonServerState,
    frontend_shutdown: CancellationToken,
    grpc_listener: tokio::net::TcpListener,
    http_listener: Option<AddrIncoming>,
    server_type: Arc<dyn ServerType>,
) -> Result<()> {
    let trace_header_parser = TraceHeaderParser::new()
        .with_jaeger_trace_context_header_name(
            &common_state
                .run_config()
                .tracing_config()
                .traces_jaeger_trace_context_header_name,
        )
        .with_jaeger_debug_name(
            &common_state
                .run_config()
                .tracing_config()
                .traces_jaeger_debug_name,
        );
    // Construct and start up gRPC server
    let grpc_server = rpc::serve(
        grpc_listener,
        Arc::clone(&server_type),
        trace_header_parser.clone(),
        frontend_shutdown.clone(),
    )
    .fuse();
    info!(?server_type, "gRPC server listening");
    let captured_server_type = Arc::clone(&server_type);
    let captured_shutdown = frontend_shutdown.clone();
    let http_server = async move {
        if let Some(http_listener) = http_listener {
            info!(server_type=?captured_server_type, "HTTP server listening");
            http::serve(
                http_listener,
                captured_server_type,
                captured_shutdown,
                trace_header_parser,
            )
            .await?
        } else {
            // don't resolve otherwise will cause server to shutdown
            captured_shutdown.cancelled().await
        }
        Ok(())
    }
    .fuse();
    // Purposefully use log not tokio-tracing to ensure correctly hooked up
    log::info!("InfluxDB IOx {:?} server ready", server_type);
    // Get IOx background worker join handle
    let server_handle = Arc::clone(&server_type).join().fuse();
    // Shutdown signal
    let signal = wait_for_signal().fuse();
    // There are two different select macros - tokio::select and futures::select
    //
    // tokio::select takes ownership of the passed future "moving" it into the
    // select block. This works well when not running select inside a loop, or
    // when using a future that can be dropped and recreated, often the case
    // with tokio's futures e.g. `channel.recv()`
    //
    // futures::select is more flexible as it doesn't take ownership of the provided
    // future. However, to safely provide this it imposes some additional
    // requirements
    //
    // All passed futures must implement FusedFuture - it is IB to poll a future
    // that has returned Poll::Ready(_). A FusedFuture has an is_terminated()
    // method that indicates if it is safe to poll - e.g. false if it has
    // returned Poll::Ready(_). futures::select uses this to implement its
    // functionality. futures::FutureExt adds a fuse() method that
    // wraps an arbitrary future and makes it a FusedFuture
    //
    // The additional requirement of futures::select is that if the future passed
    // outlives the select block, it must be Unpin or already Pinned
    // pin_mut constructs a Pin<&mut T> from a T by preventing moving the T
    // from the current stack frame and constructing a Pin<&mut T> to it
    pin_mut!(signal);
    pin_mut!(server_handle);
    pin_mut!(grpc_server);
    pin_mut!(http_server);
    // Return the first error encountered
    let mut res = Ok(());
    // Graceful shutdown can be triggered by sending SIGINT or SIGTERM to the
    // process, or by a background task exiting - most likely with an error
    //
    // Graceful shutdown should then proceed in the following order
    // 1. Stop accepting new HTTP and gRPC requests and drain existing connections
    // 2. Trigger shutdown of internal background workers loops
    //
    // This is important to ensure background tasks, such as polling the tracker
    // registry, don't exit before HTTP and gRPC requests dependent on them
    while !grpc_server.is_terminated() && !http_server.is_terminated() {
        futures::select! {
            _ = signal => info!(?server_type, "Shutdown requested"),
            _ = server_handle => {
                error!(?server_type, "server worker shutdown prematurely");
                res = res.and(Err(Error::LostServer));
            },
            result = grpc_server => match result {
                Ok(_) if frontend_shutdown.is_cancelled() => info!(?server_type, "gRPC server shutdown"),
                Ok(_) => {
                    error!(?server_type, "Early gRPC server exit");
                    res = res.and(Err(Error::LostRpc));
                }
                Err(error) => {
                    error!(%error, ?server_type, "gRPC server error");
                    res = res.and(Err(Error::ServingRpc{source: error}));
                }
            },
            result = http_server => match result {
                Ok(_) if frontend_shutdown.is_cancelled() => info!(?server_type, "HTTP server shutdown"),
                Ok(_) => {
                    error!(?server_type, "Early HTTP server exit");
                    res = res.and(Err(Error::LostHttp));
                }
                Err(error) => {
                    error!(%error, ?server_type, "HTTP server error");
                    res = res.and(Err(Error::ServingHttp{source: error}));
                }
            },
        }
        frontend_shutdown.cancel()
    }
    info!(?server_type, "frontend shutdown completed");
    server_type.shutdown();
    if !server_handle.is_terminated() {
        server_handle.await;
    }
    info!(?server_type, "backend shutdown completed");
    res
}
pub async fn serve ( common_state : CommonServerState , frontend_shutdown : CancellationToken , grpc_listener : tokio :: net :: TcpListener , http_listener : Option < AddrIncoming > , server_type : Arc < dyn ServerType > , ) -> Result < ( ) > { let trace_header_parser = TraceHeaderParser :: new ( ) . with_jaeger_trace_context_header_name ( & common_state . run_config ( ) . tracing_config ( ) . traces_jaeger_trace_context_header_name , ) . with_jaeger_debug_name ( & common_state . run_config ( ) . tracing_config ( ) . traces_jaeger_debug_name , ) ; // Construct and start up gRPC server let grpc_server = rpc :: serve ( grpc_listener , Arc :: clone ( & server_type ) , trace_header_parser . clone ( ) , frontend_shutdown . clone ( ) , ) . fuse ( ) ; info ! ( server_type " " ) ; let captured_server_type = Arc :: clone ( & server_type ) ; let captured_shutdown = frontend_shutdown . clone ( ) ; let http_server = async move { if let Some ( http_listener ) = http_listener { info ! ( server_type captured_server_type " " ) ; http :: serve ( http_listener , captured_server_type , captured_shutdown , trace_header_parser , ) . await ? } else { // don't resolve otherwise will cause server to shutdown captured_shutdown . cancelled ( ) . await } Ok ( ( ) ) } . fuse ( ) ; // Purposefully use log not tokio-tracing to ensure correctly hooked up log :: info ! ( " " server_type ) ; // Get IOx background worker join handle let server_handle = Arc :: clone ( & server_type ) . join ( ) . fuse ( ) ; // Shutdown signal let signal = wait_for_signal ( ) . fuse ( ) ; // There are two different select macros - tokio::select and futures::select // // tokio::select takes ownership of the passed future "moving" it into the // select block. This works well when not running select inside a loop, or // when using a future that can be dropped and recreated, often the case // with tokio's futures e.g. `channel.recv()` // // futures::select is more flexible as it doesn't take ownership of the provided // future. However, to safely provide this it imposes some additional // requirements // // All passed futures must implement FusedFuture - it is IB to poll a future // that has returned Poll::Ready(_). A FusedFuture has an is_terminated() // method that indicates if it is safe to poll - e.g. false if it has // returned Poll::Ready(_). futures::select uses this to implement its // functionality. futures::FutureExt adds a fuse() method that // wraps an arbitrary future and makes it a FusedFuture // // The additional requirement of futures::select is that if the future passed // outlives the select block, it must be Unpin or already Pinned // pin_mut constructs a Pin<&mut T> from a T by preventing moving the T // from the current stack frame and constructing a Pin<&mut T> to it pin_mut ! ( signal ) ; pin_mut ! ( server_handle ) ; pin_mut ! ( grpc_server ) ; pin_mut ! ( http_server ) ; // Return the first error encountered let mut res = Ok ( ( ) ) ; // Graceful shutdown can be triggered by sending SIGINT or SIGTERM to the // process, or by a background task exiting - most likely with an error // // Graceful shutdown should then proceed in the following order // 1. Stop accepting new HTTP and gRPC requests and drain existing connections // 2. Trigger shutdown of internal background workers loops // // This is important to ensure background tasks, such as polling the tracker // registry, don't exit before HTTP and gRPC requests dependent on them while ! grpc_server . is_terminated ( ) && ! http_server . is_terminated ( ) { futures :: select ! { signal info ( server_type " " ) server_handle { error ( server_type " " ) res res and ( Err ( Error LostServer ) ) } result grpc_server match result { Ok ( ) if frontend_shutdown is_cancelled ( ) info ( server_type " " ) Ok ( ) { error ( server_type " " ) res res and ( Err ( Error LostRpc ) ) } Err ( error ) { error ( error server_type " " ) res res and ( Err ( Error ServingRpc { source error } ) ) } } result http_server match result { Ok ( ) if frontend_shutdown is_cancelled ( ) info ( server_type " " ) Ok ( ) { error ( server_type " " ) res res and ( Err ( Error LostHttp ) ) } Err ( error ) { error ( error server_type " " ) res res and ( Err ( Error ServingHttp { source error } ) ) } } } frontend_shutdown . cancel ( ) } info ! ( server_type " " ) ; server_type . shutdown ( ) ; if ! server_handle . is_terminated ( ) { server_handle . await ; } info ! ( server_type " " ) ; res }
Instantiates the gRPC and optional HTTP listeners and returns a
Future that completes when these listeners, the Server, Databases,
etc... have all exited or the frontend_shutdown token is called.
Instantiates the gRPC and optional HTTP listeners and returns a Future that completes when these listeners the Server Databases etc ... have all exited or the frontend_shutdown token is called .
common_state
CommonServerState
frontend_shutdown
CancellationToken
grpc_listener
tokio::net::TcpListener
http_listener
Option<AddrIncoming>
server_type
Arc<dyn ServerType>
common_state
CommonServerState
frontend_shutdown
CancellationToken
grpc_listener
tokio::net::TcpListener
http_listener
Option<AddrIncoming>
server_type
Arc<dyn ServerType>