Function spawn_keep_alive_stream 

pub(crate) fn spawn_keep_alive_stream<S>(
    session_id: SessionId,
    sender: S,
    routing: DestinationRouting,
    notification_mode: SurbNotificationMode,
    cfg: Arc<BalancerStateValues>,
    controller: &RateController,
    admit: impl Fn() -> bool + Send + 'static,
) -> AbortHandle
where
    S: Sink<(DestinationRouting, ApplicationDataOut)> + Clone + Send + Sync + Unpin + 'static,
    S::Error: Error + Send + Sync + 'static,

Spawns a task for a rate-limited stream of Keep-Alive messages to the Session counterparty.

controller is supplied rather than created here because the Exit shares this one stream between two independent producers — its SURB-level notification and the PIX filler — and both the rate and the admission decision below have to be steered from the same handle. The stream is suspended until the controller is given a non-zero rate.

admit is consulted once per released packet and drops it when it answers false. It is a predicate rather than a sink wrapper because the sink here must stay Clone — try_for_each_concurrent clones it per message — and the natural wrapper for a sink that may emit nothing, SinkExt::with_flat_map, is not. Filtering after the rate limiter also puts the decision where the Exit wants it: the rate says how often a packet may go, and this says whether this particular one should.