{"record":{"id":"16da6cb191c76206","repo":"nautechsystems/nautilus_trader","slug":"stream-receiver-already-taken-or-not-connected","errorCode":null,"errorMessage":"Stream receiver already taken or not connected","messagePattern":"Stream receiver already taken or not connected","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"crates/adapters/bitmex/src/websocket/client.rs","lineNumber":643,"sourceCode":"                \"WebSocket connection timeout after {timeout_secs} seconds\"\n            ))\n        })?;\n\n        Ok(())\n    }\n\n    /// Provides the internal stream as a channel-based stream.\n    ///\n    /// # Panics\n    ///\n    /// This function panics:\n    /// - If the websocket is not connected.\n    /// - If `stream` has already been called somewhere else (stream receiver is then taken).\n    pub fn stream(&mut self) -> impl Stream<Item = BitmexWsMessage> + use<> {\n        let rx = self\n            .out_rx\n            .take()\n            .expect(\"Stream receiver already taken or not connected\");\n        let mut rx = Arc::try_unwrap(rx).expect(\"Cannot take ownership - other references exist\");\n        async_stream::stream! {\n            while let Some(msg) = rx.recv().await {\n                yield msg;\n            }\n        }\n    }\n\n    /// Closes the client.\n    ///\n    /// # Errors\n    ///\n    /// Returns an error if the WebSocket is not connected or if closing fails.\n    pub async fn close(&mut self) -> Result<(), BitmexWsError> {\n        log::debug!(\"Starting close process\");\n\n        self.signal.store(true, Ordering::Relaxed);\n","sourceCodeStart":625,"sourceCodeEnd":661,"githubUrl":"https://github.com/nautechsystems/nautilus_trader/blob/18893faf8b356be3320add8de2f861b0b647cf06/crates/adapters/bitmex/src/websocket/client.rs#L625-L661","documentation":"BitmexWebSocketClient::stream() hands out the sole message receiver via Option::take on self.out_rx. The field is None when the websocket is not connected or when a previous stream() call already consumed the receiver, in which case the expect panics. The Arc::try_unwrap right after enforces that no cloned client still shares the receiver.","triggerScenarios":"Calling stream() before connect(); calling stream() twice on the same BitmexWebSocketClient; the first stream's consumer having already taken ownership while another caller retries stream().","commonSituations":"Resubscribing after a stream error by calling stream() again on the same client; both a market-data actor and a logging task attempting to consume Bitmex messages; supervisor loops that recreate the stream after a drop without reconnecting.","solutions":["Call stream() exactly once, after connect(), and treat the returned stream as the sole consumer for that connection.","On any reconnect, build a new BitmexWebSocketClient (connect then stream) rather than re-streaming the old instance.","Route messages to multiple consumers yourself via an mpsc/broadcast channel fed from the single stream.","Before calling, ensure no clone of the client survives, otherwise the subsequent Arc::try_unwrap expect will panic too."],"exampleFix":"// before\nclient.connect().await?;\nlet s1 = client.stream();\n// after stream end, on the same client:\nlet s2 = client.stream(); // panics: receiver already taken\n// after\nclient.connect().await?;\nlet s1 = client.stream();\n// for a new stream: create a fresh client, connect, then stream once","handlingStrategy":"validation","validationCode":"// Enforce connect-then-single-stream usage with your own state\nenum WsState { Disconnected, Connected(bool) } // bool = stream taken\nfn take_stream(client: &mut BitmexWebSocketClient, st: &mut WsState)\n    -> Result<impl Stream<Item = BitmexWsMessage>, String> {\n    match st { WsState::Connected(false) => { *st = WsState::Connected(true); Ok(client.stream()) },\n               _ => Err(\"connect() first and stream() only once\".to_string()) }\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Call connect() before stream(), and stream() exactly once per connection.","Handle stream termination by rebuilding the client, not by re-calling stream().","Avoid cloning the client; distribute messages downstream with channels."],"tags":["websocket","panic","single-consumer","bitmex","rust"],"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"}