Real-time

async fn run(self) {
    let backoff = Backoff::default();
    let mut attempt: u32 = 0;
    loop {
        if self.stopped() {
            return;
        }
        let Ok(ctx) = self.ctx.upgrade() else { return };
        match WsConnection::connect(&ctx, &self.ws_url, WsOptions::default()).await {
            Ok(conn) => {
                drop(ctx);
                *lock(&self.shared.conn) = Some(conn.clone());
                self.link.set(Link::Open);
                let mut inbound = conn.messages();
                let mut first = true;
                while let Some(item) = next(&mut inbound).await {
                    if self.stopped() {
                        return;
                    }
                    match item {
                        Ok(WsMessage::Text(text)) => {
                            if first {
                                // A server that accepts and drops at once must keep backing off:
                                // the count resets when a message arrives, not when the socket opens.
                                first = false;
                                attempt = 0;
                            }
                            self.push(text);
                        }
                        Ok(WsMessage::Binary(_)) => {}
                        // The server said goodbye on purpose: do not hammer it.
                        Err(WsError::Closed { code: 1000, .. }) => {
                            return self.give_up("the server closed the connection");
                        }
                        Err(_) => break,
                    }
                }
                lock(&self.shared.conn).take();
            }
            // The credentials are the problem, not the network: another attempt fails the same way.
            Err(WsError::Refused {
                status: Some(401 | 403),
                ..
            }) => return self.give_up("the server refused the credentials"),
            Err(_) => drop(ctx),
        }
        if self.stopped() {
            return;
        }
        attempt += 1;
        self.link.set(Link::Reconnecting { attempt });

        if attempt % FALLBACK_EVERY == 0 && !self.sse_url.is_empty() {
            self.fall_back().await;
            if self.stopped() {
                return;
            }
            self.link.set(Link::Reconnecting { attempt });
        }

        let Ok(ctx) = self.ctx.upgrade() else { return };
        let delay = backoff.delay(attempt, &*ctx.rng());
        drop(ctx);
        if self.ctx.sleep(delay).await.is_err() {
            return;
        }
    }
}