pub async fn serve<M>(
    socket: TcpListener,
    application: Arc<ApplicationState>,
    server: Arc<Server<M>>,
    shutdown: CancellationToken,
    serving_readiness: ServingReadiness,
) -> Result<()>
where
    M: ConnectionManager + Send + Sync + Debug + 'static,
{
    let stream = TcpListenerStream::new(socket);

    let (mut health_reporter, health_service) = tonic_health::server::health_reporter();
    let reflection_service = tonic_reflection::server::Builder::configure()
        .register_encoded_file_descriptor_set(generated_types::FILE_DESCRIPTOR_SET)
        .build()
        .context(ReflectionError)?;

    let mut builder = tonic::transport::Server::builder();

    // important that this one is NOT gated so that it can answer health requests
    add_service!(builder, health_reporter, health_service);
    add_service!(builder, health_reporter, reflection_service);
    add_service!(builder, health_reporter, testing::make_server());
    add_gated_service!(
        builder,
        health_reporter,
        serving_readiness,
        storage::make_server(
            Arc::clone(&server),
            Arc::clone(application.metric_registry()),
        )
    );
    add_gated_service!(
        builder,
        health_reporter,
        serving_readiness,
        flight::make_server(Arc::clone(&server))
    );
    add_gated_service!(
        builder,
        health_reporter,
        serving_readiness,
        write::make_server(Arc::clone(&server))
    );
    add_gated_service!(
        builder,
        health_reporter,
        serving_readiness,
        write_pb::make_server(Arc::clone(&server))
    );
    // Also important this is not behind a readiness check (as it is
    // used to change the check!)
    add_service!(
        builder,
        health_reporter,
        management::make_server(
            Arc::clone(&application),
            Arc::clone(&server),
            serving_readiness.clone()
        )
    );
    add_service!(
        builder,
        health_reporter,
        operations::make_server(Arc::clone(application.job_registry()))
    );

    builder
        .serve_with_incoming_shutdown(stream, shutdown.cancelled())
        .await
        .context(TransportError)
}
/// Instantiate a server listening on the specified address
/// implementing the IOx, Storage, and Flight gRPC interfaces, the
/// underlying hyper server instance. Resolves when the server has
/// shutdown.
Instantiate a server listening on the specified address
implementing the IOx, Storage, and Flight gRPC interfaces, the
underlying hyper server instance. Resolves when the server has
shutdown.
Instantiate a server listening on the specified address implementing the IOx Storage and Flight gRPC interfaces the underlying hyper server instance . Resolves when the server has shutdown .
pub async fn serve<M>(
    socket: TcpListener,
    application: Arc<ApplicationState>,
    server: Arc<Server<M>>,
    shutdown: CancellationToken,
    serving_readiness: ServingReadiness,
) -> Result<()>
where
    M: ConnectionManager + Send + Sync + Debug + 'static,
{
    let stream = TcpListenerStream::new(socket);
    let (mut health_reporter, health_service) = tonic_health::server::health_reporter();
    let reflection_service = tonic_reflection::server::Builder::configure()
        .register_encoded_file_descriptor_set(generated_types::FILE_DESCRIPTOR_SET)
        .build()
        .context(ReflectionError)?;
    let mut builder = tonic::transport::Server::builder();
    // important that this one is NOT gated so that it can answer health requests
    add_service!(builder, health_reporter, health_service);
    add_service!(builder, health_reporter, reflection_service);
    add_service!(builder, health_reporter, testing::make_server());
    add_gated_service!(
        builder,
        health_reporter,
        serving_readiness,
        storage::make_server(
            Arc::clone(&server),
            Arc::clone(application.metric_registry()),
        )
    );
    add_gated_service!(
        builder,
        health_reporter,
        serving_readiness,
        flight::make_server(Arc::clone(&server))
    );
    add_gated_service!(
        builder,
        health_reporter,
        serving_readiness,
        write::make_server(Arc::clone(&server))
    );
    add_gated_service!(
        builder,
        health_reporter,
        serving_readiness,
        write_pb::make_server(Arc::clone(&server))
    );
    // Also important this is not behind a readiness check (as it is
    // used to change the check!)
    add_service!(
        builder,
        health_reporter,
        management::make_server(
            Arc::clone(&application),
            Arc::clone(&server),
            serving_readiness.clone()
        )
    );
    add_service!(
        builder,
        health_reporter,
        operations::make_server(Arc::clone(application.job_registry()))
    );
    builder
        .serve_with_incoming_shutdown(stream, shutdown.cancelled())
        .await
        .context(TransportError)
}
pub async fn serve < M > ( socket : TcpListener , application : Arc < ApplicationState > , server : Arc < Server < M > > , shutdown : CancellationToken , serving_readiness : ServingReadiness , ) -> Result < ( ) > where M : ConnectionManager + Send + Sync + Debug + ' static , { let stream = TcpListenerStream :: new ( socket ) ; let ( mut health_reporter , health_service ) = tonic_health :: server :: health_reporter ( ) ; let reflection_service = tonic_reflection :: server :: Builder :: configure ( ) . register_encoded_file_descriptor_set ( generated_types :: FILE_DESCRIPTOR_SET ) . build ( ) . context ( ReflectionError ) ? ; let mut builder = tonic :: transport :: Server :: builder ( ) ; // important that this one is NOT gated so that it can answer health requests add_service ! ( builder health_reporter health_service ) ; add_service ! ( builder health_reporter reflection_service ) ; add_service ! ( builder health_reporter testing make_server ( ) ) ; add_gated_service ! ( builder health_reporter serving_readiness storage make_server ( Arc clone ( server ) Arc clone ( application metric_registry ( ) ) ) ) ; add_gated_service ! ( builder health_reporter serving_readiness flight make_server ( Arc clone ( server ) ) ) ; add_gated_service ! ( builder health_reporter serving_readiness write make_server ( Arc clone ( server ) ) ) ; add_gated_service ! ( builder health_reporter serving_readiness write_pb make_server ( Arc clone ( server ) ) ) ; // Also important this is not behind a readiness check (as it is // used to change the check!) add_service ! ( builder health_reporter management make_server ( Arc clone ( application ) Arc clone ( server ) serving_readiness clone ( ) ) ) ; add_service ! ( builder health_reporter operations make_server ( Arc clone ( application job_registry ( ) ) ) ) ; builder . serve_with_incoming_shutdown ( stream , shutdown . cancelled ( ) ) . await . context ( TransportError ) }
Instantiate a server listening on the specified address
implementing the IOx, Storage, and Flight gRPC interfaces, the
underlying hyper server instance.
Instantiate a server listening on the specified address implementing the IOx Storage and Flight gRPC interfaces the underlying hyper server instance .
socket
TcpListener
application
Arc<ApplicationState>
server
Arc<Server<M>>
shutdown
CancellationToken
serving_readiness
ServingReadiness
socket
TcpListener
application
Arc<ApplicationState>
server
Arc<Server<M>>
shutdown
CancellationToken
serving_readiness
ServingReadiness