273bd1b
use std::collections::HashMap;
use std::time::Instant;
use tokio::sync::mpsc;
use tokio::time::{MissedTickBehavior, interval};
use super::io::{Input, Output};
use super::state::Session;
use crate::SocketHub;
use crate::engine::{self, Packet as EnginePacket};
pub(crate) async fn run(
hub: SocketHub,
sid: String,
user_id: i64,
access_token: String,
input: mpsc::Sender<Input>,
mut receiver: mpsc::Receiver<Input>,
output: Output,
send_open: bool,
) {
let options = hub.options().clone();
let mut shutdown = hub.shutdown_receiver();
let mut session = Session {
hub,
sid,
user_id,
access_token,
device_id: String::new(),
deliveries: HashMap::new(),
input,
output,
namespaces: HashMap::new(),
pending_binary: None,
pending_acks: HashMap::new(),
next_ack: 0,
last_pong: Instant::now(),
last_actor_heartbeat: Instant::now(),
max_payload: options.max_payload,
auth_deadline: (user_id == 0).then(|| tokio::time::Instant::now() + options.auth_timeout),
};
if send_open {
let open = engine::open(
&session.sid,
options.heartbeat_interval,
options.heartbeat_timeout,
options.max_payload,
&[],
);
if !session.output.send(vec![open]).await {
session.cleanup().await;
return;
}
}
let mut heartbeat = interval(options.heartbeat_interval);
heartbeat.set_missed_tick_behavior(MissedTickBehavior::Delay);
heartbeat.tick().await;
let mut maintenance = interval(std::time::Duration::from_secs(1));
loop {
if *shutdown.borrow() {
break;
}
tokio::select! {
_ = shutdown.changed() => break,
_ = maintenance.tick() => {
let now = tokio::time::Instant::now();
if session.auth_deadline.is_some_and(|deadline| now >= deadline) {
break;
}
session.pending_acks.retain(|_, (_, deadline, sender)| *deadline > now && !sender.is_closed());
}
incoming = receiver.recv() => {
let Some(input) = incoming else {
break;
};
if !session.handle(input).await {
break;
}
}
_ = heartbeat.tick() => {
let elapsed = session.last_pong.elapsed();
if elapsed >= options.heartbeat_interval + options.heartbeat_timeout {
break;
}
if !session
.output
.send(vec![EnginePacket::Ping(String::new())])
.await
{
break;
}
}
}
}
session.cleanup().await;
}