nautechsystems/nautilus_trader · error

Stream receiver already taken or not connected

Error message

Stream receiver already taken or not connected

What it means

BitmexWebSocketClient::stream() hands out the sole message receiver via Option::take on self.out_rx. The field is None when the websocket is not connected or when a previous stream() call already consumed the receiver, in which case the expect panics. The Arc::try_unwrap right after enforces that no cloned client still shares the receiver.

Source

Thrown at crates/adapters/bitmex/src/websocket/client.rs:643

                "WebSocket connection timeout after {timeout_secs} seconds"
            ))
        })?;

        Ok(())
    }

    /// Provides the internal stream as a channel-based stream.
    ///
    /// # Panics
    ///
    /// This function panics:
    /// - If the websocket is not connected.
    /// - If `stream` has already been called somewhere else (stream receiver is then taken).
    pub fn stream(&mut self) -> impl Stream<Item = BitmexWsMessage> + use<> {
        let rx = self
            .out_rx
            .take()
            .expect("Stream receiver already taken or not connected");
        let mut rx = Arc::try_unwrap(rx).expect("Cannot take ownership - other references exist");
        async_stream::stream! {
            while let Some(msg) = rx.recv().await {
                yield msg;
            }
        }
    }

    /// Closes the client.
    ///
    /// # Errors
    ///
    /// Returns an error if the WebSocket is not connected or if closing fails.
    pub async fn close(&mut self) -> Result<(), BitmexWsError> {
        log::debug!("Starting close process");

        self.signal.store(true, Ordering::Relaxed);

View on GitHub (pinned to 18893faf8b)

Solutions

  1. Call stream() exactly once, after connect(), and treat the returned stream as the sole consumer for that connection.
  2. On any reconnect, build a new BitmexWebSocketClient (connect then stream) rather than re-streaming the old instance.
  3. Route messages to multiple consumers yourself via an mpsc/broadcast channel fed from the single stream.
  4. Before calling, ensure no clone of the client survives, otherwise the subsequent Arc::try_unwrap expect will panic too.

Example fix

// before
client.connect().await?;
let s1 = client.stream();
// after stream end, on the same client:
let s2 = client.stream(); // panics: receiver already taken
// after
client.connect().await?;
let s1 = client.stream();
// for a new stream: create a fresh client, connect, then stream once
Defensive patterns

Strategy: validation

Validate before calling

// Enforce connect-then-single-stream usage with your own state
enum WsState { Disconnected, Connected(bool) } // bool = stream taken
fn take_stream(client: &mut BitmexWebSocketClient, st: &mut WsState)
    -> Result<impl Stream<Item = BitmexWsMessage>, String> {
    match st { WsState::Connected(false) => { *st = WsState::Connected(true); Ok(client.stream()) },
               _ => Err("connect() first and stream() only once".to_string()) }
}

Prevention

When it happens

Trigger: Calling stream() before connect(); calling stream() twice on the same BitmexWebSocketClient; the first stream's consumer having already taken ownership while another caller retries stream().

Common situations: Resubscribing after a stream error by calling stream() again on the same client; both a market-data actor and a logging task attempting to consume Bitmex messages; supervisor loops that recreate the stream after a drop without reconnecting.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

AI-assisted analysis of nautechsystems/nautilus_trader@18893faf8b (2026-09-08). Data as JSON: /api/errors/16da6cb191c76206. Report an issue: GitHub.