Add a fold function to tar_eio_gz

eio/tar_eio_gz.ml

@@ -2,32 +2,59 @@ open Eio
 
 module Monad = struct
   type 'a t = 'a
-  let (>>=) a f = f a
+
+  let ( >>= ) a f = f a
   let return = Fun.id
 end
 
 module Reader = struct
   type in_channel = Flow.source_ty Resource.t
   type 'a t = 'a
+
   let read = Flow.single_read
   let really_read f b = Flow.read_exact f b
-  let skip f (n: int) =
+
+  let skip f (n : int) =
     let buffer_size = 32768 in
     let buffer = Cstruct.create buffer_size in
-    let rec loop (n: int) =
+    let rec loop (n : int) =
       if n <= 0 then ()
       else
         let amount = min n buffer_size in
         let block = Cstruct.sub buffer 0 amount in
         really_read f block;
-        loop (n - amount) in
+        loop (n - amount)
+    in
     loop n
 end
 
 module Writer = struct
   type out_channel = Flow.sink_ty Resource.t
   type 'a t = 'a
+
   let really_write f b = Flow.write f [ b ]
 end
 
-include Tar_gz.Make(Monad)(Writer)(Reader)
+include Tar_gz.Make (Monad) (Writer) (Reader)
+
+type source = in_channel
+type sink = out_channel
+
+let of_sink ?bits ?q ~level ~mtime os f =
+  of_out_channel ?bits ?q ~level ~mtime os (f :> Flow.sink_ty Resource.t)
+
+let of_source f =
+  of_in_channel ~internal:(Cstruct.create 65536)
+    (f :> Flow.source_ty Eio.Resource.t)
+
+let fold ?level f source init =
+  let rec aux global acc =
+    match get_next_header ?level ~global source with
+    | hdr, global ->
+        let acc = f hdr acc in
+        let to_skip = Tar.Header.(Int64.to_int (to_sectors hdr) * length) in
+        skip source to_skip;
+        aux global acc
+    | exception Tar.Header.End_of_stream -> acc
+  in
+  aux None init

eio/tar_eio_gz.mli

@@ -1,35 +1,55 @@
-open Eio
-
 (** Read tar.gz files with eio *)
 
-type in_channel
+open Eio
 
-val of_in_channel : internal:Cstruct.t ->  Flow.source_ty Resource.t -> in_channel
+type source
 
-(** Returns the next header block or fails with {!Tar.Header.End_of_stream}
-    if two consecutive zero-filled blocks are discovered. Assumes stream is
+val of_source : 'a Flow.source -> source
+
+val get_next_header :
+  ?level:Tar.Header.compatibility ->
+  global:Tar.Header.Extended.t option ->
+  source ->
+  Tar.Header.t * Tar.Header.Extended.t option
+(** Returns the next header block or fails with {!Tar.Header.End_of_stream} if
+    two consecutive zero-filled blocks are discovered. Assumes stream is
     positioned at the possible start of a header block.
 
     @raise Stdlib.End_of_file if the stream unexpectedly fails. *)
-val get_next_header : ?level:Tar.Header.compatibility -> global:Tar.Header.Extended.t option
-  -> in_channel -> (Tar.Header.t * Tar.Header.Extended.t option)
 
-val really_read : in_channel -> Cstruct.t -> unit
+val really_read : source -> Cstruct.t -> unit
 (** [really_read fd buf] fills [buf] with data from [fd] or raises
     {!Stdlib.End_of_file}. *)
 
-val skip : in_channel -> int -> unit
+val fold :
+  ?level:Tar.Header.compatibility ->
+  (Tar.Header.t -> 'a -> 'a) ->
+  source ->
+  'a ->
+  'a
+
+val skip : source -> int -> unit
 
-type out_channel
+type sink
 
-val of_out_channel : ?bits:int -> ?q:int -> level:int ->
-  mtime:int32 -> Gz.os -> Flow.sink_ty Resource.t -> out_channel
+val of_sink :
+  ?bits:int ->
+  ?q:int ->
+  level:int ->
+  mtime:int32 ->
+  Gz.os ->
+  'a Flow.sink ->
+  sink
 
-val write_block : ?level:Tar.Header.compatibility -> ?global:Tar.Header.Extended.t ->
-  Tar.Header.t -> out_channel -> (unit -> string option) -> unit
+val write_block :
+  ?level:Tar.Header.compatibility ->
+  ?global:Tar.Header.Extended.t ->
+  Tar.Header.t ->
+  sink ->
+  (unit -> string option) ->
+  unit
 (** [write_block hdr oc stream] writes [hdr], then {i deflate} the given
-    [stream], then zero-pads so the stream is positionned for the next
-    block.
+    [stream], then zero-pads so the stream is positionned for the next block.
 
     A simple usage to write a file:
     {[
@@ -47,5 +67,5 @@ val write_block : ?level:Tar.Header.compatibility -> ?global:Tar.Header.Extended
         Unix.close fd
     ]} *)
 
-val write_end : out_channel -> unit
+val write_end : sink -> unit
 (** [write_end oc] writes a stream terminator to [oc]. *)