Function ws_main_loop

crates/client-api/src/routes/subscribe.rs:613–709

The main `select!` loop of the websocket client actor. > This function is defined standalone with generic parameters so that its > behavior can be tested in isolation, not requiring I/O and allowing to > mock effects easily. The loop's responsibilities are: - Drive the tasks handling the send and receive ends of the websockets to completion, terminating when either of them completes. - Termina

(
    state: Arc<ActorState>,
    hotswap: impl Fn() -> HotswapWatcher,
    idle_timer: impl Future<Output = ()>,
    mut send_task: JoinHandle<()>,
    mut recv_task: JoinHandle<()>,
    unordered_tx

///
/// [close handshake]: https://datatracker.ietf.org/doc/html/rfc6455#section-7
async fn ws_main_loop<HotswapWatcher>(
    state: Arc<ActorState>,
    hotswap: impl Fn() -> HotswapWatcher,
    idle_timer: impl Future<Output = ()>,
    mut send_task: JoinHandle<()>,
    mut recv_task: JoinHandle<()>,
    unordered_tx: impl Fn(UnorderedWsMessage),
) where
    HotswapWatcher: Future<Output = Result<(), NoSuchModule>>,
{
    // Ensure we terminate both tasks if either exits.
    let abort_send = send_task.abort_handle();
    let abort_recv = recv_task.abort_handle();
    defer! {
        abort_send.abort();
        abort_recv.abort();
    };
    // Set up the ping interval.
    let mut ping_interval = tokio::time::interval(state.config.ping_interval);
    // Arm the first hotswap watcher.
    let watch_hotswap = hotswap();

    pin_mut!(watch_hotswap);
    pin_mut!(idle_timer);

    loop {
        let closed = state.closed();

        tokio::select! {
            // Drive send and receive tasks to completion,
            // propagating panics.
            //
            // If either task completes,
            // the connection is considered closed and we break the loop.
            //
            // NOTE: We don't abort the tasks until this function returns,
            // so the `Err` can't contain an `is_cancelled()` value.
            //
            // Even if the tasks were cancelled (e.g. if the caller retains
            // [`tokio::task::AbortHandle`]s), the reasonable thing to do is to
            // exit the loop as if the tasks completed normally.
            res = &mut send_task => {
                if let Err(e) = res
                    && e.is_panic() {
                        panic::resume_unwind(e.into_panic())
                    }
                break;
            },
            res = &mut recv_task => {
                if let Err(e) = res
                    && e.is_panic() {
                        panic::resume_unwind(e.into_panic())
                    }
                break;
            },

            // Exit if we haven't heard from the client for too long.
            _ = &mut idle_timer => {