Stream receiver already taken or client not connected - stre

Error message

Stream receiver already taken or client not connected - stream() can only be called once

What it means

AxOrdersWebSocketClient::stream() removes the single out_rx receiver via Option::take; if the client was never connected (out_rx still None) or stream() was already called, take() yields None and the expect panics. The receiver is intentionally single-use so exactly one consumer owns the orders event stream.

Source

Thrown at crates/adapters/architect_ax/src/websocket/orders/client.rs:701

    pub async fn get_open_orders(&self) -> AxOrdersWsResult<i64> {
        let request_id = self.next_request_id();

        self.send_cmd(HandlerCommand::GetOpenOrders { request_id })
            .await?;

        Ok(request_id)
    }

    /// Returns a stream of WebSocket messages.
    ///
    /// # Panics
    ///
    /// Panics if called before `connect()` or if the stream has already been taken.
    pub fn stream(&mut self) -> impl futures_util::Stream<Item = AxOrdersWsMessage> + 'static {
        let rx = self
            .out_rx
            .take()
            .expect("Stream receiver already taken or client not connected - stream() can only be called once");
        let mut rx = Arc::try_unwrap(rx).expect(
            "Cannot take ownership of stream - client was cloned and other references exist",
        );
        async_stream::stream! {
            while let Some(msg) = rx.recv().await {
                yield msg;
            }
        }
    }

    pub(crate) fn begin_shutdown(&self) {
        self.cancellation_token.load().cancel();
        self.signal.store(true, Ordering::Release);
    }

    /// Disconnects the WebSocket connection gracefully.
    pub async fn disconnect(&self) {
        log::debug!("Disconnecting WebSocket");

View on GitHub (pinned to 18893faf8b)

Solutions

Ensure connect() completes before stream() and call stream() once, owning the returned stream for the connection's lifetime.
Rebuild the client (new + connect + stream) for each reconnect cycle rather than reusing a drained client.
Centralize stream consumption in one task and distribute messages internally via mpsc channels.
Add an assertion/log before calling stream() to confirm the client is connected and stream was not yet taken.

Example fix

// before
let client = AxOrdersWebSocketClient::new(...);
let stream = client.stream(); // panics: never connected
// after
let mut client = AxOrdersWebSocketClient::new(...);
client.connect().await?;
let stream = client.stream(); // once, after connect

Defensive patterns

Strategy: validation

Validate before calling

// Guard before calling stream(): ensure connected and not yet streamed (track with your own flag)
if !connected || already_streamed {
    return Err("orders client must be connected and stream() called at most once".into());
}
let stream = client.stream();

Prevention

Always connect() before stream(); call stream() once and keep the stream until shutdown.
Recreate the whole client for each reconnect cycle.
Centralize order-stream consumption in a single task.

When it happens

Trigger: Calling stream() before connect(); calling stream() a second time on the same AxOrdersWebSocketClient after the first stream was handed out.

Common situations: Restarting a subscription loop after a dropped stream without reconnecting; order-entry code and a monitor task both trying to consume the orders stream; test code calling stream() in multiple helper functions.

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/8576687a485edba5. Report an issue: GitHub.