What causes "Stream receiver already taken" in nautechsystems/nautilus_trader?

Call `get_stream_receiver` exactly once per subscriber, storing the returned Receiver for the stream loop

Stream receiver already taken

Error message

Stream receiver already taken

What it means

`get_stream_receiver` hands out the `mpsc::Receiver<BusMessage>` stored in the Redis message bus subscriber via `Option::take` — it can only be given out once. This error means the receiver was already taken (typically by `take_receiver`/`stream` startup), so a second consumer cannot be attached.

Source

Thrown at crates/infrastructure/src/redis/msgbus.rs:407

            });
        });

        log::debug!("Closed");
    }
}

impl RedisMessageBusBacking {
    /// Retrieves the Redis stream receiver for this message bus 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::Receiver<BusMessage>> {
        self.stream_rx
            .take()
            .ok_or_else(|| anyhow::anyhow!("Stream receiver already taken"))
    }

    /// Streams messages arriving on the stream receiver channel.
    pub fn stream(
        mut stream_rx: tokio::sync::mpsc::Receiver<BusMessage>,
    ) -> impl Stream<Item = BusMessage> + 'static {
        async_stream::stream! {
            while let Some(msg) = stream_rx.recv().await {
                yield msg;
            }
        }
    }

    pub async fn close_async(&mut self) {
        await_handle(self.pub_handle.take(), MSGBUS_PUBLISH).await;
        await_handle(self.stream_handle.take(), MSGBUS_STREAM).await;
        await_handle(self.heartbeat_handle.take(), MSGBUS_HEARTBEAT).await;
    }

View on GitHub (pinned to 18893faf8b)

Solutions

Call `get_stream_receiver` exactly once per subscriber, storing the returned Receiver for the stream loop
Before re-initializing, check whether the subscriber already started streaming and skip the second call
If a second consumer is needed, create a separate subscriber/subscription instead of reusing the same receiver
On reconnect, recreate the subscriber object rather than re-taking its receiver

Example fix

// before
let rx1 = subscriber.get_stream_receiver()?;
let rx2 = subscriber.get_stream_receiver()?; // Err: Stream receiver already taken
// after
let mut rx_opt = Some(subscriber.get_stream_receiver()?);
if let Some(rx) = rx_opt.take() {
    // consume rx exactly once
}

Defensive patterns

Strategy: type-guard

Validate before calling

// only attempt to take the receiver if streaming has not started
if !subscriber.is_streaming() {
    let rx = subscriber.get_stream_receiver()?;
    // start stream(rx) once
}

Type guard

// Rust
fn try_get_receiver(sub: &mut RedisBusSubscriber) -> Option<tokio::sync::mpsc::Receiver<BusMessage>> {
    sub.get_stream_receiver().ok()
}

Try / catch

// Rust
let rx = match subscriber.get_stream_receiver() {
    Ok(rx) => rx,
    Err(e) if e.to_string().contains("already taken") => {
        log::debug!("stream already started; reusing existing consumer");
        return Ok(()); // not an error for idempotent setup
    }
    Err(e) => return Err(e),
};

Prevention

Take the receiver exactly once, at a single well-known initialization point
Make stream initialization idempotent (skip if already streaming)
Never share one subscriber's receiver between multiple consumers
On reconnect, recreate the subscriber instead of re-taking its receiver

When it happens

Trigger: Calling `get_stream_receiver()` (or `take_receiver`) twice on the same RedisBusSubscriber/MessageBus instance, e.g. calling `stream()` again after already starting the stream, or two components both trying to consume the same Redis stream subscription.

Common situations: Accidentally initializing the message bus stream twice (e.g. on reconnect logic without checking prior state); sharing one subscriber between two consumers; re-running an async setup task after a partial failure where the receiver was already consumed.

Related errors

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