How to implement shutdown on inactivity

[kulak]

How would one implement web application shutdown on inactivity? This would be useful in systemd socket activation scenarios and other cases when request is dispatched via Unix domain sockets, which ntex supports. The later I know ntex supports well as I got it work.

In my previous GO language projects I basically have middleware which processes all incoming requests and records time of the last response. Then there is usually a background routine that checks the last response time and initiates clean application shutdown.

I am looking to do the same with ntex, but I am new to ntex, tokio async runtime and RUST itself, so I it hard to take on this type of project without pointers.

[kulak]

Here is what I did. It works, but I admit that tokio threading is currently above my knowledge level. RUST fragmentation when working with threading, synchronization and time does not help when starting with it.

main.rs file:

use anyhow::Result;
use dotenvy;
use flexi_logger::Logger;
use ntex::web::{self, middleware, App, HttpRequest};
use std::sync::Arc;
use time::OffsetDateTime;
use tokio::{sync::Mutex, spawn};
use log::{info, warn};

mod inactivity;

async fn index(req: HttpRequest) -> &'static str {
    println!("REQ: {:?}", req.head().uri.path());
    "Hello world!"
}

#[ntex::main]
async fn main() -> Result<()> {
    dotenvy::dotenv()?;

    let _logger = Logger::try_with_str("info")?.start();
    warn!("Started Application");

    let last_time = Arc::new(Mutex::new(OffsetDateTime::now_utc()));
    let server_last_time = last_time.clone();

    let server = web::server(move || {
        App::new()
            .wrap(middleware::Logger::default())
            .wrap(inactivity::Inactivity {
                last_accessed: server_last_time.clone(),
            })
            .service((
                web::resource("/index.html").to(|| async { "Hello world!" }),
                web::resource("/").to(index),
            ))
    })
    .bind("127.0.0.1:8080")?
    .run();

    let shutdown_server = server.clone();
    spawn(async move {
        inactivity::until_idle(last_time).await;
        info!("reached limit of inactivity, shutting down the server");
        shutdown_server.stop(true).await;
    });

    server.await?;

    warn!("Finished Application");
    Ok(())
}

inactivity.rs:

use ntex::service::{Middleware, Service, ServiceCtx};
use ntex::util::BoxFuture;
use ntex::web::{Error, WebRequest, WebResponse};
use std::sync::Arc;
use tokio::sync::Mutex;
use tokio::time::interval;
use time::OffsetDateTime;
use core::time::Duration;
use log::info;

pub struct Inactivity {
    pub last_accessed: Arc<Mutex<OffsetDateTime>>,
}

// Middleware factory is `Middleware` trait from ntex-service crate
// `S` - type of the next service
// `B` - type of response's body
impl<S> Middleware<S> for Inactivity {
    type Service = InactivityMiddleware<S>;

    fn create(&self, service: S) -> Self::Service {
        InactivityMiddleware { 
            service: service,
            last_accessed: self.last_accessed.clone(),
        }
    }
}

pub struct InactivityMiddleware<S> {
    service: S,
    last_accessed: Arc<Mutex<OffsetDateTime>>,
}

impl<S, Err> Service<WebRequest<Err>> for InactivityMiddleware<S>
where
    Err: 'static,
    S: Service<WebRequest<Err>, Response = WebResponse, Error = Error>,
{
    type Response = WebResponse;
    type Error = Error;
    type Future<'f> = BoxFuture<'f, Result<Self::Response, Self::Error>> where S: 'f;

    ntex::forward_poll_ready!(service);
    ntex::forward_poll_shutdown!(service);

    fn call<'a>(
        &'a self,
        req: WebRequest<Err>,
        ctx: ServiceCtx<'a, Self>,
    ) -> Self::Future<'a> {
        // println!("Hi from start. You requested: {}", req.path());

        Box::pin(async move {
            let mut lock = self.last_accessed.lock().await;
            *lock = OffsetDateTime::now_utc();

            let res = ctx.call(&self.service, req).await?;

            // println!("Hi from response");
            Ok(res)
        })
    }
}

pub async fn until_idle(last_time: Arc<Mutex<OffsetDateTime>>) {
    let mut interval = interval(Duration::from_secs(5));
    loop {
        {                
            let now = OffsetDateTime::now_utc();
            let last = last_time.lock().await;
            let idle = now - *last;
            // info!("inactivity check... {}", idle.whole_seconds());
            if idle.whole_seconds() > 7 {
                info!("inactivity limit reached");
                break;
            };
        }
        interval.tick().await;
    }
}

I'd like to put this part of main code into inactivity to make middleware reusable. Unfortunately, I can't figure out how to take server as argument in the function, because server definitions are so loaded with generic signature.

    let shutdown_server = server.clone();
    spawn(async move {
        inactivity::until_idle(last_time).await;
        info!("reached limit of inactivity, shutting down the server");
        shutdown_server.stop(true).await;
    });

Here is repository: https://github.com/kulak/ntex-middleware

[fafhrd91]

I think this is right approach. but I would use Arc if high load is required.