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

AxDataWebSocketClient::stream() takes the Option<Arc<receiver>> out of self.out_rx; that field is None when connect() has not run yet or when the receiver was already consumed by a previous stream() call. Because the receiver can only be handed out once, a second call panics with this expect message instead of returning an error.

Source

Thrown at crates/adapters/architect_ax/src/websocket/data/client.rs:1027

    fn restore_unsubscribe_state(&self, topic: &str, was_pending: bool) {
        self.subscriptions.confirm_unsubscribe(topic);
        self.subscriptions.mark_subscribe(topic);
        if !was_pending {
            self.subscriptions.confirm_subscribe(topic);
        }
    }

    /// 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 = AxDataWsMessage> + '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

Call stream() exactly once per client, immediately after connect(), and keep the returned stream alive for the client's lifetime.
On reconnect, create a fresh client instance (connect + stream) instead of reusing the old one.
Before calling, guard with a check such as if client.can_stream() { ... } if such a predicate exists, or track the call with your own boolean.
Split work across tasks by cloning the client only for sending, never to obtain a second stream.

Example fix

// before
client.connect().await?;
let s1 = client.stream();
// ... later
let s2 = client.stream(); // panics: already taken
// after
client.connect().await?;
let stream = client.stream(); // exactly once, consume until it ends
// on reconnect: build a brand-new client and stream from it

Defensive patterns

Strategy: validation

Validate before calling

// Track usage yourself since the client panics instead of returning Err
struct StreamOnce { taken: bool }
fn take_stream(client: &mut AxDataWebSocketClient, state: &mut StreamOnce)
    -> Result<impl futures_util::Stream<Item = AxDataWsMessage>, String> {
    if state.taken { return Err("stream() already called on this client".into()); }
    state.taken = true;
    Ok(client.stream())
}

Prevention

Establish the invariant: connect() then exactly one stream() per client instance.
Never call stream() inside retry or reconnection loops; rebuild the client instead.
Wrap the stream lifetime in one owning task to make double-consumption impossible.

When it happens

Trigger: Calling stream() twice on the same AxDataWebSocketClient; calling stream() before connect(); creating a client without calling connect() and immediately calling stream().

Common situations: Re-running a consume loop after a reconnect attempt without rebuilding the client; test harnesses calling stream() in setup and again in the test body; wrapping stream() in retry logic that re-invokes it after a stream ends.

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/6b330d347b84c70a. Report an issue: GitHub.