micro-tower 0.1.0

micro-tower is a framework to generate and manage micro services.
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

use std::net::SocketAddr;

use bytes::BytesMut;
use tokio::net::TcpListener;
use tokio::task::JoinHandle;
use tower::{BoxError, Service, ServiceExt};

use crate::{api, shutdown::Controller};

pub fn spawn<SB>(
    addr: SocketAddr,
    mut builder: SB,
    controller: &Controller,
) -> JoinHandle<Result<(), BoxError>>
where
    SB: Service<SocketAddr, Error = BoxError> + Send + 'static,
    SB::Future: Send,
    SB::Response: Service<BytesMut, Response = BytesMut, Error = api::Error> + Send,
    <SB::Response as Service<BytesMut>>::Future: Send,
{
    let controller = controller.clone();
    tokio::spawn(async move {
        let listener = TcpListener::bind(addr).await?;
        tracing::info!(message = "listening on", port = addr.port());

        loop {
            tracing::trace!(message = "wait for new connections", port = addr.port());

            let (stream, addr) = tokio::select! {
                result = listener.accept() => result?,
                _ = controller.wait_for_shutdown() => return Ok(())
            };

            let service = match builder.ready().await {
                Ok(service) => service,
                Err(err) => {
                    let report = crate::report!(err.as_ref());
                    tracing::error!("{report:?}");
                    continue;
                }
            };
            let service = match service.call(addr).await {
                Ok(service) => service,
                Err(err) => {
                    let report = crate::report!(err.as_ref());
                    tracing::error!("{report:?}");
                    continue;
                }
            };

            tracing::info!(message = "new connection", addr = format!("{addr}"));

            let controller = controller.clone();
            tokio::spawn(async move {
                if let Err(err) = super::stream::spawn_fut(stream, service, controller).await {
                    let report = crate::report!(err.as_ref());
                    tracing::error!("{report:?}");
                }
            });
        }
    })
}