nautechsystems/nautilus_trader · error

Message stream receiver already taken or not connected

Error message

Message stream receiver already taken or not connected

What it means

The dYdX websocket client hands message receivers out as a one-shot resource: `out_rx` is stored as an `Option` behind a mutex, and `stream()` takes it, returning an async stream of `DydxWsOutputMessage`. Calling `stream()` a second time (or before the client is connected) finds the receiver already taken and panics. Only one consumer of the message stream is supported per client instance.

Source

Thrown at crates/adapters/dydx/src/websocket/client.rs:578

    ) -> Option<tokio::sync::mpsc::UnboundedReceiver<DydxWsOutputMessage>> {
        self.out_rx.lock().take()
    }

    /// Returns a stream of venue-specific WebSocket messages.
    ///
    /// Takes ownership of the message receiver and returns it as a `Stream`.
    ///
    /// # Panics
    ///
    /// Panics if the message receiver has already been taken or the client is not connected.
    pub fn stream(
        &mut self,
    ) -> impl futures_util::Stream<Item = DydxWsOutputMessage> + Send + 'static {
        let mut rx = self
            .out_rx
            .lock()
            .take()
            .expect("Message stream receiver already taken or not connected");

        async_stream::stream! {
            while let Some(msg) = rx.recv().await {
                yield msg;
            }
        }
    }

    /// Connects the websocket client and opens the primary pool slot.
    ///
    /// Additional slots are spawned lazily by `subscribe_*` methods once the
    /// per-channel limit is reached on every existing slot.
    ///
    /// # Errors
    ///
    /// Returns an error if the connection cannot be established.
    pub async fn connect(&mut self) -> DydxWsResult<()> {
        let connect_lock = Arc::clone(&self.connect_lock);

View on GitHub (pinned to 18893faf8b)

Solutions

  1. Call `stream()` exactly once per client; store the returned stream in a single consumer task and fan out messages internally (e.g. via broadcast channel).
  2. On reconnect, construct a new client/connection rather than re-calling stream() on the old one.
  3. Check `is_connected`/connection state before calling stream() to avoid the not-connected case.
  4. If multiple consumers are needed, wrap the single stream in a `tokio::sync::broadcast` sender.

Example fix

// before
let s1 = client.stream().await;
let s2 = client.stream().await; // panics: already taken
// after
let stream = client.stream().await;
let (tx, _) = tokio::sync::broadcast::channel(1024);
tokio::spawn(async move {
    use futures_util::StreamExt;
    let mut stream = stream;
    while let Some(msg) = stream.next().await {
        let _ = tx.send(msg);
    }
});
Defensive patterns

Strategy: type-guard

Validate before calling

if self.out_rx.lock().is_none() {
    // stream already taken or client not connected — build a new client instead
}

Type guard

fn can_stream(client: &DydxWsClient) -> bool {
    client.out_rx.lock().is_some()
}

Prevention

When it happens

Trigger: Calling `client.stream()` twice on the same DydxWsClient; calling `stream()` on a client created without a connection (no `out_rx` stored); cloning/holding a client and having two tasks each call stream().

Common situations: Reconnecting logic that creates a new stream on the same client instead of building a fresh client; application code subscribing to the stream in two modules; tests that call stream() in setup and again per test case.

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/97d201e781195be5. Report an issue: GitHub.