Alternative Lwt adapter

.travis.yml

@@ -6,11 +6,14 @@ install: wget https://raw.githubusercontent.com/ocaml/ocaml-travisci-skeleton/ma
 script: bash -ex ./.travis-docker.sh
 env:
   global:
-  - PINS="httpaf-async:. httpaf:."
+  - PINS="httpaf-async:. httpaf-lwt:. httpaf:."
   matrix:
   - PACKAGE="httpaf"       DISTRO="ubuntu-16.04"    OCAML_VERSION="4.06.0"
   - PACKAGE="httpaf-async" DISTRO="ubuntu-16.04"    OCAML_VERSION="4.06.0"
+  - PACKAGE="httpaf-lwt"   DISTRO="ubuntu-16.04"    OCAML_VERSION="4.06.0"
   - PACKAGE="httpaf"       DISTRO="ubuntu-16.04"    OCAML_VERSION="4.04.2"
   - PACKAGE="httpaf-async" DISTRO="ubuntu-16.04"    OCAML_VERSION="4.04.2"
+  - PACKAGE="httpaf-lwt"   DISTRO="ubuntu-16.04"    OCAML_VERSION="4.04.2"
   - PACKAGE="httpaf"       DISTRO="debian-unstable" OCAML_VERSION="4.03.0"
   - PACKAGE="httpaf-async" DISTRO="debian-unstable" OCAML_VERSION="4.03.0"
+  - PACKAGE="httpaf-lwt"   DISTRO="debian-unstable" OCAML_VERSION="4.03.0"

benchmarks/jbuild

@@ -1,6 +1,11 @@
 (jbuild_version 1)
 
-(executables
+(executable
  ((libraries (httpaf httpaf-async async core))
   (modules (wrk_async_benchmark))
-  (names (wrk_async_benchmark))))
+  (name wrk_async_benchmark)))
+
+(executable
+ ((name wrk_lwt_benchmark)
+  (modules (Wrk_lwt_benchmark))
+  (libraries (httpaf httpaf-lwt lwt.unix))))

benchmarks/wrk_lwt_benchmark.ml

