Compare commits
2
Commits
cb8447224f
...
b9e44e1fa3
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b9e44e1fa3 | ||
|
|
acf5ea8ebe |
No files matched your search
+37
-14
@@ -557,22 +557,45 @@ async fn handle_socket(socket: WebSocket, state: Arc<AppState>, client_type: Str
|
||||
);
|
||||
});
|
||||
|
||||
tokio::select! {
|
||||
_ = (&mut send_task) => {
|
||||
tracing::info!("Websocket send task finished for session {}", session_id);
|
||||
recv_task.abort();
|
||||
},
|
||||
_ = (&mut recv_task) => {
|
||||
tracing::info!("Websocket recv task finished for session {}", session_id);
|
||||
send_task.abort();
|
||||
},
|
||||
struct SessionCleanup {
|
||||
session_id: String,
|
||||
state: Arc<AppState>,
|
||||
send_task: Option<tokio::task::JoinHandle<()>>,
|
||||
recv_task: Option<tokio::task::JoinHandle<()>>,
|
||||
}
|
||||
|
||||
impl Drop for SessionCleanup {
|
||||
fn drop(&mut self) {
|
||||
self.state.clients.write().unwrap().remove(&self.session_id);
|
||||
if let Some(task) = self.send_task.take() {
|
||||
task.abort();
|
||||
}
|
||||
if let Some(task) = self.recv_task.take() {
|
||||
task.abort();
|
||||
}
|
||||
tracing::info!(
|
||||
"Websocket session {} closed and cleaned up",
|
||||
self.session_id
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
let mut cleanup = SessionCleanup {
|
||||
session_id: session_id.clone(),
|
||||
state: Arc::clone(&state),
|
||||
send_task: Some(send_task),
|
||||
recv_task: Some(recv_task),
|
||||
};
|
||||
|
||||
state.clients.write().unwrap().remove(&session_id);
|
||||
tracing::info!(
|
||||
"Websocket session {} closed and removed from state",
|
||||
session_id
|
||||
);
|
||||
tokio::select! {
|
||||
_ = cleanup.send_task.as_mut().unwrap() => {
|
||||
tracing::info!("Websocket send task finished for session {}", session_id);
|
||||
},
|
||||
_ = cleanup.recv_task.as_mut().unwrap() => {
|
||||
tracing::info!("Websocket recv task finished for session {}", session_id);
|
||||
},
|
||||
};
|
||||
// Drop guard automatically handles removal and aborts the other task.
|
||||
}
|
||||
|
||||
#[derive(serde::Deserialize, serde::Serialize, Debug)]
|
||||
|
||||
+7
-4
@@ -117,20 +117,23 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
});
|
||||
|
||||
tokio::select! {
|
||||
_ = shutdown_rx.recv() => { tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
|
||||
_ = shutdown_rx.recv() => {
|
||||
tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
|
||||
tracing::info!("Shutdown received while connected");
|
||||
|
||||
break;
|
||||
}
|
||||
_ = &mut send_task => {
|
||||
tracing::error!("Send task exited");
|
||||
tokio::time::sleep(tokio::time::Duration::from_millis(50)).await; recv_task.abort();
|
||||
tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
|
||||
recv_task.abort();
|
||||
break;
|
||||
}
|
||||
_ = &mut recv_task => {
|
||||
tracing::error!("Recv task exited");
|
||||
send_task.abort();
|
||||
}
|
||||
}
|
||||
tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::error!("Failed to connect via WSS: {}", e);
|
||||
|
||||
Reference in new issue
Block a user