{"record":{"id":"e76dc9afe0eec4d5","repo":"nautechsystems/nautilus_trader","slug":"stream-receiver-already-taken","errorCode":null,"errorMessage":"Stream receiver already taken","messagePattern":"Stream receiver already taken","errorType":"exception","errorClass":"anyhow::Error","httpStatus":null,"severity":"error","filePath":"crates/common/src/live/listener.rs","lineNumber":94,"sourceCode":"            SerializationEncoding::default(),\n        );\n\n        if let Err(e) = self.tx.send(msg) {\n            log::error!(\"Failed to send message: {e}\");\n        }\n    }\n\n    /// Gets the stream receiver for this instance.\n    ///\n    /// # Errors\n    ///\n    /// Returns an error if the stream receiver has already been taken.\n    pub fn get_stream_receiver(\n        &mut self,\n    ) -> anyhow::Result<tokio::sync::mpsc::UnboundedReceiver<BusMessage>> {\n        self.rx\n            .take()\n            .ok_or_else(|| anyhow::anyhow!(\"Stream receiver already taken\"))\n    }\n\n    /// Streams messages arriving on the receiver channel.\n    pub fn stream(\n        stream_rx: tokio::sync::mpsc::UnboundedReceiver<BusMessage>,\n    ) -> impl Stream<Item = BusMessage> + 'static {\n        futures::stream::unfold(stream_rx, |mut rx| async {\n            rx.recv().await.map(|msg| (msg, rx))\n        })\n        .fuse()\n    }\n}\n\n#[cfg(test)]\nmod tests {\n    use bytes::Bytes;\n    use futures::StreamExt;\n    use ustr::Ustr;","sourceCodeStart":76,"sourceCodeEnd":112,"githubUrl":"https://github.com/nautechsystems/nautilus_trader/blob/18893faf8b356be3320add8de2f861b0b647cf06/crates/common/src/live/listener.rs#L76-L112","documentation":"MessageBusListener stores its mpsc receiver as `Option<rx>`; `get_stream_receiver` takes it exactly once (`.take()`), moving ownership into the async stream loop. Calling it a second time — after the receiver has already been handed out — returns this error, since a single consumer channel cannot be cloned.","triggerScenarios":"Calling get_stream_receiver() twice on the same listener instance; calling it after stream()/py_stream() already consumed the receiver; a Python host plus Rust code both grabbing the receiver; re-initializing a stream after publish_after_close teardown.","commonSituations":"Double registration of the message-bus stream endpoint (e.g. both a Python callback consumer and a Rust task); hot-reload or restart logic re-attaching a listener without recreating it; tests asserting single-consumer semantics.","solutions":["Take the receiver exactly once and store/own the returned UnboundedReceiver for the lifetime of the consumer.","If the receiver was taken, drop and recreate the MessageBusListener instead of re-calling get_stream_receiver.","Restructure so a single component consumes the stream and fans messages out internally (broadcast or forwarding).","Guard call sites with an Option<Receiver> you control and only call once when it is None."],"exampleFix":"// before\nlet rx1 = listener.get_stream_receiver()?;\nlet rx2 = listener.get_stream_receiver()?; // panics/errs here\n// after\nlet rx = listener.get_stream_receiver()?; // take once\nspawn(MessageBusListener::stream(rx)); // single consumer fans out","handlingStrategy":"try-catch","validationCode":"// Rust: only call when the receiver is still available\nif listener.has_stream_receiver() {\n    let rx = listener.get_stream_receiver()?;\n}","typeGuard":"fn try_get(listener: &mut MessageBusListener) -> Option<Receiver<BusMessage>> {\n    listener.get_stream_receiver().ok()\n}","tryCatchPattern":"match listener.get_stream_receiver() {\n    Ok(rx) => spawn(MessageBusListener::stream(rx)),\n    Err(e) if e.to_string() == \"Stream receiver already taken\" => debug!(\"stream already attached\"),\n    Err(e) => return Err(e),\n}","preventionTips":["Design a single owner for the stream receiver; fan out via broadcast internally.","Never call get_stream_receiver from both Python and Rust hosts on the same listener.","Recreate the listener object instead of re-attaching after teardown.","Wrap the take-once call in OnceCell so repeat callers get the same consumer handle."],"tags":["rust","channel","single-consumer","state-error"],"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"}