Mqtt 5.0 Session expiration interval

[StanLi-ops]

Whether the framework has processed "Clean Start" and "Session Expiry Interval".
I don't quite understand the framework's handling of sessions.

Take https://github.com/ntex-rs/ntex-mqtt/blob/master/examples/session.rs as an example:
How is the "MySession" of the two clients A and B handled in the framework.
Suppose that client A has set "Clean Start: false, Session Expiry Interval: 30s", and reconnects to the original "MySession" within 30s of disconnection. Whether the original "MySession" is still there, and if so, where to get it.
If the framework does not handle "Clean Start" and "Session Expiry Interval", how should I manage the "MySession" of multiple clients in a unified way, I have no idea at first.

[fafhrd91]

ntex-mqtt does not handle session storage and session expiration. it is up to you to handle both storage and management of sessions

[StanLi-ops]

This is an example of my rewriting example/session.rs.
I want to confirm whether my implementation is correct:

use std::{
    collections::HashMap,
    sync::{Arc, Mutex},
    thread::{sleep, JoinHandle},
};

use futures::future::ok;
use ntex::service::{fn_factory_with_config, fn_service};
use ntex_mqtt::{
    v5::{
        self,
        codec::{Auth, DisconnectReasonCode},
    },
    MqttServer,
};

#[derive(Clone, Debug)]
pub struct Session {
    // our custom session information
    client_id: String,
}

#[derive(Debug)]
struct ServerError;

impl From<()> for ServerError {
    fn from(_: ()) -> Self {
        ServerError
    }
}

impl std::convert::TryFrom<ServerError> for v5::PublishAck {
    type Error = ServerError;

    fn try_from(err: ServerError) -> Result<Self, Self::Error> {
        Err(err)
    }
}

async fn handshake_v5<Io>(
    handshake: v5::Handshake<Io>,
    mut session_list: SessionList,
) -> Result<v5::HandshakeAck<Io, Session>, ServerError> {
    let id = handshake.packet().client_id.to_string();

    let have = match session_list.list.get(&id) {
        Some(_) => true,
        None => false,
    };

    let clean_start = handshake.packet().clean_start;

    let session = match (have, clean_start) {
        (true, true) => {
            println!("clean old session");
            let session = Session { client_id: id.clone() };
            session_list.clean_start(&id, session.clone());
            session
        }
        (true, false) => {
            println!("use old session");
            session_list.stop_remove(&id);
            session_list.get(&id).unwrap()
        }
        (false, true) => {
            println!("new session");
            let session = Session { client_id: id.clone() };
            session_list.list.insert(id.clone(), session.clone());
            session
        }
        (false, false) => {
            println!("new session");
            let session = Session { client_id: id.clone() };
            session_list.list.insert(id.clone(), session.clone());
            session
        }
    };
    Ok(handshake.ack(session))
}

async fn control_v5<E>(
    session: v5::Session<Session>,
    mut session_list: SessionList,
    control: v5::ControlMessage<E>,
) -> Result<v5::ControlResult, ServerError> {
    match control {
        v5::ControlMessage::Auth(a) => Ok(a.ack(Auth::default())),
        v5::ControlMessage::Ping(p) => Ok(p.ack()),
        v5::ControlMessage::Disconnect(d) => {
            let id = &session.client_id;
            match d.packet().session_expiry_interval_secs {
                Some(secs) => match secs {
                    0xFFFFFFFF => Ok(d.ack()),
                    _ => {
                        session_list.remove(id, secs);
                        Ok(d.ack())
                    }
                },
                None => Ok(d.ack()),
            }
        }
        v5::ControlMessage::Subscribe(s) => Ok(s.ack()),
        v5::ControlMessage::Unsubscribe(u) => Ok(u.ack()),
        v5::ControlMessage::Closed(c) => Ok(c.ack()),
        v5::ControlMessage::Error(e) => Ok(e.ack(DisconnectReasonCode::UnspecifiedError)),
        v5::ControlMessage::ProtocolError(pe) => Ok(pe.ack()),
    }
}

#[derive(Debug, Clone)]
pub struct SessionList {
    list: HashMap<String, Session>,
    delete_list: Arc<Mutex<HashMap<String, (JoinHandle<()>, std::sync::mpsc::Sender<()>)>>>,
}
impl SessionList {
    pub fn new() -> Self {
        Self { list: HashMap::new(), delete_list: Arc::new(Mutex::new(HashMap::new())) }
    }
    pub fn remove(&mut self, client_id: &str, secs: u32) {
        match secs {
            0 => match self.list.remove(client_id) {
                Some(s) => println!("remove client {}", s.client_id),
                None => println!("client id does not exist"),
            },
            _ => {
                while let Ok(mut delete_list) = self.delete_list.lock() {
                    let id = client_id.to_string();
                    let mut list = self.list.clone();
                    let (tx, rx) = std::sync::mpsc::channel::<()>();
                    let delete_thread = std::thread::spawn(move || {
                        for i in 0..secs + 1 {
                            if let Ok(_) = rx.try_recv() {
                                break;
                            } else {
                                while i == secs {
                                    match list.remove(&id) {
                                        Some(s) => println!("remove client {}", s.client_id),
                                        None => println!("client id does not exist"),
                                    }
                                }
                                sleep(std::time::Duration::from_secs(1));
                            }
                        }
                    });
                    delete_list.insert(client_id.to_string(), (delete_thread, tx));
                }
            }
        };
    }

    pub fn stop_remove(&mut self, client_id: &str) {
        while let Ok(mut delete_list) = self.delete_list.lock() {
            match delete_list.get_mut(client_id) {
                Some((_, tx)) => {
                    tx.send(()).unwrap();
                    delete_list.remove(client_id);
                }
                None => println!("client id does not exist"),
            }
        }
    }

    pub fn get(&self, id: &str) -> Option<Session> {
        match self.list.get(id) {
            Some(s) => Some(s.clone()),
            None => None,
        }
    }

    pub fn clean_start(&mut self, id: &str, session: Session) {
        if let Some(s) = self.list.get_mut(&id.to_string()) {
            *s = session
        } else {
            self.list.insert(id.to_string(), session);
        }
    }
}

#[ntex::main]
async fn main() -> std::io::Result<()> {
    std::env::set_var("RUST_LOG", "session=trace,ntex=trace,ntex_mqtt=trace,basic=trace");
    env_logger::init();

    log::info!("Hello");

    let session_list = SessionList::new();

    ntex::server::Server::build()
        .bind("mqtt", "127.0.0.1:1883", move || {
            let session_list_handshake = session_list.clone();
            let session_list_control = session_list.clone();
            MqttServer::new().v5(v5::MqttServer::new(fn_factory_with_config(move |_| {
                let session_list = session_list_handshake.clone();
                ok::<_, ()>(fn_service(move |handshake| {
                    handshake_v5(handshake, session_list.clone())
                }))
            }))
            // contro
            .control(fn_factory_with_config(
                move |session: v5::Session<Session>| {
                    let session_list = session_list_control.clone();
                    ok::<_, ServerError>(fn_service(move |req| {
                        control_v5(session.clone(), session_list.clone(), req)
                    }))
                },
            )))
        })?
        .workers(1)
        .run()
        .await
}

[fafhrd91]

looks correct

[StanLi-ops]

Grateful