Adapt to janestreet 0.17 series

Author: anuragsoni

CHANGES.md

@@ -1,3 +1,7 @@
+# 0.12.0
+
+* Adapt to Janestreet's 0.17 series.
+
 # 0.11.0
 
 * Forward a server context object to all http services. This can be used to lookup peer-socket address and ssl details (if using ssl) for the underlying connection.

dune-project

@@ -5,7 +5,7 @@
 (generate_opam_files true)
 
 (source
- (sourcehut soni/shuttle_http))
+ (github anuragsoni/shuttle_http))
 
 (maintainers "Anurag Soni <anurag@sonianurag.com>")
 
@@ -22,19 +22,21 @@
   "Shuttle_http is a low level library for implementing HTTP/1.1 web services and clients in OCaml.")
  (depends
   (async
-   (>= v0.16.0))
+   (>= v0.17.0))
+  (async_log
+   (>= v0.17.0))
   (async_ssl
-   (>= v0.16.0))
+   (>= v0.17.0))
   (core
-   (>= v0.16.0))
+   (>= v0.17.0))
   (jane_rope
-   (>= v0.16.0))
+   (>= v0.17.0))
   (ocaml
-   (>= 4.14.0))
+   (>= 5.1.0))
   (ppx_jane
-   (>= v0.16.0))
+   (>= v0.17.0))
   (re2
-   (>= v0.16.0))
+   (>= v0.17.0))
   (core_unix :with-test)))
 
 (package

http/src/client.ml

@@ -232,15 +232,15 @@ module Connection = struct
                in
                match verification_result with
                | Error err ->
-                 Ivar.fill ivar (Error err);
+                 Ivar.fill_exn ivar (Error err);
                  Deferred.unit
                | Ok () ->
                  let conn = { reader; writer; address } in
                  let t = Sequencer.create conn in
                  Throttle.at_kill t (fun conn ->
                    let%bind () = Output_channel.close conn.writer in
                    Input_channel.close conn.reader);
-                 Ivar.fill ivar (Ok t);
+                 Ivar.fill_exn ivar (Ok t);
                  closed t));
         Ivar.read ivar)
   ;;
@@ -277,7 +277,7 @@ module Connection = struct
                 let response = Response0.with_body response body in
                 if not (Response.keep_alive response && Request.keep_alive request)
                 then close t;
