Add ?no_close to the new Lwt_io.establish_server

Related #258.

src/unix/lwt_io.ml

@@ -1419,7 +1419,7 @@ let shutdown_server_2 server = Lazy.force server.shutdown
 
 let shutdown_server server = Lwt.async (fun () -> shutdown_server_2 server)
 
-let _establish_server_base
+let establish_server_base
     bind ?fd ?(buffer_size = !default_buffer_size) ?(backlog=5) sockaddr f =
   let sock = match fd with
     | None -> Lwt_unix.socket (Unix.domain_of_sockaddr sockaddr) Unix.SOCK_STREAM 0
@@ -1469,10 +1469,11 @@ let establish_server ?fd ?buffer_size ?backlog sockaddr f =
   let blocking_bind fd addr =
     Lwt.return (Lwt_unix.Versioned.bind_1 fd addr) [@ocaml.warning "-3"]
   in
-  _establish_server_base blocking_bind ?fd ?buffer_size ?backlog sockaddr f
+  establish_server_base blocking_bind ?fd ?buffer_size ?backlog sockaddr f
   |> fst
 
-let establish_server_safe ?fd ?buffer_size ?backlog sockaddr f =
+let establish_server_safe
+    ?fd ?buffer_size ?backlog ?(no_close = false) sockaddr f =
   let best_effort_close channel =
     (* First, check whether the channel is closed. f may have already tried to
        close the channel, received an exception, and handled it somehow. If so,
@@ -1501,12 +1502,15 @@ let establish_server_safe ?fd ?buffer_size ?backlog sockaddr f =
           !Lwt.async_exception_hook exn;
           Lwt.return_unit)
 
-      >>= fun () -> best_effort_close input_channel
-      >>= fun () -> best_effort_close output_channel)
+      >>= fun () ->
+        if no_close then Lwt.return_unit
+        else
+          best_effort_close input_channel >>= fun () ->
+          best_effort_close output_channel)
   in
 
   let server, started =
-    _establish_server_base
+    establish_server_base
       Lwt_unix.Versioned.bind_2
       ?fd ?buffer_size ?backlog sockaddr handler in
   started >>= fun () ->

src/unix/lwt_io.mli

@@ -600,21 +600,27 @@ sig
     ?fd : Lwt_unix.file_descr ->
     ?buffer_size : int ->
     ?backlog : int ->
+    ?no_close : bool ->
     Unix.sockaddr -> (input_channel * output_channel -> unit Lwt.t) ->
       server Lwt.t
-  (** [establish_server_safe ?fd ?buffer_size ?backlog sockaddr f] creates a
-      server which listens for incoming connections. New connections are passed
-      to [f]. When threads returned by [f] complete, the connections are closed
-      automatically.
+  (** [establish_server_2 sockaddr f] creates a server which listens for
+      incoming connections. New connections are passed to [f]. When threads
+      returned by [f] complete, the connections are closed automatically. To
+      prevent automatic closing, apply [establish_server_2] with
+      [~no_close:true].
+
+      The [?fd] and [?backlog] arguments have the same meaning as in
+      {!Lwt_io.establish_server}. [?buffer_size] sets the internal buffer size
+      of the channels passed to [f].
 
       The server does not wait for each thread. It begins accepting new
       connections immediately.
 
       If a thread raises an exception, it is passed to
       [!Lwt.async_exception_hook]. Likewise, if the automatic [close] of a
       connection raises an exception, it is passed to
-      [!Lwt.async_exception_hook]. To handle exceptions raised by [close], call
-      it manually inside [f]. *)
+      [!Lwt.async_exception_hook]. To robustly handle these exceptions, you
+      should call [close] manually inside [f], and use your own handler. *)
 
   val shutdown_server_1 : server -> unit
   [@@ocaml.deprecated