{"record":{"id":"8576687a485edba5","repo":"nautechsystems/nautilus_trader","slug":"stream-receiver-already-taken-or-client-not-connec-857668","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/orders/client.rs","lineNumber":701,"sourceCode":"    pub async fn get_open_orders(&self) -> AxOrdersWsResult<i64> {\n        let request_id = self.next_request_id();\n\n        self.send_cmd(HandlerCommand::GetOpenOrders { request_id })\n            .await?;\n\n        Ok(request_id)\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 = AxOrdersWsMessage> + '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":683,"sourceCodeEnd":719,"githubUrl":"https://github.com/nautechsystems/nautilus_trader/blob/18893faf8b356be3320add8de2f861b0b647cf06/crates/adapters/architect_ax/src/websocket/orders/client.rs#L683-L719","documentation":"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.","triggerScenarios":"Calling stream() before connect(); calling stream() a second time on the same AxOrdersWebSocketClient after the first stream was handed out.","commonSituations":"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.","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."],"exampleFix":"// before\nlet client = AxOrdersWebSocketClient::new(...);\nlet stream = client.stream(); // panics: never connected\n// after\nlet mut client = AxOrdersWebSocketClient::new(...);\nclient.connect().await?;\nlet stream = client.stream(); // once, after connect","handlingStrategy":"validation","validationCode":"// Guard before calling stream(): ensure connected and not yet streamed (track with your own flag)\nif !connected || already_streamed {\n    return Err(\"orders client must be connected and stream() called at most once\".into());\n}\nlet stream = client.stream();","typeGuard":null,"tryCatchPattern":null,"preventionTips":["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."],"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"}