comprehensive_warm_channels 0.6.1

A harness for creating consistently-shaped servers will less boilerplate
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
120
121
122
123
124
125
126
127

//! [`comprehensive`] integration for [`warm_channels`]
//!
//! This crate defines a single Comprehensive [`Resource`]
//! [`WarmChannelsDiag`] which makes diagnostic information about all
//! [`warm_channels`] client channels available over HTTP.
//!
//! This is a separate crate for dependency reasons (to avoid a crate in the
//! `comprehensive` repo depending on `warm_channels` which depends back on
//! another crate in the `comprehensive` repo).

#![warn(missing_docs)]

use comprehensive::v1::{AssemblyRuntime, Resource, resource};
use comprehensive_traits::http_diag::{Body, DiagRequest, HttpDiagHandler, HttpDiagHandlerInstaller};
use pin_project_lite::pin_project;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use tower::util::BoxCloneSyncService;
use tower_service::Service;
use warm_channels::ChannelDiagService;

pub use warm_channels;

#[doc(hidden)]
#[derive(clap::Args, Debug)]
#[group(skip)]
pub struct WarmChannelsDiagArgs {
    #[arg(long, default_value = "/debug/channels")]
    warm_channels_diag_path: String,
}

/// Diagnostics introspection HTTP handler for [`warm_channels`].
///
/// If this resource and [`comprehensive_http::diag::HttpServer`] are both
/// in a [`comprehensive::Assembly`] then a diagnostics page about active
/// channels is made available, by default at path `/debug/channels`.
///
/// It should not be necessary to depend on this resource directly as gRPC
/// clients created with [`comprehensive_grpc::GrpcClient`] declare a
/// dependency on it.
///
/// # Command line flags
///
/// | Flag                        | Default           | Meaning               |
/// |-----------------------------|-------------------|-----------------------|
/// | `--warm-channels-diag-path` | `/debug/channels` | HTTP path to serve at |
///
/// [`comprehensive_http::diag::HttpServer`]: https://docs.rs/comprehensive_http/latest/comprehensive_http/diag/type.HttpServer.html
/// [`comprehensive_grpc::GrpcClient`]: https://docs.rs/comprehensive_grpc/latest/comprehensive_grpc/derive.GrpcClient.html
pub struct WarmChannelsDiag(WarmChannelsDiagArgs);

#[resource]
#[export(dyn HttpDiagHandler)]
impl Resource for WarmChannelsDiag {
    fn new(
        _: comprehensive::NoDependencies,
        a: WarmChannelsDiagArgs,
        _: &mut AssemblyRuntime<'_>,
    ) -> Result<Arc<Self>, std::convert::Infallible> {
        Ok(Arc::new(Self(a)))
    }
}

pin_project! {
    struct MapResponseBodyFuture<F> {
        #[pin] inner: F
    }
}

impl<F, B, E> Future for MapResponseBodyFuture<F>
where
    F: Future<Output = Result<http::Response<B>, E>>,
    B: Into<Body>,
{
    type Output = Result<http::Response<Body>, E>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        match self.project().inner.poll(cx) {
            Poll::Ready(Ok(r)) => {
                let (parts, body) = r.into_parts();
                Poll::Ready(Ok(http::Response::from_parts(parts, body.into())))
            }
            Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
            Poll::Pending => Poll::Pending,
        }
    }
}

#[derive(Clone)]
struct WarmChannelsDiagService<T>(T);

impl<T, B> Service<DiagRequest> for WarmChannelsDiagService<T>
where
    T: Service<http::Request<Body>, Response = http::Response<B>>,
    B: Into<Body>,
{
    type Response = http::Response<Body>;
    type Error = <T as Service<http::Request<Body>>>::Error;
    type Future = MapResponseBodyFuture<<T as Service<http::Request<Body>>>::Future>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        Service::poll_ready(&mut self.0, cx)
    }

    fn call(&mut self, request: DiagRequest) -> Self::Future {
        let original_uri = request.get_original_uri().cloned();
        let mut req = request.into_inner();
        if let Some(uri) = original_uri {
            req.extensions_mut().insert(warm_channels::OriginalUri(uri));
        }
        MapResponseBodyFuture {
            inner: Service::call(&mut self.0, req),
        }
    }
}

impl HttpDiagHandler for WarmChannelsDiag {
    fn install_handlers(self: Arc<Self>, installer: &mut dyn HttpDiagHandlerInstaller) {
        if !self.0.warm_channels_diag_path.is_empty() {
            installer.nest_diag_service(
                &self.0.warm_channels_diag_path,
                BoxCloneSyncService::new(WarmChannelsDiagService(ChannelDiagService::default())),
            );
        }
    }
}