casper-node 2.0.3

The Casper blockchain node
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119

use std::{convert::Infallible, net::SocketAddr, sync::Arc, time::Duration};

use futures::{future, TryFutureExt};
use hyper::server::{conn::AddrIncoming, Builder};
use once_cell::sync::OnceCell;
use tokio::sync::oneshot;
use tower::builder::ServiceBuilder;
use tracing::{info, warn};
use warp::Filter;

use casper_types::ProtocolVersion;

use super::{filters, ReactorEventT};
use crate::effect::EffectBuilder;

/// Run the REST HTTP server.
///
/// A message received on `shutdown_receiver` will cause the server to exit cleanly.
pub(super) async fn run<REv: ReactorEventT>(
    builder: Builder<AddrIncoming>,
    effect_builder: EffectBuilder<REv>,
    api_version: ProtocolVersion,
    shutdown_receiver: oneshot::Receiver<()>,
    qps_limit: u64,
    local_addr: Arc<OnceCell<SocketAddr>>,
) {
    // REST filters.
    let rest_status = filters::create_status_filter(effect_builder, api_version);
    let rest_metrics = filters::create_metrics_filter(effect_builder);
    let rest_validator_changes =
        filters::create_validator_changes_filter(effect_builder, api_version);
    let rest_chainspec_filter = filters::create_chainspec_filter(effect_builder, api_version);

    let service = warp::service(
        rest_status
            .or(rest_metrics)
            .or(rest_validator_changes)
            .or(rest_chainspec_filter),
    );

    // Start the server, passing a oneshot receiver to allow the server to be shut down gracefully.
    let make_svc =
        hyper::service::make_service_fn(move |_| future::ok::<_, Infallible>(service.clone()));

    let rate_limited_service = ServiceBuilder::new()
        .rate_limit(qps_limit, Duration::from_secs(1))
        .service(make_svc);

    let server = builder.serve(rate_limited_service);
    if let Err(err) = local_addr.set(server.local_addr()) {
        warn!(%err, "failed to set local addr for reflection");
    }
    info!(address = %server.local_addr(), "started REST server");

    // Shutdown the server gracefully.
    let _ = server
        .with_graceful_shutdown(async move {
            shutdown_receiver.await.ok();
        })
        .map_err(|error| {
            warn!(%error, "error running REST server");
        })
        .await;
}

/// Run the REST HTTP server with CORS enabled.
///
/// A message received on `shutdown_receiver` will cause the server to exit cleanly.
pub(super) async fn run_with_cors<REv: ReactorEventT>(
    builder: Builder<AddrIncoming>,
    effect_builder: EffectBuilder<REv>,
    api_version: ProtocolVersion,
    shutdown_receiver: oneshot::Receiver<()>,
    qps_limit: u64,
    local_addr: Arc<OnceCell<SocketAddr>>,
    cors_origin: String,
) {
    // REST filters.
    let rest_status = filters::create_status_filter(effect_builder, api_version);
    let rest_metrics = filters::create_metrics_filter(effect_builder);
    let rest_validator_changes =
        filters::create_validator_changes_filter(effect_builder, api_version);
    let rest_chainspec_filter = filters::create_chainspec_filter(effect_builder, api_version);

    let service = warp::service(
        rest_status
            .or(rest_metrics)
            .or(rest_validator_changes)
            .or(rest_chainspec_filter)
            .with(match cors_origin.as_str() {
                "*" => warp::cors().allow_any_origin(),
                origin => warp::cors().allow_origin(origin),
            }),
    );

    // Start the server, passing a oneshot receiver to allow the server to be shut down gracefully.
    let make_svc =
        hyper::service::make_service_fn(move |_| future::ok::<_, Infallible>(service.clone()));

    let rate_limited_service = ServiceBuilder::new()
        .rate_limit(qps_limit, Duration::from_secs(1))
        .service(make_svc);

    let server = builder.serve(rate_limited_service);
    if let Err(err) = local_addr.set(server.local_addr()) {
        warn!(%err, "failed to set local addr for reflection");
    }
    info!(address = %server.local_addr(), "started REST server");

    // Shutdown the server gracefully.
    let _ = server
        .with_graceful_shutdown(async move {
            shutdown_receiver.await.ok();
        })
        .map_err(|error| {
            warn!(%error, "error running REST server");
        })
        .await;
}