{"record":{"id":"fa850f77d4912b25","repo":"nautechsystems/nautilus_trader","slug":"stream-receiver-already-taken-or-client-not-connec-fa850f","errorCode":null,"errorMessage":"Stream receiver already taken or client not connected","messagePattern":"Stream receiver already taken or client not connected","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"critical","filePath":"crates/adapters/bybit/src/websocket/client.rs","lineNumber":950,"sourceCode":"\n        let cmd = HandlerCommand::Unsubscribe { topics: payloads };\n        if let Err(e) = self.cmd_tx.read().await.send(cmd) {\n            log::debug!(\"Failed to send unsubscribe command: error={e}\");\n        }\n\n        Ok(())\n    }\n\n    /// Returns a stream of venue-typed [`BybitWsMessage`] items.\n    ///\n    /// # Panics\n    ///\n    /// Panics if called before [`Self::connect`] or if the stream has already been taken.\n    pub fn stream(&mut self) -> impl futures_util::Stream<Item = BybitWsMessage> + use<> {\n        let rx = self\n            .out_rx\n            .take()\n            .expect(\"Stream receiver already taken or client 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    /// Returns the number of currently registered subscriptions.\n    #[must_use]\n    pub fn subscription_count(&self) -> usize {\n        self.subscriptions.len()\n    }\n\n    /// Returns the credential associated with this client, if any.\n    #[must_use]\n    pub fn credential(&self) -> Option<&Credential> {\n        self.credential.as_ref()","sourceCodeStart":932,"sourceCodeEnd":968,"githubUrl":"https://github.com/nautechsystems/nautilus_trader/blob/18893faf8b356be3320add8de2f861b0b647cf06/crates/adapters/bybit/src/websocket/client.rs#L932-L968","documentation":"BybitWebSocketClient::stream() consumes the single mpsc receiver (out_rx) via Option::take and returns the message stream. The expect fires when the receiver was never set (connect() not called, so out_rx is None) or when stream() was already called once, since the Option is emptied after the first call. It is a deliberate one-shot API: each client can hand out its message stream exactly once.","triggerScenarios":"Calling client.stream() before connect(), or calling stream() a second time on the same client instance (the out_rx Option is None after the first take).","commonSituations":"Re-running stream() after a reconnect attempt on the same client; calling stream() in two places (e.g. one task for logging, one for event handling); constructing the client via a path that skips connect().","solutions":["Call connect() before stream() so out_rx is populated.","Take the stream exactly once and share the yielded messages via a broadcaster (e.g. tokio::sync::broadcast) or an mpsc fan-out if multiple consumers need them.","If a reconnect is needed, create a new client (new connection yields a new receiver) instead of reusing the old instance.","Return the already-created stream from the first call and reuse it rather than calling stream() again."],"exampleFix":"// before\nlet s1 = client.stream();\nlet s2 = client.stream(); // panics: already taken\n// after\nlet mut s = client.stream(); // take once\nwhile let Some(msg) = s.next().await { /* fan out via broadcast channel if needed */ }","handlingStrategy":"type-guard","validationCode":"fn can_stream(client: &mut BybitWsClient, taken: bool) -> bool { !taken } // track: stream() is one-shot\n// Prefer: struct OnceStream { taken: AtomicBool } and check taken.swap(true, Ordering::SeqCst) before calling","typeGuard":"fn stream_available(client: &BybitWsClient) -> bool { client.has_receiver() } // if exposed; otherwise track a bool around your own stream() call site","tryCatchPattern":"// Rust panics are not catchable idiomatically; guard instead:\nif stream_already_taken { return Err(anyhow!(\"stream already taken\")); }\nlet stream = client.stream();","preventionTips":["Call connect() before stream() and call stream() exactly once per client instance.","Centralize stream consumption in one dedicated task; distribute messages with a broadcast channel.","Create a fresh client per (re)connection instead of reusing the old receiver."],"tags":["rust","websocket","panic","bybit","state"],"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-14T00:17:10.932Z"}