let identity ?name () =
  let q = Queue.create () in
  make ?name ~feed:(Queue.enqueue q) ()
    ~next:(fun stopped ->
      match Queue.dequeue q with
      | Some o -> `output o
      | None -> if stopped then `end_of_stream else `not_ready)