{"record":{"id":"97d201e781195be5","repo":"nautechsystems/nautilus_trader","slug":"message-stream-receiver-already-taken-or-not-conne","errorCode":null,"errorMessage":"Message stream receiver already taken or not connected","messagePattern":"Message stream receiver already taken or not connected","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"crates/adapters/dydx/src/websocket/client.rs","lineNumber":578,"sourceCode":"    ) -> Option<tokio::sync::mpsc::UnboundedReceiver<DydxWsOutputMessage>> {\n        self.out_rx.lock().take()\n    }\n\n    /// Returns a stream of venue-specific WebSocket messages.\n    ///\n    /// Takes ownership of the message receiver and returns it as a `Stream`.\n    ///\n    /// # Panics\n    ///\n    /// Panics if the message receiver has already been taken or the client is not connected.\n    pub fn stream(\n        &mut self,\n    ) -> impl futures_util::Stream<Item = DydxWsOutputMessage> + Send + 'static {\n        let mut rx = self\n            .out_rx\n            .lock()\n            .take()\n            .expect(\"Message stream receiver already taken or not connected\");\n\n        async_stream::stream! {\n            while let Some(msg) = rx.recv().await {\n                yield msg;\n            }\n        }\n    }\n\n    /// Connects the websocket client and opens the primary pool slot.\n    ///\n    /// Additional slots are spawned lazily by `subscribe_*` methods once the\n    /// per-channel limit is reached on every existing slot.\n    ///\n    /// # Errors\n    ///\n    /// Returns an error if the connection cannot be established.\n    pub async fn connect(&mut self) -> DydxWsResult<()> {\n        let connect_lock = Arc::clone(&self.connect_lock);","sourceCodeStart":560,"sourceCodeEnd":596,"githubUrl":"https://github.com/nautechsystems/nautilus_trader/blob/18893faf8b356be3320add8de2f861b0b647cf06/crates/adapters/dydx/src/websocket/client.rs#L560-L596","documentation":"The dYdX websocket client hands message receivers out as a one-shot resource: `out_rx` is stored as an `Option` behind a mutex, and `stream()` takes it, returning an async stream of `DydxWsOutputMessage`. Calling `stream()` a second time (or before the client is connected) finds the receiver already taken and panics. Only one consumer of the message stream is supported per client instance.","triggerScenarios":"Calling `client.stream()` twice on the same DydxWsClient; calling `stream()` on a client created without a connection (no `out_rx` stored); cloning/holding a client and having two tasks each call stream().","commonSituations":"Reconnecting logic that creates a new stream on the same client instead of building a fresh client; application code subscribing to the stream in two modules; tests that call stream() in setup and again per test case.","solutions":["Call `stream()` exactly once per client; store the returned stream in a single consumer task and fan out messages internally (e.g. via broadcast channel).","On reconnect, construct a new client/connection rather than re-calling stream() on the old one.","Check `is_connected`/connection state before calling stream() to avoid the not-connected case.","If multiple consumers are needed, wrap the single stream in a `tokio::sync::broadcast` sender."],"exampleFix":"// before\nlet s1 = client.stream().await;\nlet s2 = client.stream().await; // panics: already taken\n// after\nlet stream = client.stream().await;\nlet (tx, _) = tokio::sync::broadcast::channel(1024);\ntokio::spawn(async move {\n    use futures_util::StreamExt;\n    let mut stream = stream;\n    while let Some(msg) = stream.next().await {\n        let _ = tx.send(msg);\n    }\n});","handlingStrategy":"type-guard","validationCode":"if self.out_rx.lock().is_none() {\n    // stream already taken or client not connected — build a new client instead\n}","typeGuard":"fn can_stream(client: &DydxWsClient) -> bool {\n    client.out_rx.lock().is_some()\n}","tryCatchPattern":null,"preventionTips":["Call stream() exactly once per client instance and fan out via broadcast","Recreate the client on reconnect instead of re-calling stream()","Designate a single consumer task for the websocket message stream"],"tags":["rust","panic","websocket","stream","single-consumer"],"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"}