product-os-server 0.0.55

//! HTTP server implementation module
//!
//! This module provides the HTTP (non-TLS) server implementation using Hyper and Tokio.
//!
//! # Overview
//!
//! The HTTP server creates a TCP listener and handles incoming connections asynchronously.
//! Each connection is spawned as a separate Tokio task for concurrent request handling.
//!
//! # Features
//!
//! - **Connection Info**: Optional extraction of client connection information (IP address, port)
//! - **HTTP/2 Support**: Automatic protocol negotiation via hyper-util
//! - **Graceful Upgrades**: Support for WebSocket and other connection upgrades
//!
//! # Requirements
//!
//! This module requires the `executor_tokio` feature to be enabled.

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

use std::prelude::v1::*;

use core::error::Error;

use product_os_net::SocketAddr;
use hyper::service::Service;
use product_os_async_executor::{Executor, TokioExecutor};
use product_os_router::{Request, Service as RouterService, Router };

/// Creates an HTTP server bound to the given socket address using the Tokio runtime.
///
/// # Arguments
///
/// * `socket_address` - The address and port to bind to
/// * `router` - The axum router to handle requests
/// * `with_connect_info` - Whether to extract connection info for handlers
///
/// # Errors
///
/// Returns an error if the TCP listener cannot bind to the given address.
pub async fn create_http_tokio_service(socket_address: SocketAddr, router: Router, with_connect_info: bool) -> Result<(), Box<dyn Error + Send + Sync>> {
    tracing::info!("Starting 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;
            }
        };

        match with_connect_info {
            true => {
                let make_service = hyper_util::service::TowerToHyperService::new(router.clone().into_make_service_with_connect_info::<SocketAddr>());
                let tower_service = match make_service.call(socket_address).await {
                    Ok(svc) => svc,
                    Err(e) => {
                        tracing::error!("Failed to create tower service: {:?}", e);
                        continue;
                    }
                };

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

                    let hyper_service = hyper::service::service_fn(move |request: Request<hyper::body::Incoming>| {
                        tower_service.clone().call(request)
                    });

                    match TokioExecutor::context_sync() {
                        Ok(executor) => {
                            let _ret = hyper_util::server::conn::auto::Builder::new(executor)
                                .serve_connection_with_upgrades(stream, hyper_service)
                                .await;
                        }
                        Err(e) => tracing::error!("Failed to get executor for connection from {}: {:?}", addr, e)
                    }
                });
            }
            false => {
                let tower_service = router.clone();

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

                    let hyper_service = hyper::service::service_fn(move |request: Request<hyper::body::Incoming>| {
                        tower_service.clone().call(request)
                    });

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