risingwave_stream::executor::exchange::input::remote_input

Function run_inner

Source 
fn run_inner(
    stream: Streaming<GetStreamResponse>,
    permits_tx: UnboundedSender<Value>,
    up_down_ids: UpDownActorIds,
    up_down_frag: UpDownFragmentIds,
    metrics: Arc<StreamingMetrics>,
    batched_permits_limit: usize,
) -> impl Stream<Item = Result<DispatcherMessage, StreamExecutorError>>