Struct ShuffleReceiver 

pub struct ShuffleReceiver { /* private fields */ }

Inbound side of the shuffle fabric: a Tonic ShuffleTransport server surfacing every received frame, attributed to its sending peer, on the bounded crossfire queue.

Implementations

impl ShuffleReceiver

pub async fn bind(local_id: ShufflePeerId, addr: SocketAddr) -> Result<Self>

Bind on addr and start serving. The bound address (with any ephemeral port resolved) is exposed via Self::local_addr.

Errors

Returns io::Error on bind failure.

pub async fn bind_with_kv( local_id: ShufflePeerId, addr: SocketAddr, kv: Arc<dyn ClusterKv>, ) -> Result<Self>

Bind and publish the listener’s address into kv under SHUFFLE_ADDR_KEY for peer discovery.

Errors

Returns io::Error on bind failure.

pub fn local_addr(&self) -> SocketAddr

Local socket address the server is bound to.

pub async fn recv(&self) -> Option<(ShufflePeerId, ShuffleMessage)>

Await the next (peer_id, msg). None once the server task has stopped and every queued item is drained. Single-owner; concurrent callers serialise via rx_returned. Cancellation-safe — a dropped recv() future returns the receiver to its slot via the RAII guard.

pub fn drain_available(&self) -> Vec<(ShufflePeerId, ShuffleMessage)>

Drain every currently-available (peer_id, msg) without blocking. Empty when the queue is empty or a recv() currently holds the receiver.

pub fn drain_vnode_data_for(&self, stage: &str) -> Vec<RecordBatch>

Non-blocking drain of the ShuffleMessage::VnodeData batches for stage; other stages stay bucketed for their own drainer. Empty if the queue is empty or a recv() holds it.

pub fn drain_staged_with_prefix( &self, prefix: &str, ) -> FxHashMap<String, Vec<RecordBatch>>

Single lock-cycle drain of every staged stage whose key starts with prefix, lifting those out and leaving operator stages untouched. Lets the subscription router pull all __sub:: batches in one pass.

pub fn stage_batch(&self, stage: String, batch: RecordBatch)

Stage batch under stage for a later Self::drain_vnode_data_for / Self::drain_all_staged — used when no operator for stage exists yet at drain time.

pub fn drain_staged_barriers(&self) -> Vec<(ShufflePeerId, CheckpointBarrier)>

Take the barriers stashed by Self::drain_vnode_data_for (peers that fanned out before this node began aligning).

pub fn drain_all_staged(&self) -> Vec<(String, RecordBatch)>

Empty the per-stage holdover, returning every buffered (stage, batch).

Trait Implementations

impl Debug for ShuffleReceiver

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Drop for ShuffleReceiver

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.