let sub (subs: 'a t) : 'a Pipe.Reader.t =
    let (r, w) = Pipe.create () in
    subs := w::!subs;
    r