nautechsystems/nautilus_trader · error · anyhow::Error

Stream receiver already taken

Error message

Stream receiver already taken

What it means

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.

Source

Thrown at crates/common/src/live/listener.rs:94

            SerializationEncoding::default(),
        );

        if let Err(e) = self.tx.send(msg) {
            log::error!("Failed to send message: {e}");
        }
    }

    /// Gets the stream receiver for this instance.
    ///
    /// # Errors
    ///
    /// Returns an error if the stream receiver has already been taken.
    pub fn get_stream_receiver(
        &mut self,
    ) -> anyhow::Result<tokio::sync::mpsc::UnboundedReceiver<BusMessage>> {
        self.rx
            .take()
            .ok_or_else(|| anyhow::anyhow!("Stream receiver already taken"))
    }

    /// Streams messages arriving on the receiver channel.
    pub fn stream(
        stream_rx: tokio::sync::mpsc::UnboundedReceiver<BusMessage>,
    ) -> impl Stream<Item = BusMessage> + 'static {
        futures::stream::unfold(stream_rx, |mut rx| async {
            rx.recv().await.map(|msg| (msg, rx))
        })
        .fuse()
    }
}

#[cfg(test)]
mod tests {
    use bytes::Bytes;
    use futures::StreamExt;
    use ustr::Ustr;

View on GitHub (pinned to 18893faf8b)

Solutions

  1. Take the receiver exactly once and store/own the returned UnboundedReceiver for the lifetime of the consumer.
  2. If the receiver was taken, drop and recreate the MessageBusListener instead of re-calling get_stream_receiver.
  3. Restructure so a single component consumes the stream and fans messages out internally (broadcast or forwarding).
  4. Guard call sites with an Option<Receiver> you control and only call once when it is None.

Example fix

// before
let rx1 = listener.get_stream_receiver()?;
let rx2 = listener.get_stream_receiver()?; // panics/errs here
// after
let rx = listener.get_stream_receiver()?; // take once
spawn(MessageBusListener::stream(rx)); // single consumer fans out
Defensive patterns

Strategy: try-catch

Validate before calling

// Rust: only call when the receiver is still available
if listener.has_stream_receiver() {
    let rx = listener.get_stream_receiver()?;
}

Type guard

fn try_get(listener: &mut MessageBusListener) -> Option<Receiver<BusMessage>> {
    listener.get_stream_receiver().ok()
}

Try / catch

match listener.get_stream_receiver() {
    Ok(rx) => spawn(MessageBusListener::stream(rx)),
    Err(e) if e.to_string() == "Stream receiver already taken" => debug!("stream already attached"),
    Err(e) => return Err(e),
}

Prevention

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

AI-assisted analysis of nautechsystems/nautilus_trader@18893faf8b (2026-09-08). Data as JSON: /api/errors/e76dc9afe0eec4d5. Report an issue: GitHub.