You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: hall.rs
    path of file: ./repos/SquireCore/squire_sdk/src/server/gathering
    
    the code of the file until where you have to start completion: use std::collections::{HashMap, HashSet};

use async_trait::async_trait;
use axum::extract::ws::WebSocket;
use instant::{Duration, Instant};
use squire_lib::tournament::TournamentId;
use tokio::sync::{
    mpsc::{channel, Receiver, Sender},
    oneshot::channel as oneshot_channel,
};

use super::{Gathering, GatheringMessage, PersistMessage, PersistReadyMessage};
use crate::{
    actor::{ActorBuilder, ActorClient, ActorState, Scheduler},
    server::session::SessionWatcher,
    sync::TournamentManager,
};

/* TODO:
 *  - Clients might close their websocket communicates. Because of this, we need to ensure that the
 *  `SelectAll` stream doesn't yield empty frames endlessly.
 *    - Similarly, we need to understand the behavior if all stream close
 *
 *  - The 
    
    async fn process(&mut self, scheduler: &mut Scheduler<Self>, msg: Self::Message) {
        match msg {
            GatheringHallMessage::NewGathering(id) => self.process_new_gathering(id).await,
            GatheringHallMessage::NewConnection(id, user, ws) => {
                self.process_new_onlooker(id, user, ws).await
            }
            GatheringHallMessage::Persist => {
                let mut to_persist = HashSet::new();
                let mut persist_reqs = HashMap::new();
                while let Ok(PersistReadyMessage(id)) = self.persists.try_recv() {
                    let _ = to_persist.insert(id);
                }
                for id in to_persist.drain() {
                    let sender = self.gatherings.get_mut(&id).unwrap();
                    let (send, recv) = oneshot_channel();
                    let msg = GatheringMessage::GetTournament(send);
                    sender.send(msg);
                    let tourn = recv.await.unwrap();
                    let _ = persist_reqs.insert(id, tourn);
                }

                persist_reqs
                    .drain()
                    .for_each(|(_, tourn)| self.persister.send(tourn));
                schedule_persist(scheduler);
            }
        }
    }