Struct Server

pub struct Server { /* private fields */ }

A gRPC Server that implements the tokio-console wire format.

Client applications, such as the tokio-console CLI connect to the gRPC server, and stream data about the runtime’s history (such as a list of the currently active tasks, or statistics summarizing polling times). A Server also interprets commands from a client application, such a request to focus in on a specific task, and translates that into a stream of details specific to that task.

Implementations

impl Server

pub const DEFAULT_IP: IpAddr = _

A Server by default binds socket address 127.0.0.1 to service remote procedure calls.

Note that methods like init and spawn will parse the socket address from the TOKIO_CONSOLE_BIND environment variable before falling back on constructing a socket address from this default.

See also Builder::server_addr.

pub const DEFAULT_PORT: u16 = 6_669u16

A Server by default binds port 6669 to service remote procedure calls.

Note that methods like init and spawn will parse the socket address from the TOKIO_CONSOLE_BIND environment variable before falling back on constructing a socket address from this default.

See also Builder::server_addr.

pub async fn serve(self) -> Result<(), Box<dyn Error + Send + Sync + 'static>>

Starts the gRPC service with the default gRPC settings.

To configure gRPC server settings before starting the server, use serve_with instead. This method is equivalent to calling serve_with and providing the default gRPC server settings:

server.serve_with(tonic::transport::Server::default()).await

pub async fn serve_with( self, builder: Server, ) -> Result<(), Box<dyn Error + Send + Sync + 'static>>

Starts the gRPC service with the given tonic gRPC transport server builder.

The builder parameter may be used to configure gRPC-specific settings prior to starting the server.

This spawns both the server task and the event aggregation worker task on the current async runtime.

pub async fn serve_with_grpc_web( self, builder: Server, ) -> Result<(), Box<dyn Error + Send + Sync + 'static>>

Starts the gRPC service with the default gRPC settings and gRPC-Web support.

Examples

To serve the instrument server with gRPC-Web support with the default settings:

server.serve_with_grpc_web(tonic::transport::Server::default()).await

To serve the instrument server with gRPC-Web support and a custom CORS configuration, use the following code:

use console_subscriber::{ConsoleLayer, ServerParts};
use tonic_web::GrpcWebLayer;
use tower_http::cors::{CorsLayer, AllowOrigin};
use http::header::HeaderName;

let (console_layer, server) = ConsoleLayer::builder().with_default_env().build();
// Customize the CORS configuration.
let cors = CorsLayer::new()
    .allow_origin(AllowOrigin::mirror_request())
    .allow_credentials(true)
    .max_age(DEFAULT_MAX_AGE)
    .expose_headers(
        DEFAULT_EXPOSED_HEADERS
            .iter()
            .cloned()
            .map(HeaderName::from_static)
            .collect::<Vec<HeaderName>>(),
    )
    .allow_headers(
        DEFAULT_ALLOW_HEADERS
            .iter()
            .cloned()
            .map(HeaderName::from_static)
            .collect::<Vec<HeaderName>>(),
    );

let ServerParts {
    instrument_server,
    aggregator,
    ..
} = server.into_parts();
tokio::spawn(aggregator.run());

// Serve the instrument server with gRPC-Web support and the CORS configuration.
let router = tonic::transport::Server::builder()
    .accept_http1(true)
    .layer(cors)
    .layer(GrpcWebLayer::new())
    .add_service(instrument_server);
let serve = router.serve(std::net::SocketAddr::new(
    std::net::IpAddr::V4(std::net::Ipv4Addr::new(127, 0, 0, 1)),
    // 6669 is a restricted port on Chrome, so we cannot use it. We use a different port instead.
    9999,
));

// Finally, spawn the server.
serve.await.expect("console subscriber server failed");

For a comprehensive understanding and complete code example, please refer to the grpc-web example in the examples directory.

pub fn into_parts(self) -> ServerParts

Returns the parts needed to spawn a gRPC server and the aggregator that supplies it.

Note that a server spawned in this way will disregard any value set by Builder::server_addr, as the user becomes responsible for defining the address when calling Router::serve.

Additionally, the user of this API must ensure that the Aggregator is running for as long as the gRPC server is. If the server stops running, the aggregator task can be aborted.

Examples

The parts can be used to serve the instrument server together with other endpoints from the same gRPC server.

use console_subscriber::{ConsoleLayer, ServerParts};

let (console_layer, server) = ConsoleLayer::builder().build();
let ServerParts {
    instrument_server,
    aggregator,
    ..
} = server.into_parts();

let aggregator_handle = tokio::spawn(aggregator.run());
let router = tonic::transport::Server::builder()
    //.add_service(some_other_service)
    .add_service(instrument_server);
let serve = router.serve(std::net::SocketAddr::new(
    std::net::IpAddr::V4(std::net::Ipv4Addr::new(127, 0, 0, 1)),
    6669,
));

// Finally, spawn the server.
tokio::spawn(serve);

Trait Implementations

impl Instrument for Server

type WatchUpdatesStream = ReceiverStream<Result<Update, Status>>

Server streaming response type for the WatchUpdates method.

type WatchTaskDetailsStream = ReceiverStream<Result<TaskDetails, Status>>

Server streaming response type for the WatchTaskDetails method.

fn watch_updates<'life0, 'async_trait>( &'life0 self, req: Request<InstrumentRequest>, ) -> Pin<Box<dyn Future<Output = Result<Response<Self::WatchUpdatesStream>, Status>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Produces a stream of updates representing the behavior of the instrumented async runtime.

fn watch_task_details<'life0, 'async_trait>( &'life0 self, req: Request<TaskDetailsRequest>, ) -> Pin<Box<dyn Future<Output = Result<Response<Self::WatchTaskDetailsStream>, Status>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Produces a stream of updates describing the activity of a specific task.

fn pause<'life0, 'async_trait>( &'life0 self, _req: Request<PauseRequest>, ) -> Pin<Box<dyn Future<Output = Result<Response<PauseResponse>, Status>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Registers that the console observer wants to pause the stream.

fn resume<'life0, 'async_trait>( &'life0 self, _req: Request<ResumeRequest>, ) -> Pin<Box<dyn Future<Output = Result<Response<ResumeResponse>, Status>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Registers that the console observer wants to resume the stream.