Merge branch 'master' into wip-nonblock

.github/workflows/main.yml

@@ -16,5 +16,6 @@ jobs:
     - run: opam pin -n .
     - run: opam depext -yt tiny_httpd tiny_httpd_camlzip
     - run: opam install -t . --deps-only
-    - run: opam exec -- dune build
+    - run: opam exec -- dune build @install
     - run: opam exec -- dune runtest
+      if: ${{ matrix.operating-system == 'unbuntu-latest' }}

echo.sh

@@ -1,2 +1,2 @@
 #!/bin/sh
-exec dune exec "src/examples/echo.exe" -- $@
+exec dune exec "examples/echo.exe" -- $@

examples/dune

@@ -9,6 +9,12 @@
   (modules sse_client)
   (libraries unix))
 
+(executable
+  (name echo)
+  (flags :standard -warn-error -a+8)
+  (modules echo)
+  (libraries tiny_httpd tiny_httpd_camlzip))
+
 (rule
   (targets test_output.txt)
   (deps (:script ./run_test.sh) ./sse_client.exe ./sse_server.exe)

examples/sse_server.ml

@@ -44,6 +44,14 @@ let () =
          Unix.sleepf 0.1;
        done;
     );
+  S.add_route_server_sent_handler server S.Route.(exact "count" @/ int @/ return)
+    (fun n _req (module EV : S.SERVER_SENT_GENERATOR)  ->
+       for i=0 to n do
+         EV.send_event ~data:(string_of_int i) ();
+         Unix.sleepf 0.1;
+       done;
+       EV.close();
+    );
 
   Printf.printf "listening on http://localhost:%d/\n%!" (S.port server);
   match S.run server with

src/Tiny_httpd.ml

@@ -536,9 +536,6 @@ module Request = struct
                 headers; body=()})
     with
     | End_of_file | Sys_error _ -> Ok None
-    | Byte_stream.Timeout ->
-       _debug (fun k -> k"Timeout");
-       Ok None
     | Bad_req (c,s) -> Error (c,s)
     | e ->
       Error (400, Printexc.to_string e)
@@ -704,14 +701,13 @@ end
 module Sem_ = struct
   type t = {
     mutable n : int;
-    max: int;
     mutex : Mutex.t;
     cond : Condition.t;
   }
 
   let create n =
     if n <= 0 then invalid_arg "Semaphore.create";
-    { n; max=n; mutex=Mutex.create(); cond=Condition.create(); }
+    { n; mutex=Mutex.create(); cond=Condition.create(); }
 
   let acquire m t =
     Mutex.lock t.mutex;
@@ -728,8 +724,6 @@ module Sem_ = struct
     t.n <- t.n + m;
     Condition.broadcast t.cond;
     Mutex.unlock t.mutex
-
-  let num_acquired self = self.max - self.n
 end
 
 module Route = struct
@@ -832,6 +826,7 @@ module type SERVER_SENT_GENERATOR = sig
     ?retry:string ->
     data:string ->
     unit -> unit
+  val close : unit -> unit
 end
 type server_sent_generator = (module SERVER_SENT_GENERATOR)
 