@@ -0,0 +1,59 @@
+(* TODO Cleanup *)
+(* TODO Organize like the echo server? *)
+
+module Body = Httpaf.Body
+module Headers = Httpaf.Headers
+module Reqd = Httpaf.Reqd
+module Response = Httpaf.Response
+module Status = Httpaf.Status
+
+let text = "CHAPTER I. Down the Rabbit-Hole  Alice was beginning to get very tired of sitting by her sister on the bank, and of having nothing to do: once or twice she had peeped into the book her sister was reading, but it had no pictures or conversations in it, <and what is the use of a book,> thought Alice <without pictures or conversations?> So she was considering in her own mind (as well as she could, for the hot day made her feel very sleepy and stupid), whether the pleasure of making a daisy-chain would be worth the trouble of getting up and picking the daisies, when suddenly a White Rabbit with pink eyes ran close by her. There was nothing so very remarkable in that; nor did Alice think it so very much out of the way to hear the Rabbit say to itself, <Oh dear! Oh dear! I shall be late!> (when she thought it over afterwards, it occurred to her that she ought to have wondered at this, but at the time it all seemed quite natural); but when the Rabbit actually took a watch out of its waistcoat-pocket, and looked at it, and then hurried on, Alice started to her feet, for it flashed across her mind that she had never before seen a rabbit with either a waistcoat-pocket, or a watch to take out of it, and burning with curiosity, she ran across the field after it, and fortunately was just in time to see it pop down a large rabbit-hole under the hedge. In another moment down went Alice after it, never once considering how in the world she was to get out again. The rabbit-hole went straight on like a tunnel for some way, and then dipped suddenly down, so suddenly that Alice had not a moment to think about stopping herself before she found herself falling down a very deep well. Either the well was very deep, or she fell very slowly, for she had plenty of time as she went down to look about her and to wonder what was going to happen next. First, she tried to look down and make out what she was coming to, but it was too dark to see anything; then she looked at the sides of the well, and noticed that they were filled with cupboards......"
+let text = Lwt_bytes.of_string text
+
+let headers =
+  Headers.of_list [
+    "Content-Length", string_of_int (Lwt_bytes.length text)
+  ]
+
+let error_handler _ ?request error start_response =
+  let response_body = start_response Headers.empty in
+  begin match error with
+  | `Exn exn ->
+    Body.write_string response_body (Printexc.to_string exn);
+    Body.write_string response_body "\n";
+  | #Status.standard as error ->
+    Body.write_string response_body (Status.default_reason_phrase error)
+  end;
+  Body.close_writer response_body
+
+let request_handler _ reqd =
+  let { Httpaf.Request.target } = Reqd.request reqd in
+  let request_body       = Reqd.request_body reqd in
+  Body.close_reader request_body;
+  match target with
+  | "/" -> Reqd.respond_with_bigstring reqd (Response.create ~headers `OK) text;
+  | _   -> Reqd.respond_with_string    reqd (Response.create `Not_found) "Route not found"
+
+let connection_handler =
+  Httpaf_lwt.Server.create_connection_handler ?config:None ~request_handler ~error_handler
+
+let () =
+  let open Lwt.Infix in
+
+  let port = ref 8080 in
+  Arg.parse
+    ["-p", Arg.Set_int port, " Listening port number (8080 by default)"]
+    ignore
+    "Echoes POST requests. Runs forever.";
+
+  let listen_address = Unix.(ADDR_INET (inet_addr_loopback, !port)) in
+
+  Lwt.async begin fun () ->
+    Lwt_io.establish_server_with_client_socket
+      ~backlog:11_000 listen_address connection_handler
+    >>= fun _server ->
+    Lwt.return_unit
+  end;
+
+  let forever, _ = Lwt.wait () in
+  Lwt_main.run forever

examples/lwt/jbuild

@@ -0,0 +1,5 @@
+(jbuild_version 1)
+
+(executables
+ ((names (lwt_get lwt_post lwt_echo_server))
+  (libraries (httpaf httpaf-lwt lwt lwt.unix))))

examples/lwt/lwt_echo_server.ml

@@ -0,0 +1,125 @@
+(* TODO This needs to be paired with the requester example. *)
+(* TODO Usage to comment. *)
+
+let connection_handler : Unix.sockaddr -> Lwt_unix.file_descr -> unit Lwt.t =
+  let module Body = Httpaf.Body in
+  let module Headers = Httpaf.Headers in
+  let module Reqd = Httpaf.Reqd in
+  let module Response = Httpaf.Response in
+  let module Status = Httpaf.Status in
+
+  let request_handler : Unix.sockaddr -> _ Reqd.t -> unit =
+      fun _client_address request_descriptor ->
+
+    let request = Reqd.request request_descriptor in
+    match request.meth with
+    | `POST ->
+      let request_body = Reqd.request_body request_descriptor in
+
+      let response_content_type =
+        match Headers.get request.headers "Content-Type" with
+        | Some request_content_type -> request_content_type
+        | None -> "application/octet-stream"
+      in
+
+      (* Due to a possible bug in http/af, read from the body only once, and
+         create the response based on the data in that first read.
+
+         The bug is (possibly) in the client. Client_connection seems to go into
+         a read loop despite Client_connection.shutdown being called, due to the
+         reader being in the Partial state, and the next operation function
+         unconditionally returning `Read in that case.
+
+         One workaround for this is to have the server send a Content-Length
+         header. To do that, this code has the server simply reply after the
+         first chunk is read, and use that chunk's length.
+
+         The code I would expect to work, without the possible bug, is commented
+         out below. *)
+
+      Body.schedule_read
+        request_body
+        ~on_eof:ignore
+        ~on_read:(fun request_data ~off ~len ->
+          let response =
+            Response.create
+              ~headers:(Headers.of_list [
+                "Content-Type", response_content_type;
+                "Content-Length", string_of_int len;
+                "Connection", "close";
+              ])
+              `OK
+          in
+
+          let response_body =
+            Reqd.respond_with_streaming request_descriptor response in
+
+          Body.write_bigstring response_body request_data ~off ~len;
+          Body.close_writer response_body)
+
+      (*
+      let rec respond () =
+        Body.schedule_read
+          request_body
+          ~on_eof:(fun () -> Body.close_writer response_body)
+          ~on_read:(fun request_data ~off ~len ->
+            Body.write_bigstring response_body request_data ~off ~len;
+            respond ())
+      in
+      respond ()
+      *)
+
+    | _ ->
+      Reqd.respond_with_string
+        request_descriptor (Response.create `Method_not_allowed) ""
+  in
+
+  let error_handler :
+      Unix.sockaddr ->
+      ?request:Httpaf.Request.t ->
+      _ ->
+      (Headers.t -> [`write] Body.t) ->
+        unit =
+      fun _client_address ?request:_ error start_response ->
+
+    let response_body = start_response Headers.empty in
+
+    begin match error with
+    | `Exn exn ->
+      Body.write_string response_body (Printexc.to_string exn);
+      Body.write_string response_body "\n";
+
+    | #Status.standard as error ->
+      Body.write_string response_body (Status.default_reason_phrase error)
+    end;
+
+    Body.close_writer response_body
+  in
+
+  Httpaf_lwt.Server.create_connection_handler
+    ?config:None
+    ~request_handler
+    ~error_handler
+
+
+
+let () =
+  let open Lwt.Infix in
+
+  let port = ref 8080 in
+  Arg.parse
+    ["-p", Arg.Set_int port, " Listening port number (8080 by default)"]
+    ignore
+    "Echoes POST requests. Runs forever.";
+
+  let listen_address = Unix.(ADDR_INET (inet_addr_loopback, !port)) in
+
+  Lwt.async begin fun () ->
+    Lwt_io.establish_server_with_client_socket
+      listen_address connection_handler
+    >>= fun _server ->
+    Lwt.return_unit
+  end;
+
+  let forever, _ = Lwt.wait () in
+  Lwt_main.run forever

examples/lwt/lwt_get.ml

@@ -0,0 +1,73 @@
+(* TODO Cleanup *)
+
+module Body = Httpaf.Body
+module Response = Httpaf.Response
+
+let response_handler : unit Lwt.u -> Response.t -> [ `read ] Body.t -> unit =
+    fun notify_request_finished response response_body ->
+
+  match response.status with
+  | `OK ->
+    let rec read_response () =
+      Body.schedule_read
+        response_body
+        ~on_eof:(fun () -> Lwt.wakeup_later notify_request_finished ())
+        ~on_read:(fun response_fragment ~off ~len ->
+          let response_fragment_string = Bytes.create len in
+          Lwt_bytes.blit_to_bytes
+            response_fragment off
+            response_fragment_string 0
+            len;
+          print_string (Bytes.unsafe_to_string response_fragment_string);
+
+          read_response ())
+    in
+    read_response ()
+
+  | _ ->
+    Format.fprintf Format.err_formatter "%a\n%!" Response.pp_hum response;
+    exit 1
+
+(* TODO A real error handler *)
+let error_handler _ =
+  assert false
+
+open Lwt.Infix
+
+let () =
+  let host = ref None in
+  let port = ref 80 in
+
+  Arg.parse
+    ["-p", Set_int port, " port number"]
+    (fun host_argument -> host := Some host_argument)
+    "lwt_get.exe [-p N] HOST";
+
+  let host =
+    match !host with
+    | None -> failwith "No hostname provided"
+    | Some host -> host
+  in
+
+  Lwt_main.run begin
+    Lwt_unix.getaddrinfo host (string_of_int !port) [Unix.(AI_FAMILY PF_INET)]
+    >>= fun addresses ->
+
+    let socket = Lwt_unix.socket Unix.PF_INET Unix.SOCK_STREAM 0 in
+    Lwt_unix.connect socket (List.hd addresses).Unix.ai_addr
+    >>= fun () ->
+
+    let headers = Httpaf.Headers.of_list ["Host", host] in
+    let request = Httpaf.Request.create ~headers `GET "/" in
+    let request_finished, notify_request_finished = Lwt.wait () in
+    let request_body =
+      Httpaf_lwt.Client.request
+        socket
+        request
+        ~error_handler
+        ~response_handler:(response_handler notify_request_finished)
+    in
+    Body.close_writer request_body;
+
+    request_finished
+  end

