Cohttp expose closefn function

cohttp-async/src/server.ml

@@ -87,14 +87,14 @@ let handle_client handle_request sock rd wr =
             Body.drain body >>| fun () -> Ivar.fill_if_empty finished ()
         | `Response (req, body, (res, res_body)) ->
             (* There are scenarios if a client leaves before consuming the full response,
-            we might have a reference to an async Pipe that doesn't get drained.
+               we might have a reference to an async Pipe that doesn't get drained.
 
-               Not draining or closing a pipe can lead to issues if its holding a resource like
-               a file handle as those resources will never be closed, leading to a leak.
+                  Not draining or closing a pipe can lead to issues if its holding a resource like
+                  a file handle as those resources will never be closed, leading to a leak.
 
-               Async writers have a promise that's fulfilled whenever they are closed,
-               so we can use it to schedule a close operation on the stream to ensure that we
-               don't leave a stream open if the underlying channels are closed. *)
+                  Async writers have a promise that's fulfilled whenever they are closed,
+                  so we can use it to schedule a close operation on the stream to ensure that we
+                  don't leave a stream open if the underlying channels are closed. *)
             (match res_body with
             | `Empty | `String _ | `Strings _ -> ()
             | `Pipe stream ->

cohttp-async/test/test_async_integration.ml

@@ -111,7 +111,8 @@ let ts =
             ("Pipe with empty strings", Pipe.of_list [ ""; ""; "" ], true);
           ]
         in
-        Deferred.List.iter ~how:`Sequential tests ~f:(fun (msg, pipe, expected) ->
+        Deferred.List.iter ~how:`Sequential tests 
+           ~f:(fun (msg, pipe, expected) ->
             is_empty (`Pipe pipe) >>| fun real ->
             assert_equal ~msg expected real)
         >>= fun () ->

cohttp-lwt-jsoo/src/cohttp_lwt_jsoo.ml

@@ -180,6 +180,9 @@ struct
 
   let callv ?ctx:_ _uri _reqs = Lwt.fail Cohttp_lwt_xhr_callv_not_implemented
 
+  let call_with_closefn ?ctx:_ ?headers:_ ?body:_ ?chunked:_ _meth _uri =
+    assert false
+
   (* ??? *)
 end
 

cohttp-lwt-unix/bin/dune

@@ -1,14 +1,6 @@
 (executables
  (names cohttp_curl_lwt cohttp_proxy_lwt cohttp_server_lwt)
- (libraries
-  cohttp-lwt-unix
-  cohttp_server
-  logs
-  logs.lwt
-  logs.fmt
-  logs.cli
-  cmdliner
-  conduit-lwt
-  fmt.tty)
+ (libraries cohttp-lwt-unix cohttp_server logs logs.lwt logs.fmt logs.cli
+   cmdliner conduit-lwt fmt.tty)
  (package cohttp-lwt-unix)
  (public_names cohttp-curl-lwt cohttp-proxy-lwt cohttp-server-lwt))

cohttp-lwt/src/client.ml

@@ -46,8 +46,8 @@ module Make (IO : S.IO) (Net : S.Net with module IO = IO) = struct
     | `DELETE -> false
     | _ -> true
 
-  let call ?(ctx = Net.default_ctx) ?headers ?(body = `Empty) ?chunked meth uri
-      =
+  let call_with_closefn ?(ctx = Net.default_ctx) ?headers ?(body = `Empty)
+      ?chunked meth uri =
     let headers = match headers with None -> Header.init () | Some h -> h in
     Net.connect_uri ~ctx uri >>= fun (_conn, ic, oc) ->
     let closefn () = Net.close ic oc in
@@ -88,7 +88,11 @@ module Make (IO : S.IO) (Net : S.Net with module IO = IO) = struct
     |> fun t ->
     Lwt.on_cancel t closefn;
     Lwt.on_failure t (fun _exn -> closefn ());
-    t
+    Lwt.return (t, closefn)
+
+  let call ?(ctx = Net.default_ctx) ?headers ?(body = `Empty) ?chunked meth uri
+      =
+    call_with_closefn ~ctx ?headers ~body ?chunked meth uri >>= fun (t, _) -> t
 
   (* The HEAD should not have a response body *)
   let head ?ctx ?headers uri = call ?ctx ?headers `HEAD uri >|= fst

cohttp-lwt/src/s.ml

@@ -74,7 +74,21 @@ module type Client = sig
       (using [ocaml-tls]) or SSL (using [ocaml-ssl]), on [*:443] or on the
       specified port by the user. If neitehr [ocaml-tls] or [ocaml-ssl] are
       installed on the system, [cohttp]/[conduit] tries the usual ([*:80]) or
-      the specified port by the user in a non-secured way. *)
+      the specified port by the user in a non-secured way.
+
+      The function returns response and body. *)
+
+  val call_with_closefn :
+    ?ctx:ctx ->
+    ?headers:Cohttp.Header.t ->
+    ?body:Body.t ->
+    ?chunked:bool ->
+    Cohttp.Code.meth ->
+    Uri.t ->
+    ((Cohttp.Response.t * Body.t) Lwt.t * (unit -> unit)) Lwt.t
+  (** [call_with_closefn ?ctx ?headers ?body ?chunked meth uri] is the same as
+      [call] but returns response, body and [close_fn] which force releases the
+      connection. *)
 
   val head :
     ?ctx:ctx -> ?headers:Cohttp.Header.t -> Uri.t -> Cohttp.Response.t Lwt.t

cohttp-mirage/src/dune

@@ -5,4 +5,4 @@
  (preprocess
   (pps ppx_sexp_conv))
  (libraries conduit-mirage cohttp-lwt mirage-channel mirage-kv mirage-flow
-   magic-mime astring))
+   magic-mime astring cstruct))

examples/async/dune

@@ -1,6 +1,7 @@
 (executables
  (names hello_world receive_post s3_cp)
- (libraries mirage-crypto cohttp-async base async_kernel core_unix.command_unix))
+ (libraries mirage-crypto cohttp-async base async_kernel 
+   core_unix.command_unix cstruct))
 
 (alias
  (name runtest)

examples/async/s3_cp.ml

@@ -44,7 +44,6 @@ open Core
 open Async
 open Cohttp
 open Cohttp_async
-
 module Time = Time_float
 
 let ksrt (k, _) (k', _) = String.compare k k'