Function spawn_stream_pumps 

fn spawn_stream_pumps<S, C>(
    peer: PeerId,
    stream: S,
    rx: AsyncRx<Array<<C as Decoder>::Item>>,
    cache: Cache<PeerId, PeerSink<<C as Decoder>::Item>>,
    token: Arc<()>,
    codec: C,
    ingress_from_peers: Sender<(PeerId, <C as Decoder>::Item)>,
    frame_writer_backpressure_bytes: usize,
)
where
    S: AsyncRead + AsyncWrite + Send + 'static,
    C: Encoder<<C as Decoder>::Item> + Decoder + Send + Sync + Clone + 'static,
    <C as Encoder<<C as Decoder>::Item>>::Error: Debug + Display + Send + Sync + 'static,
    <C as Decoder>::Error: Debug + Display + Send + Sync + 'static,
    <C as Decoder>::Item: AsRef<[u8]> + Clone + Send + 'static,

Spawn the write and read pump tasks for an open peer stream.

The write pump drains rx into the framed stream writer; the read pump forwards decoded frames to ingress_from_peers. Both tasks invalidate cache[peer] when they complete, but only if the cache entry still holds the same token — this prevents a stale task from wiping a newer sink.