nautechsystems/nautilus_trader · critical

Stream receiver already taken or client not connected

Error message

Stream receiver already taken or client not connected

What it means

BybitWebSocketClient::stream() consumes the single mpsc receiver (out_rx) via Option::take and returns the message stream. The expect fires when the receiver was never set (connect() not called, so out_rx is None) or when stream() was already called once, since the Option is emptied after the first call. It is a deliberate one-shot API: each client can hand out its message stream exactly once.

Source

Thrown at crates/adapters/bybit/src/websocket/client.rs:950

        let cmd = HandlerCommand::Unsubscribe { topics: payloads };
        if let Err(e) = self.cmd_tx.read().await.send(cmd) {
            log::debug!("Failed to send unsubscribe command: error={e}");
        }

        Ok(())
    }

    /// Returns a stream of venue-typed [`BybitWsMessage`] items.
    ///
    /// # Panics
    ///
    /// Panics if called before [`Self::connect`] or if the stream has already been taken.
    pub fn stream(&mut self) -> impl futures_util::Stream<Item = BybitWsMessage> + use<> {
        let rx = self
            .out_rx
            .take()
            .expect("Stream receiver already taken or client 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;
            }
        }
    }

    /// Returns the number of currently registered subscriptions.
    #[must_use]
    pub fn subscription_count(&self) -> usize {
        self.subscriptions.len()
    }

    /// Returns the credential associated with this client, if any.
    #[must_use]
    pub fn credential(&self) -> Option<&Credential> {
        self.credential.as_ref()

View on GitHub (pinned to 18893faf8b)

Solutions

1. Call connect() before stream() so out_rx is populated.
2. Take the stream exactly once and share the yielded messages via a broadcaster (e.g. tokio::sync::broadcast) or an mpsc fan-out if multiple consumers need them.
3. If a reconnect is needed, create a new client (new connection yields a new receiver) instead of reusing the old instance.
4. Return the already-created stream from the first call and reuse it rather than calling stream() again.

Example fix

// before
let s1 = client.stream();
let s2 = client.stream(); // panics: already taken
// after
let mut s = client.stream(); // take once
while let Some(msg) = s.next().await { /* fan out via broadcast channel if needed */ }

fn can_stream(client: &mut BybitWsClient, taken: bool) -> bool { !taken } // track: stream() is one-shot
// Prefer: struct OnceStream { taken: AtomicBool } and check taken.swap(true, Ordering::SeqCst) before calling

fn stream_available(client: &BybitWsClient) -> bool { client.has_receiver() } // if exposed; otherwise track a bool around your own stream() call site

// Rust panics are not catchable idiomatically; guard instead:
if stream_already_taken { return Err(anyhow!("stream already taken")); }
let stream = client.stream();

When it happens

Trigger: Calling client.stream() before connect(), or calling stream() a second time on the same client instance (the out_rx Option is None after the first take).

Common situations: Re-running stream() after a reconnect attempt on the same client; calling stream() in two places (e.g. one task for logging, one for event handling); constructing the client via a path that skips connect().

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

• Cannot take ownership - other references exist
• InstrumentState channel requires kind and currency parameter
• funding_interval_hour out of bounds
• {field} must be non-negative, was {value}
• millisecond timestamp overflowed

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