{"record":{"id":"6b330d347b84c70a","repo":"nautechsystems/nautilus_trader","slug":"stream-receiver-already-taken-or-client-not-connec","errorCode":null,"errorMessage":"Stream receiver already taken or client not connected - stream() can only be called once","messagePattern":"Stream receiver already taken or client not connected - stream\\(\\) can only be called once","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"crates/adapters/architect_ax/src/websocket/data/client.rs","lineNumber":1027,"sourceCode":"\n    fn restore_unsubscribe_state(&self, topic: &str, was_pending: bool) {\n        self.subscriptions.confirm_unsubscribe(topic);\n        self.subscriptions.mark_subscribe(topic);\n        if !was_pending {\n            self.subscriptions.confirm_subscribe(topic);\n        }\n    }\n\n    /// Returns a stream of WebSocket messages.\n    ///\n    /// # Panics\n    ///\n    /// Panics if called before `connect()` or if the stream has already been taken.\n    pub fn stream(&mut self) -> impl futures_util::Stream<Item = AxDataWsMessage> + 'static {\n        let rx = self\n            .out_rx\n            .take()\n            .expect(\"Stream receiver already taken or client not connected - stream() can only be called once\");\n        let mut rx = Arc::try_unwrap(rx).expect(\n            \"Cannot take ownership of stream - client was cloned and other references exist\",\n        );\n        async_stream::stream! {\n            while let Some(msg) = rx.recv().await {\n                yield msg;\n            }\n        }\n    }\n\n    pub(crate) fn begin_shutdown(&self) {\n        self.cancellation_token.load().cancel();\n        self.signal.store(true, Ordering::Release);\n    }\n\n    /// Disconnects the WebSocket connection gracefully.\n    pub async fn disconnect(&self) {\n        log::debug!(\"Disconnecting WebSocket\");","sourceCodeStart":1009,"sourceCodeEnd":1045,"githubUrl":"https://github.com/nautechsystems/nautilus_trader/blob/18893faf8b356be3320add8de2f861b0b647cf06/crates/adapters/architect_ax/src/websocket/data/client.rs#L1009-L1045","documentation":"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.","triggerScenarios":"Calling stream() twice on the same AxDataWebSocketClient; calling stream() before connect(); creating a client without calling connect() and immediately calling stream().","commonSituations":"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.","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."],"exampleFix":"// before\nclient.connect().await?;\nlet s1 = client.stream();\n// ... later\nlet s2 = client.stream(); // panics: already taken\n// after\nclient.connect().await?;\nlet stream = client.stream(); // exactly once, consume until it ends\n// on reconnect: build a brand-new client and stream from it","handlingStrategy":"validation","validationCode":"// Track usage yourself since the client panics instead of returning Err\nstruct StreamOnce { taken: bool }\nfn take_stream(client: &mut AxDataWebSocketClient, state: &mut StreamOnce)\n    -> Result<impl futures_util::Stream<Item = AxDataWsMessage>, String> {\n    if state.taken { return Err(\"stream() already called on this client\".into()); }\n    state.taken = true;\n    Ok(client.stream())\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["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."],"tags":["websocket","panic","single-consumer","rust","architect-ax"],"backgroundTag":"invalid-state-transition","analyzedSha":"18893faf8b356be3320add8de2f861b0b647cf06","analyzedAt":"2026-09-08T20:49:34.690Z","contentChangedAt":"2026-09-08T20:49:34.690Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}