@@ -845,9 +840,6 @@ type t = {
   sem_max_connections: Sem_.t;
   (* semaphore to restrict the number of active concurrent connections *)
 
-  max_keep_alive: float;
-  (* maximum time in second before closing the client connections *)
-
   new_thread: (unit -> unit) -> unit;
   (* a function to run the given callback in a separate thread (or thread pool) *)
 
@@ -874,10 +866,6 @@ type t = {
 let addr self = self.addr
 let port self = self.port
 
-let active_connections self =
-  (* -1 because we decrease the semaphore before Unix.accept *)
-  Sem_.num_acquired self.sem_max_connections - 1
-
 let add_decode_request_cb self f =  self.cb_decode_req <- f :: self.cb_decode_req
 let add_encode_response_cb self f = self.cb_encode_resp <- f :: self.cb_encode_resp
 let set_top_handler self f = self.handler <- f
@@ -978,22 +966,23 @@ let add_route_server_sent_handler ?accept self route f =
           send_response_idempotent_()
         )
       let send_event = send_event
+      let close () = raise Exit
     end in
-    f req (module SSG : SERVER_SENT_GENERATOR);
+    try f req (module SSG : SERVER_SENT_GENERATOR);
+    with Exit -> close_out oc
   in
   add_route_handler_ self ?accept ~meth:`GET route ~tr_req f
 
 let create
     ?(masksigpipe=true)
     ?(max_connections=32)
-    ?(max_keep_alive=(-1.0))
     ?(new_thread=(fun f -> ignore (Thread.create f () : Thread.t)))
     ?(addr="127.0.0.1") ?(port=8080) ?sock () : t =
   let handler _req = Response.fail ~code:404 "no top handler" in
   let max_connections = max 4 max_connections in
   { new_thread; addr; port; sock; masksigpipe; handler;
     running= true; sem_max_connections=Sem_.create max_connections;
-    path_handlers=[]; max_keep_alive;
+    path_handlers=[];
     cb_encode_resp=[]; cb_decode_req=[];
   }
 
@@ -1009,11 +998,10 @@ let find_map f l =
   in aux f l
 
 let handle_client_ (self:t) (client_sock:Unix.file_descr) : unit =
-  let write_sock = Unix.dup client_sock in
-  let _ = Unix.set_nonblock client_sock in
-  let oc = Unix.out_channel_of_descr write_sock in
+  let ic = Unix.in_channel_of_descr client_sock in
+  let oc = Unix.out_channel_of_descr client_sock in
   let buf = Buf_.create() in
-  let is = Byte_stream.of_descr ~timeout:self.max_keep_alive client_sock in
+  let is = Byte_stream.of_chan ic in
   let continue = ref true in
   while !continue && self.running do
     _debug (fun k->k "read next request");
@@ -1127,24 +1115,17 @@ let run (self:t) : (unit,_) result =
     end;
     while self.running do
       (* limit concurrency *)
+      Sem_.acquire 1 self.sem_max_connections;
       try
-        Sem_.acquire 1 self.sem_max_connections;
         let client_sock, _ = Unix.accept sock in
         self.new_thread
           (fun () ->
             try
               handle_client_ self client_sock;
               Sem_.release 1 self.sem_max_connections;
-              _debug (fun k -> k
-                "closing inactive connections (%d connections active)"
-                (active_connections self))
-            with
-            | e ->
+            with e ->
               (try Unix.close client_sock with _ -> ());
               Sem_.release 1 self.sem_max_connections;
-              _debug (fun k -> k
-                "closing connections on error (%d connections active)"
-                (active_connections self));
               raise e
           );
       with e ->

src/Tiny_httpd.mli

@@ -434,7 +434,6 @@ type t
 val create :
   ?masksigpipe:bool ->
   ?max_connections:int ->
-  ?max_keep_alive:float ->
   ?new_thread:((unit -> unit) -> unit) ->
   ?addr:string ->
   ?port:int ->
@@ -454,11 +453,7 @@ val create :
     new client connection. By default it is {!Thread.create} but one
     could use a thread pool instead.
 
-    @param max_connections maximum number of simultaneous connections. Default 32.
-    @param max_keep_alive maximum number of seconds a thread in maintened for
-      a client with nothing to read. Default: -1.0, meaning threads are maintened
-      until client close the socket.
-      This parameter exists since 0.11.
+    @param max_connections maximum number of simultaneous connections.
     @param addr address (IPv4 or IPv6) to listen on. Default ["127.0.0.1"].
     @param port to listen on. Default [8080].
     @param sock an existing socket given to the server to listen on, e.g. by
@@ -477,10 +472,6 @@ val is_ipv6 : t -> bool
 val port : t -> int
 (** Port on which the server listens. *)
 
-val active_connections : t -> int
-(** number of currently opened connections with a client.
-    @since 0.11 *)
-
 val add_decode_request_cb :
   t ->
   (unit Request.t -> (unit Request.t * (byte_stream -> byte_stream)) option) -> unit
@@ -604,6 +595,10 @@ module type SERVER_SENT_GENERATOR = sig
     unit -> unit
   (** Send an event from the server.
       If data is a multiline string, it will be sent on separate "data:" lines. *)
+
+  val close : unit -> unit
+  (** Close connection.
+      @since 0.11 *)
 end
 
 type server_sent_generator = (module SERVER_SENT_GENERATOR)
@@ -648,3 +643,4 @@ val _debug : ((('a, out_channel, unit, unit, unit, unit) format6 -> 'a) -> unit)
 val _enable_debug: bool -> unit
 
 (**/**)
+

src/examples/dune

@@ -1,5 +0,0 @@
-
-(executables
-  (names echo)
-  (flags :standard -warn-error -a+8)
-  (libraries tiny_httpd tiny_httpd_camlzip))

tests/.gitignore

@@ -0,0 +1 @@
+foo_50

tests/dl-out.expect

@@ -0,0 +1,2 @@
+serve directory . on http://127.0.0.1:8084
+       0        0 52428800 data2

tests/download_chunked.sh

@@ -0,0 +1,13 @@
+#!/usr/bin/env sh
+
+SERVER=$1
+PORT=8084
+"$SERVER" . -p $PORT &
+
+sleep 0.1
+
+curl -N "http://localhost:${PORT}/foo_50" -o data2 \
+  -H 'Tranfer-encoding: chunked'
+
+kill %1
+wc data2

tests/dune

@@ -0,0 +1,52 @@
+
+(rule
+  (targets echo1.out)
+  (deps (:bin ../examples/echo.exe))
+  (locks /port)
+  (enabled_if (= %{system} "linux"))
+  (action (with-stdout-to %{targets} (run ./echo1.sh %{bin}))))
+
+(rule
+  (alias runtest)
+  (action (diff echo1.expect echo1.out)))
+
+(rule
+  (targets sse_count.out)
+  (deps (:bin ../examples/sse_server.exe))
+  (locks /port)
+  (enabled_if (= %{system} "linux"))
+  (action (with-stdout-to %{targets} (run ./sse_count.sh %{bin}))))
+
+(rule
+  (alias runtest)
+  (action (diff sse_count.expect sse_count.out)))
+
+(rule
+  (targets upload-out)
+  (deps (:bin ../src/bin/http_of_dir.exe) foo_50)
+  (locks /port)
+  (enabled_if (= %{system} "linux"))
+  (action (with-stdout-to %{targets}
+                          (run ./upload_chunked.sh %{bin}))))
+
+(rule
+  (alias runtest)
+  (action (diff upload-out.expect upload-out)))
+
+(rule
+  (targets dl-out)
+  (deps (:bin ../src/bin/http_of_dir.exe) foo_50)
+  (locks /port)
+  (enabled_if (= %{system} "linux"))
+  (action (with-stdout-to %{targets}
+                          (run ./download_chunked.sh %{bin}))))
+
+(rule
+  (alias runtest)
+  (action (diff dl-out.expect dl-out)))
+
+
+(rule
+  (targets foo_50)
+  (action
+    (bash "dd if=/dev/zero of=%{targets} bs=1M count=50")))

tests/echo1.expect

@@ -0,0 +1,9 @@
+listening on http://127.0.0.1:8085
+echo:
+{meth=GET; host=localhost:8085;
+ headers=[user-agent: test
+          accept: */*
+          host: localhost:8085];
+ path="/echo/?a=b&c=d"; body=""; path_components=["echo"];
+ query=["c","d";"a","b"]}
+(query: "c" = "d";"a" = "b")

tests/echo1.sh

@@ -0,0 +1,9 @@
+#!/usr/bin/env sh
+
+ECHO=$1
+PORT=8085
+
+"$ECHO" -p $PORT &
+sleep 0.1
+curl -N "http://localhost:${PORT}/echo/?a=b&c=d" -H user-agent:test
+kill %1

tests/sse_count.expect

@@ -0,0 +1,23 @@
+listening on http://localhost:8086/
+data: 0
+
+data: 1
+
+data: 2
+
+data: 3
+
+data: 4
+
+data: 5
+
+data: 6
+
+data: 7
+
+data: 8
+
+data: 9
+
+data: 10
+

tests/sse_count.sh

@@ -0,0 +1,10 @@
+#!/usr/bin/env sh
+
+SSE_SERVER=$1
+PORT=8086
+
+"$SSE_SERVER" -p $PORT &
+sleep 0.1
+
+curl -N "http://localhost:${PORT}/count/10" -H user-agent:test
+kill %1

tests/upload-out.expect

@@ -0,0 +1,2 @@
+serve directory . on http://127.0.0.1:8087
+upload successful       0        0 52428800 data

tests/upload_chunked.sh

@@ -0,0 +1,15 @@
+#!/usr/bin/env sh
+
+rm data
+
+SERVER=$1
+PORT=8087
+
+"$SERVER" . -p $PORT --upload --max-upload 100000000000 &
+
+sleep 0.1
+
+cat foo_50 | curl -N -X PUT http://localhost:$PORT/data --data-binary @- -H 'Transfer-Encoding: chunked'
+
+kill %1
+wc data

tiny_httpd.opam

@@ -15,6 +15,7 @@ depends: [
   "ocaml" { >= "4.04.0" }
   "odoc" {with-doc}
   "qtest" { >= "2.9" & with-test}
+  "conf-libcurl" {with-test}
   "qcheck" {with-test & >= "0.9" }
   "ounit2" {with-test}
   "ptime" {with-test}