examples/lwt/lwt_post.ml

@@ -0,0 +1,81 @@
+module Body = Httpaf.Body
+module Response = Httpaf.Response
+
+let response_handler : unit Lwt.u -> Response.t -> [ `read ] Body.t -> unit =
+    fun notify_request_finished response response_body ->
+
+  match response.status with
+  | `OK ->
+    let rec read_response () =
+      Body.schedule_read
+        response_body
+        ~on_eof:(fun () -> Lwt.wakeup_later notify_request_finished ())
+        ~on_read:(fun response_fragment ~off ~len ->
+          let response_fragment_string = Bytes.create len in
+          Lwt_bytes.blit_to_bytes
+            response_fragment off
+            response_fragment_string 0
+            len;
+          print_string (Bytes.unsafe_to_string response_fragment_string);
+
+          read_response ())
+    in
+    read_response ()
+
+  | _ ->
+    Format.fprintf Format.err_formatter "%a\n%!" Response.pp_hum response;
+    exit 1
+
+(* TODO Real error handler *)
+let error_handler _ =
+  assert false
+
+open Lwt.Infix
+
+let () =
+  let host = ref None in
+  let port = ref 8080 in
+
+  Arg.parse
+    ["-p", Set_int port, " port number"]
+    (fun host_argument -> host := Some host_argument)
+    "lwt_get.exe [-p N] HOST";
+
+  let host =
+    match !host with
+    | None -> failwith "No hostname provided"
+    | Some host -> host
+  in
+
+  Lwt_main.run begin
+    Lwt_io.(read stdin)
+    >>= fun request_content ->
+
+    Lwt_unix.getaddrinfo host (string_of_int !port) [Unix.(AI_FAMILY PF_INET)]
+    >>= fun addresses ->
+
+    let socket = Lwt_unix.socket Unix.PF_INET Unix.SOCK_STREAM 0 in
+    Lwt_unix.connect socket (List.hd addresses).Unix.ai_addr
+    >>= fun () ->
+
+    let headers =
+      Httpaf.Headers.of_list [
+        "Host", host;
+        "Connection", "close";
+        "Content-Length", string_of_int (String.length request_content);
+      ]
+    in
+    let request = Httpaf.Request.create ~headers `POST "/" in
+    let request_finished, notify_request_finished = Lwt.wait () in
+    let request_body =
+      Httpaf_lwt.Client.request
+        socket
+        request
+        ~error_handler
+        ~response_handler:(response_handler notify_request_finished)
+    in
+    Body.write_string request_body request_content;
+    Body.close_writer request_body;
+
+    request_finished
+  end