Add Pgx_async.execute_pipe

This is an Async-only helper that returns a Pipe.Reader.t

pgx_async/src/pgx_async.ml

@@ -129,6 +129,11 @@ let with_conn ?host ?port ?user ?password ?database ?unix_domain_socket_dir
   Monitor.protect (fun () -> f dbh)
     ~finally:(fun () -> close dbh)
 
+let execute_pipe ?params db query =
+  Pipe.create_reader ~close_on_exception:true @@ fun writer ->
+  execute_iter ?params db query ~f:(fun row ->
+    Pipe.write_if_open writer row)
+
 module Value = struct
   include Pgx.Value
 

pgx_async/src/pgx_async.mli

@@ -19,6 +19,11 @@ val with_conn
   -> (t -> 'a Deferred.t)
   -> 'a Deferred.t
 
+(** Like [execute] but returns a pipe so you can operate on the results before they have all returned.
+    Note that [execute_iter] and [execute_fold] can perform significantly better because they don't have
+    as much overhead. *)
+val execute_pipe : ?params:Pgx.row -> t -> string -> Pgx.row Pipe.Reader.t
+
 module Value : sig
   include Pgx_value_intf.S