-                Ivar.fill ivar response;
+                Ivar.fill_exn ivar response;
                 (match Response.body response with
                  | Body.Fixed _ | Empty -> return (`Finished ())
                  | Stream stream ->

http/src/dune

@@ -2,4 +2,4 @@
  (public_name shuttle_http)
  (preprocess
   (pps ppx_jane))
- (libraries core async re2 async_ssl jane_rope))
+ (libraries core async re2 async_ssl jane_rope async_log))

http/src/input_channel0.ml

@@ -45,7 +45,7 @@ let close t =
   if not t.is_closed
   then (
     t.is_closed <- true;
-    Fd.close t.fd >>> fun () -> Ivar.fill t.closed ());
+    Fd.close t.fd >>> fun () -> Ivar.fill_exn t.closed ());
   closed t
 ;;
 

http/src/output_channel.mli

@@ -1,7 +1,7 @@
 open! Core
 open! Async_kernel
 open Async_unix
-module Logger : Log.Global_intf
+module Logger : Async_log.Global.S
 
 type t [@@deriving sexp_of]
 

http/src/output_channel0.ml

@@ -3,7 +3,7 @@ open! Async_kernel
 open Async_unix
 module Unix = Core_unix
 open! Async_kernel_require_explicit_time_source
-module Logger = Log.Make_global ()
+module Logger = Async_log.Global.Make ()
 
 module Flush_result = struct
   type t =
@@ -86,7 +86,7 @@ let create ?max_buffer_size ?buf_len ?write_timeout ?time_source fd =
 
 let wakeup_flushes_with_error t error =
   while not (Queue.is_empty t.flushes) do
-    Ivar.fill (Queue.dequeue_exn t.flushes).ivar error
+    Ivar.fill_exn (Queue.dequeue_exn t.flushes).ivar error
   done
 ;;
 
@@ -125,14 +125,14 @@ let close' t =
   | Closed | Start_close -> ()
   | Open ->
     t.close_state <- Start_close;
-    Ivar.fill t.close_started ();
+    Ivar.fill_exn t.close_started ();
     Deferred.any_unit
       [ Time_source.after t.time_source (Time_ns.Span.of_sec 5.)
       ; Deferred.ignore_m (flushed_or_fail t)
       ]
     >>> fun () ->
     t.close_state <- Closed;
-    Fd.close t.fd >>> fun () -> Ivar.fill t.close_finished ()
+    Fd.close t.fd >>> fun () -> Ivar.fill_exn t.close_finished ()
 ;;
 
 let close t =
@@ -155,7 +155,7 @@ let dequeue_flushes t =
     (not (Queue.is_empty t.flushes))
     && Int63.( <= ) (Queue.peek_exn t.flushes).pos t.bytes_written
   do
-    Ivar.fill (Queue.dequeue_exn t.flushes).ivar Flush_result.Flushed
+    Ivar.fill_exn (Queue.dequeue_exn t.flushes).ivar Flush_result.Flushed
   done
 ;;
 
@@ -187,7 +187,7 @@ let rec write_everything t =
   else (
     match write_nonblocking t with
     | `Eof ->
-      Ivar.fill t.remote_closed ();
+      Ivar.fill_exn t.remote_closed ();
       stop_writer t Flush_result.Remote_closed
     | `Poll_again -> wait_and_write_everything t
     | `Ok n ->
@@ -293,7 +293,7 @@ let write_from_pipe t reader =
     then (
       (* use [read_now'] as [iter] doesn't allow working on chunks of values at a time. *)
       match Pipe.read_now' ~consumer reader with
-      | `Eof -> Ivar.fill finished ()
+      | `Eof -> Ivar.fill_exn finished ()
       | `Nothing_available -> Pipe.values_available reader >>> fun _ -> loop ()
       | `Ok bufs ->
         Queue.iter bufs ~f:(fun buf -> write t buf);

http/src/server.ml

@@ -1,6 +1,6 @@
 open! Core
 open! Async
-module Logger = Log.Make_global ()
+module Logger = Async_log.Global.Make ()
 module Ssl_conn = Ssl
 
 module Ssl = struct
@@ -109,7 +109,7 @@ type 'a t =
 type 'addr service = 'addr t -> Request.t -> Response.t Deferred.t [@@deriving sexp_of]
 
 let closed t = Ivar.read t.closed
-let close t = if Ivar.is_empty t.closed then Ivar.fill t.closed ()
+let close t = if Ivar.is_empty t.closed then Ivar.fill_exn t.closed ()
 let is_ssl t = Option.is_some t.ssl
 
 let ssl_peer_certificate t =
@@ -179,14 +179,14 @@ let run_server_loop t handler =
     | Error Partial ->
       Input_channel.refill t.reader
       >>> (function
-      | `Eof -> Ivar.fill t.closed ()
-      | `Ok -> parse_request t)
+       | `Eof -> Ivar.fill_exn t.closed ()
+       | `Ok -> parse_request t)
     | Error (Fail error) ->
       t.error_handler ~exn:(Error.to_exn error) `Bad_request
       >>> fun response ->
       (write_response t response;
        Io_util.write_body (Response.body response) t.writer)
-      >>> fun () -> Ivar.fill t.closed ()
+      >>> fun () -> Ivar.fill_exn t.closed ()
     | Ok (req, consumed) ->
       Input_channel.consume t.reader consumed;
       create_request_body_reader t req
@@ -198,7 +198,7 @@ let run_server_loop t handler =
       Input_channel.refill_with_timeout t.reader span
       >>> fun v ->
       (match v with
-       | `Eof -> Ivar.fill t.closed ()
+       | `Eof -> Ivar.fill_exn t.closed ()
        | `Ok ->
          let now' = Time_ns.now () in
          let diff = Time_ns.abs_diff now now' in
@@ -208,7 +208,7 @@ let run_server_loop t handler =
       >>> fun response ->
       (write_response t response;
        Io_util.write_body (Response.body response) t.writer)
-      >>> fun () -> Ivar.fill t.closed ()
+      >>> fun () -> Ivar.fill_exn t.closed ()
     | Ok (req, consumed) ->
       Input_channel.consume t.reader consumed;
       create_request_body_reader t req
@@ -219,7 +219,7 @@ let run_server_loop t handler =
       >>> fun response ->
       (write_response t response;
        Io_util.write_body (Response.body response) t.writer)
-      >>> fun () -> Ivar.fill t.closed ()
+      >>> fun () -> Ivar.fill_exn t.closed ()
     | Ok req_body ->
       let req = Request.with_body req req_body in
       let promise = handler t req in
@@ -248,7 +248,7 @@ let run_server_loop t handler =
        | Ok () -> ()
        | Error exn ->
          Logger.error "Error while running upgrade handler: %s" (Exn.to_string exn));
-      Ivar.fill t.closed ()
+      Ivar.fill_exn t.closed ()
     | Response _ ->
       if is_keep_alive
       then (
@@ -265,7 +265,7 @@ let run_server_loop t handler =
           if Time_ns.Span.is_positive t.read_header_timeout
           then parse_request_with_timeout t t.read_header_timeout
           else parse_request t)
-      else Ivar.fill t.closed ()
+      else Ivar.fill_exn t.closed ()
   in
   Monitor.detach t.monitor;
   Scheduler.within ~priority:Priority.normal ~monitor:t.monitor (fun () ->
@@ -281,7 +281,7 @@ let run_server_loop t handler =
     then
       (write_response t response;
        Io_util.write_body (Response.body response) t.writer)
-      >>> fun () -> Ivar.fill t.closed ());
+      >>> fun () -> Ivar.fill_exn t.closed ());
   Ivar.read t.closed
 ;;
 

http/src/server.mli

@@ -1,6 +1,6 @@
 open! Core
 open! Async
-module Logger : Log.Global_intf
+module Logger : Async_log.Global.S
 
 (** [error_handler] can be used to customize how the server deals with any unhandled
     exceptions. A default implementation is provided that will respond with a status code

http/src/tcp_channel.ml

@@ -37,24 +37,24 @@ let listen
     ~on_handler_error
     where_to_listen
     (fun addr socket ->
-      let fd = Socket.fd socket in
-      let input_channel =
-        Input_channel.create ?max_buffer_size ?buf_len ?time_source fd
-      in
-      let output_channel =
-        Output_channel.create ?max_buffer_size ?buf_len ?write_timeout ?time_source fd
-      in
-      let%bind res =
-        Deferred.any
-          [ collect_errors output_channel (fun () ->
-              handler addr input_channel output_channel)
-          ; Output_channel.remote_closed output_channel |> Deferred.ok
-          ]
-      in
-      let%bind () = close_channels input_channel output_channel in
-      match res with
-      | Ok () -> Deferred.unit
-      | Error exn -> raise exn)
+       let fd = Socket.fd socket in
+       let input_channel =
+         Input_channel.create ?max_buffer_size ?buf_len ?time_source fd
+       in
+       let output_channel =
+         Output_channel.create ?max_buffer_size ?buf_len ?write_timeout ?time_source fd
+       in
+       let%bind res =
+         Deferred.any
+           [ collect_errors output_channel (fun () ->
+               handler addr input_channel output_channel)
+           ; Output_channel.remote_closed output_channel |> Deferred.ok
+           ]
+       in
+       let%bind () = close_channels input_channel output_channel in
+       match res with
+       | Ok () -> Deferred.unit
+       | Error exn -> raise exn)
 ;;
 
 let listen_inet
@@ -79,24 +79,24 @@ let listen_inet
     ~on_handler_error
     where_to_listen
     (fun addr socket ->
-      let fd = Socket.fd socket in
-      let input_channel =
-        Input_channel.create ?max_buffer_size ?buf_len ?time_source fd
-      in
-      let output_channel =
-        Output_channel.create ?max_buffer_size ?buf_len ?write_timeout ?time_source fd
-      in
-      let%bind res =
-        Deferred.any
-          [ collect_errors output_channel (fun () ->
-              handler addr input_channel output_channel)
-          ; Output_channel.remote_closed output_channel |> Deferred.ok
-          ]
-      in
-      let%bind () = close_channels input_channel output_channel in
-      match res with
-      | Ok () -> Deferred.unit
-      | Error exn -> raise exn)
+       let fd = Socket.fd socket in
+       let input_channel =
+         Input_channel.create ?max_buffer_size ?buf_len ?time_source fd
+       in
+       let output_channel =
+         Output_channel.create ?max_buffer_size ?buf_len ?write_timeout ?time_source fd
+       in
+       let%bind res =
+         Deferred.any
+           [ collect_errors output_channel (fun () ->
+               handler addr input_channel output_channel)
+           ; Output_channel.remote_closed output_channel |> Deferred.ok
+           ]
+       in
+       let%bind () = close_channels input_channel output_channel in
+       match res with
+       | Ok () -> Deferred.unit
+       | Error exn -> raise exn)
 ;;
 
 let with_connection