Asynchronously wait for x seconds before executing function

I wish to write a web server that collects information about a set of incoming connections. After receiving the first connection, I want a method to wait x seconds before executing a return function on the gathered data.

If anyone knows how this functionality can be achieved, I would greatly appreciate the help!

Here is my current implementation:

main.rs

use actix_web::{get, web, App, HttpRequest, HttpResponse, HttpServer, Responder};
use std::collections::HashMap;
use std::sync::Mutex;
mod handler;

#[get("/some/url/{key}={value}")]
async fn new_prop<T: FnMut(String, &handler::Fingerprint)>(
    con_handler: web::Data<handler::ConnectionHandler<T>>,
    request: HttpRequest,
    web::Path((key, value)): web::Path<(String, String)>,
) -> impl Responder {
    // Just an example (will not work due to threading issues)
    con_handler.insert(
        request
            .connection_info()
            .realip_remote_addr()
            .unwrap()
            .to_owned(),
        key,
        value,
        Some(request.headers()),
    );
    HttpResponse::Gone()
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    const PORT: u16 = 8000;
    println!("🚀 Server running on port: {}!", PORT);

    // Define the connection handler
    let h = web::Data::new(handler::ConnectionHandler {
        data: HashMap::new(),
        return_fn: |ip: String, fingerprint: &handler::Fingerprint| {
            println!("IP: {}, Fingerprint: {:#?}", ip, fingerprint)
        },
    });

    HttpServer::new(|| {
        App::new()
            .app_data(h.clone())
            .service(new_prop)
    })
    .bind(format!("127.0.0.1:{}", PORT))?
    .run()
    .await
}

handler.rs

use actix_web::http::header::HeaderMap;
use chrono::{DateTime, Utc};
use std::collections::HashMap;
use std::sync::Mutex;

#[derive(Debug, Clone)]
pub struct Fingerprint {
    properties: Vec<(String, String)>,
    fonts: Vec<String>,
    headers: HeaderMap,
    timestamp: DateTime<Utc>,
}

#[derive(Clone)]
pub struct ConnectionHandler<T: FnMut(String, &Fingerprint)> {
    data: HashMap<String, Fingerprint>,
    return_fn: T,
}

impl<T> ConnectionHandler<T>
where
    T: FnMut(String, &Fingerprint),
{
    pub fn insert(&mut self, ip: String, key: String, value: String, headers: Option<&HeaderMap>) {
        match self.data.get_mut(&ip) {
            Some(f) => {
                if key == "font-name" {
                    f.fonts.push(value)
                } else {
                    f.properties.push((key, value))
                }
            }
            None => {
                self.data.insert(
                    ip.clone(),
                    Fingerprint {
                        properties: Vec::new(),
                        fonts: Vec::new(),
                        headers: headers.unwrap().clone(),
                        timestamp: chrono::offset::Utc::now(),
                    },
                );
                self.insert(ip, key, value, None);

                // Wait x seconds before calling this!
                (self.return_fn)(ip, self.data.get(&ip).unwrap());
            }
        }
    }
}

The actix-web framework is based on Tokio, so maybe you are looking for tokio::time::sleep()?

Sadly I get the following error when I try that solution:
thread 'actix-rt:worker:0' panicked at 'there is no reactor running, must be called from the context of a Tokio 1.x runtime'

Is there a workaround for this?

note that if you are using actix-web v3, that depends an older version of tokio, so perhaps try delay_for from tokio v0.2

Thanks that resolved the issue! What is the logic behind actix-web depending on an older version of tokio?

The next release, actix-web 4.0, uses the latest version of Tokio, but it is still in beta testing. If you want to try it out before it is released, you can add a dependency on the beta version:

[dependencies]
actix-web = "4.0.0-beta.9"

The change log has a list of changes since the last actix-web 3.3 release.

1 Like