Utils: add Lwt_pipe.peek_all

src/lib_stdlib_lwt/lwt_pipe.ml

@@ -136,6 +136,12 @@ let rec peek ({ closed ; queue ; _ } as q) =
     wait_push q >>= fun () ->
     peek q
 
+let peek_all { queue ; closed } =
+  if closed then
+    []
+  else
+    List.rev (Queue.fold (fun acc (_, e) -> e :: acc) [] queue)
+
 exception Empty
 
 let pop_now_exn ({ closed ; queue ; empty ; current_size ; _ } as q) =

src/lib_stdlib_lwt/lwt_pipe.mli

@@ -42,6 +42,10 @@ val peek : 'a t -> 'a Lwt.t
 (** [peek] is like [pop] except it does not removes the first
     element. *)
 
+val peek_all : 'a t -> 'a list
+(** [peek_all q] returns the elements in the [q] (oldest first),
+    or [[]] if empty. *)
+
 val values_available : 'a t -> unit Lwt.t
 (** [values_available] is like [peek] but it ignores the value
     returned. *)