pass-it-on 0.17.3

A library that provides a simple notification client and server that receives messages and passes them on to endpoints
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

use crate::interfaces::http::{BASE_PATH, NOTIFICATION_PATH, VERSION_PATH, Version};
use crate::notifications::Notification;
use axum::extract::State;
use axum::http::StatusCode;
use axum::routing::{get, post};
use axum::{Json, Router};
use axum_server::Address;
use axum_server::tls_rustls::RustlsConfig;
use std::net::SocketAddr;
use std::path::Path;
use std::time::Duration;
use tokio::sync::{mpsc, watch};
use tracing::{debug, error, info, trace, warn};
use crate::Error;

const GRACE_PERIOD: Duration = Duration::from_secs(1);

pub(super) async fn start_monitoring<P: AsRef<Path>> (
    tx: mpsc::Sender<String>,
    shutdown: watch::Receiver<bool>,
    socket: SocketAddr,
    tls: bool,
    tls_cert_path: Option<P>,
    tls_key_path: Option<P>,
) -> Result<(), Error> {
    let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
    let handle = axum_server::Handle::new();
    tokio::spawn(shutdown_server(handle.clone(), shutdown));

    let routes = Router::new()
        .route(format!("/{}/{}", BASE_PATH, VERSION_PATH).as_str(), get(version_handler))
        .route(format!("/{}/{}", BASE_PATH, NOTIFICATION_PATH).as_str(), post(notification_handler))
        .with_state(tx);

    info!("Setting up Interface: HttpSocket on -> {} | TLS Enabled -> {}", socket, tls);
    let listener = std::net::TcpListener::bind(socket)?;
    listener.set_nonblocking(true)?;
    
    match tls {
        true => {
            let config = RustlsConfig::from_pem_file(tls_cert_path.unwrap(), tls_key_path.unwrap())
                .await?;
            axum_server::from_tcp_rustls(listener, config)?
                .serve(routes.into_make_service())
                .await?;
        }
        false => {
            axum_server::from_tcp(listener)?
                .handle(handle)
                .serve(routes.into_make_service())
                .await?
        }
    };
    Ok(())
}

async fn version_handler() -> Json<Version> {
    Json(Version::new())
}

async fn notification_handler(
    State(tx): State<mpsc::Sender<String>>,
    Json(notification): Json<Notification>,
) -> StatusCode {
    trace!("HTTP server received {:?}", notification);
    match tx.send(notification.to_json().unwrap_or_default()).await {
        Ok(_) => StatusCode::OK,
        Err(e) => {
            warn!("bad JSON received from http socket: {}", e);
            StatusCode::BAD_REQUEST
        }
    }
}

async fn shutdown_server<A: Address>(handle: axum_server::Handle<A>, mut shutdown: watch::Receiver<bool>) {
    match shutdown.changed().await {
        Ok(_) => {
            debug!("http_server starting graceful shutdown");
            handle.graceful_shutdown(Some(GRACE_PERIOD));
        }
        Err(e) => {
            error!("Shutdown Receive Error: {}", e);
        }
    }
}