product-os-server 0.0.55

Product OS : Server provides a full functioning advanced server capable of acting as a web server, command and control distributed network, authentication server, crawling server and more. Fully featured with high level of flexibility.
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

//! Flow-compatible HTTP server implementation
//!
//! Serves the Flow router (which implements `tower::Service`) directly via
//! hyper, without going through `axum::Router`. Activated when only the
//! `framework_flow` feature is enabled.

#![cfg(all(feature = "framework_flow", feature = "executor_tokio"))]

use std::prelude::v1::*;
use core::error::Error;

use product_os_net::SocketAddr;
use product_os_router::ProductOSRouter;
use product_os_async_executor::{Executor, TokioExecutor};

/// Creates an HTTP server that serves a `ProductOSRouter` directly.
///
/// The router implements `tower::Service<Request<Body>>` so it can be used
/// as the inner service for hyper connections.
///
/// When `with_connect_info` is `true`, the peer `SocketAddr` is injected into
/// the request's extensions so handlers can extract the client address.
pub async fn create_flow_http_service<S>(
    socket_address: SocketAddr,
    router: ProductOSRouter<S>,
    with_connect_info: bool,
) -> Result<(), Box<dyn Error + Send + Sync>>
where
    S: Clone + Send + Sync + 'static,
{
    tracing::info!("Starting Flow HTTP server listening on {}", socket_address);

    let listener = tokio::net::TcpListener::bind(socket_address)
        .await
        .map_err(|e| -> Box<dyn Error + Send + Sync> { Box::new(e) })?;

    loop {
        let (cnx, addr) = match listener.accept().await {
            Ok(conn) => conn,
            Err(e) => {
                tracing::error!("Failed to accept connection: {:?}", e);
                continue;
            }
        };

        let router = router.clone();

        tokio::spawn(async move {
            let stream = hyper_util::rt::TokioIo::new(cnx);

            let hyper_svc = hyper::service::service_fn(move |mut req: hyper::Request<hyper::body::Incoming>| {
                let mut router = router.clone();
                let peer_addr = addr;
                async move {
                    #[cfg(all(feature = "ws", not(feature = "flow_axum_compat")))]
                    if hyper_tungstenite::is_upgrade_request(&req) {
                        if let Some(ws_handler) = router.lookup_ws_handler(req.uri().path()) {
                            match hyper_tungstenite::upgrade(&mut req, None) {
                                Ok((response, websocket)) => {
                                    tokio::spawn(async move {
                                        match websocket.await {
                                            Ok(stream) => {
                                                let ws = product_os_router::flow_impl::ws::WebSocket::new(stream);
                                                ws_handler(ws).await;
                                            }
                                            Err(e) => {
                                                tracing::warn!("WebSocket upgrade failed: {:?}", e);
                                            }
                                        }
                                    });
                                    let (parts, body) = response.into_parts();
                                    let mapped = product_os_router::Body::from(body);
                                    return Ok(hyper::Response::from_parts(parts, mapped));
                                }
                                Err(e) => {
                                    tracing::warn!("WebSocket upgrade error: {:?}", e);
                                }
                            }
                        }
                    }

                    use product_os_router::Service;
                    let (mut parts, body) = req.into_parts();
                    #[cfg(not(feature = "flow_axum_compat"))]
                    let body = product_os_router::Body::from(body);
                    #[cfg(feature = "flow_axum_compat")]
                    let body = product_os_router::Body::new(body);
                    if with_connect_info {
                        parts.extensions.insert(
                            product_os_router::flow_impl::extractors::context::ConnectInfo(peer_addr),
                        );
                        #[cfg(feature = "flow_axum_compat")]
                        parts.extensions.insert(
                            axum::extract::ConnectInfo(peer_addr),
                        );
                    }
                    let req = hyper::Request::from_parts(parts, body);
                    router.call(req).await
                }
            });

            match TokioExecutor::context_sync() {
                Ok(executor) => {
                    let result = hyper_util::server::conn::auto::Builder::new(executor)
                        .serve_connection_with_upgrades(stream, hyper_svc)
                        .await;
                    if let Err(err) = result {
                        tracing::warn!("error serving connection from {}: {}", addr, err);
                    }
                }
                Err(e) => tracing::error!("Failed to get executor for connection from {}: {:?}", addr, e),
            }
        });
    }
}