Merge pull request #260 from Julow/ocsigen_response_eio

Eio-compatible representation for response Body

Author: Julow

Makefile.options

@@ -23,6 +23,6 @@ INCS=   -I ${BLD}/server/.ocsigenserver.objs/byte \
 ## ${SERVER_PACKAGE} is not only used to build the 'ocsigenserver' executable
 ## but also to generate src/baselib/ocsigen_config_static.ml
 
-SERVER_PACKAGE := lwt_ssl,bytes,lwt.unix,logs,logs-syslog,syslog-message,ipaddr,findlib,cryptokit,re,str,xml-light,dynlink,cohttp-lwt-unix
+SERVER_PACKAGE := lwt_ssl,bytes,lwt.unix,logs,logs-syslog,syslog-message,ipaddr,findlib,cryptokit,re,str,xml-light,dynlink,cohttp-lwt-unix,http
 
 LIBS := -package ${SERVER_PACKAGE} ${INCS}

configure

@@ -396,6 +396,7 @@ check_library cohttp "See: https://github.com/mirage/ocaml-cohttp"
 check_library cohttp-lwt-unix "Missing support for 'lwt' in cohttp."
 check_library react "See: http://erratique.ch/software/react"
 check_library ssl "See: http://sourceforge.net/projects/savonet/files/ocaml-ssl"
+check_library http ""
 
 check_library lwt "See: http://ocsigen.org/lwt"
 check_library lwt.unix  "Missing support for 'unix' in lwt."

dune-project

@@ -20,6 +20,7 @@
   (camlzip (>= 1.04))
   (cohttp-lwt-unix (and (>= 5.0) (< 6.0)))
   (conduit-lwt-unix (and (>= 2.0) (< 7.0)))
+  http
   cryptokit
   (ipaddr (>= 2.1))
   (lwt (>= 3.0))

ocsigenserver.opam

@@ -15,6 +15,7 @@ depends: [
   "camlzip" {>= "1.04"}
   "cohttp-lwt-unix" {>= "5.0" & < "6.0"}
   "conduit-lwt-unix" {>= "2.0" & < "7.0"}
+  "http"
   "cryptokit"
   "ipaddr" {>= "2.1"}
   "lwt" {>= "3.0"}

src/extensions/deflatemod.ml

@@ -59,131 +59,83 @@ let gzip_header =
 type output_buffer =
   { stream : Zlib.stream
   ; buf : bytes
-  ; mutable pos : int
-  ; mutable avail : int
+  ; flush : string -> unit Lwt.t
   ; mutable size : int32
-  ; mutable crc : int32
-  ; mutable add_trailer : bool }
+  ; mutable crc : int32 }
 
-let write_int32 oz n =
+let write_int32 buf offset n =
   for i = 0 to 3 do
-    Bytes.set oz.buf (oz.pos + i)
+    Bytes.set buf (offset + i)
       (Char.chr (Int32.to_int (Int32.shift_right_logical n (8 * i)) land 0xff))
-  done;
-  oz.pos <- oz.pos + 4;
-  oz.avail <- oz.avail - 4;
-  assert (oz.avail >= 0)
+  done
 
-(* puts in oz the content of buf, from pos to pos + len ;
- * f is the continuation of the current stream *)
-let rec output oz f buf pos len =
-  assert (pos >= 0 && len >= 0 && pos + len <= String.length buf);
-  if oz.avail = 0
-  then (
-    let cont () = output oz f buf pos len in
-    Logs.info ~src:section (fun fmt ->
-      fmt "Flushing because output buffer is full");
-    flush oz cont)
-  else if len = 0
-  then next_cont oz f
+let compress_flush oz used_out =
+  Logs.debug ~src:section (fun fmt -> fmt "Flushing %d bytes" used_out);
+  if used_out > 0
+  then oz.flush (Bytes.sub_string oz.buf 0 used_out)
+  else Lwt.return_unit
+
+(* gzip trailer *)
+let write_trailer oz =
+  write_int32 oz.buf 0 oz.crc;
+  write_int32 oz.buf 4 oz.size;
+  compress_flush oz 8
+
+(* puts in oz the content of buf, from pos to pos + len ; *)
+let rec compress_output oz inbuf pos len =
+  if len = 0
+  then Lwt.return_unit
   else
     let (_ : bool), used_in, used_out =
       try
-        Zlib.deflate oz.stream
-          (Bytes.unsafe_of_string buf)
-          pos len oz.buf oz.pos oz.avail Zlib.Z_NO_FLUSH
+        Zlib.deflate_string oz.stream inbuf pos len oz.buf 0
+          (Bytes.length oz.buf) Zlib.Z_NO_FLUSH
       with Zlib.Error (s, s') ->
         raise
           (Ocsigen_stream.Stream_error
              ("Error during compression: " ^ s ^ " " ^ s'))
     in
-    oz.pos <- oz.pos + used_out;
-    oz.avail <- oz.avail - used_out;
-    oz.size <- Int32.add oz.size (Int32.of_int used_in);
-    oz.crc <- Zlib.update_crc_string oz.crc buf pos used_in;
-    output oz f buf (pos + used_in) (len - used_in)
+    compress_flush oz used_out >>= fun () ->
+    compress_output oz inbuf (pos + used_in) (len - used_in)
 
-(* Flush oz, ie. produces a new_stream with the content of oz, cleans it
- * and returns the continuation of the stream *)
-and flush oz cont =
-  let len = oz.pos in
-  if len = 0
-  then cont ()
-  else
-    let buf_len = Bytes.length oz.buf in
-    let s =
-      if len = buf_len
-      then Bytes.to_string oz.buf
-      else Bytes.sub_string oz.buf 0 len
-    in
-    Logs.info ~src:section (fun fmt -> fmt "Flushing!");
-    oz.pos <- 0;
-    oz.avail <- buf_len;
-    Ocsigen_stream.cont s cont
-
-and next_cont oz stream =
-  Ocsigen_stream.next (stream : string Ocsigen_stream.stream) >>= fun e ->
-  match e with
-  | Ocsigen_stream.Finished None ->
-      Logs.info ~src:section (fun fmt ->
-        fmt "End of stream: big cleaning for zlib");
-      (* loop until there is nothing left to compress and flush *)
-      let rec finish () =
-        (* buffer full *)
-        if oz.avail = 0
-        then flush oz finish
-        else
-          (* no more input, deflates only what were left because output buffer
-           * was full *)
-          let finished, (_ : int), used_out =
-            Zlib.deflate oz.stream oz.buf 0 0 oz.buf oz.pos oz.avail
-              Zlib.Z_FINISH
-          in
-          oz.pos <- oz.pos + used_out;
-          oz.avail <- oz.avail - used_out;
-          if not finished then finish () else write_trailer ()
-      and write_trailer () =
-        if oz.add_trailer && oz.avail < 8
-        then flush oz write_trailer
-        else (
-          if oz.add_trailer then (write_int32 oz oz.crc; write_int32 oz oz.size);
-          Logs.info ~src:section (fun fmt ->
-            fmt "Zlib.deflate finished, last flush");
-          flush oz (fun () -> Ocsigen_stream.empty None))
-      in
-      finish ()
-  | Ocsigen_stream.Finished (Some s) -> next_cont oz s
-  | Ocsigen_stream.Cont (s, f) -> output oz f s 0 (String.length s)
+let rec compress_finish oz =
+  Logs.debug ~src:section (fun fmt -> fmt "Finishing");
+  (* loop until there is nothing left to compress and flush *)
+  let finished, (_ : int), used_out =
+    Zlib.deflate oz.stream oz.buf 0 0 oz.buf 0 (Bytes.length oz.buf)
+      Zlib.Z_FINISH
+  in
+  compress_flush oz used_out >>= fun () ->
+  if not finished then compress_finish oz else Lwt.return_unit
 
 (* deflate param : true = deflate ; false = gzip (no header in this case) *)
-let compress deflate stream : string Ocsigen_stream.t =
+let compress_body deflate body =
+ fun flush ->
   let zstream = Zlib.deflate_init !compress_level deflate in
-  let finalize status =
-    Ocsigen_stream.finalize stream status >>= fun _e ->
-    (try Zlib.deflate_end zstream
-     with
-     (* ignore errors, deflate_end cleans everything anyway *)
-     | Zlib.Error _ ->
-       ());
-    Lwt.return (Logs.info ~src:section (fun fmt -> fmt "Zlib stream closed"))
-  in
   let oz =
     let buffer_size = !buffer_size in
     { stream = zstream
     ; buf = Bytes.create buffer_size
-    ; pos = 0
-    ; avail = buffer_size
+    ; flush
     ; size = 0l
-    ; crc = 0l
-    ; add_trailer = not deflate }
+    ; crc = 0l }
   in
-  let new_stream () = next_cont oz (Ocsigen_stream.get stream) in
-  Logs.info ~src:section (fun fmt -> fmt "Zlib stream initialized");
-  if deflate
-  then Ocsigen_stream.make ~finalize new_stream
-  else
-    Ocsigen_stream.make ~finalize (fun () ->
-      Ocsigen_stream.cont gzip_header new_stream)
+  (if deflate then Lwt.return_unit else flush gzip_header) >>= fun () ->
+  body (fun inbuf ->
+    let len = String.length inbuf in
+    oz.size <- Int32.add oz.size (Int32.of_int len);
+    oz.crc <- Zlib.update_crc_string oz.crc inbuf 0 len;
+    compress_output oz inbuf 0 len)
+  >>= fun () ->
+  compress_finish oz >>= fun () ->
+  (if deflate then Lwt.return_unit else write_trailer oz) >>= fun () ->
+  Logs.debug ~src:section (fun fmt -> fmt "Close stream");
+  (try Zlib.deflate_end zstream
+   with
+   (* ignore errors, deflate_end cleans everything anyway *)
+   | Zlib.Error _ ->
+     ());
+  Lwt.return_unit
 
 (* We implement Content-Encoding, not Transfer-Encoding *)
 type encoding = Deflate | Gzip | Id | Star | Not_acceptable
@@ -252,8 +204,8 @@ let stream_filter contentencoding url deflate choice res =
                match Ocsigen_header.Mime_type.parse contenttype with
                | None, _ | _, None -> Lwt.return res
                | Some a, Some b when should_compress (a, b) url choice ->
-                   let response, body = Ocsigen_response.to_cohttp res in
                    let response =
+                     let response = Ocsigen_response.response res in
                      let headers = Cohttp.Response.headers response in
                      let headers =
                        let name = Ocsigen_header.Name.(to_string etag) in
@@ -273,10 +225,10 @@ let stream_filter contentencoding url deflate choice res =
                        Cohttp.Response.headers
                      ; Cohttp.Response.encoding = Cohttp.Transfer.Chunked }
                    and body =
-                     Cohttp_lwt.Body.to_stream body
-                     |> Ocsigen_stream.of_lwt_stream |> compress deflate
-                     |> Ocsigen_stream.to_lwt_stream
-                     |> Cohttp_lwt.Body.of_stream
+                     Ocsigen_response.Body.make Cohttp.Transfer.Chunked
+                       (compress_body deflate
+                          (Ocsigen_response.Body.write
+                             (Ocsigen_response.body res)))
                    in
                    Lwt.return (Ocsigen_response.update res ~body ~response)
                | _ -> Lwt.return res)

src/files/ocsigenserver.conf/gen.ml

@@ -84,7 +84,7 @@ let deps () =
     ; "ocsigenserver" ]
   in
   let packages =
-    "lwt_ssl,bytes,lwt.unix,logs,logs-syslog.unix,syslog-message,ipaddr,findlib,cryptokit,re,str,xml-light,dynlink,cohttp-lwt-unix"
+    "lwt_ssl,bytes,lwt.unix,logs,logs-syslog.unix,syslog-message,ipaddr,findlib,cryptokit,re,str,xml-light,dynlink,cohttp-lwt-unix,http"
   in
   let deps = ref [] in
   let cmd = "ocamlfind query -p-format -recursive " ^ packages in

src/server/dune

@@ -5,6 +5,7 @@
  (libraries
   xml-light
   cohttp-lwt-unix
+  http
   polytables
   ocsigen_cookie_map
   baselib

src/server/ocsigen_cohttp.ml

@@ -54,32 +54,6 @@ module Cookie = struct
     Ocsigen_cookie_map.Map_path.fold serialize_cookies cookies headers
 end
 
-(* FIXME: secure *)
-let make_cookies_header path exp name c _secure =
-  Format.sprintf "%s=%s%s%s" name c
-    (*VVV encode = true? *)
-    ("; path=/" ^ Ocsigen_lib.Url.string_of_url_path ~encode:true path)
-    (* (if secure && slot.sl_ssl then "; secure" else "")^ *)
-    ""
-  ^
-  match exp with
-  | Some s -> "; expires=" ^ Ocsigen_lib.Date.to_string s
-  | None -> ""
-
-let make_cookies_headers path t hds =
-  Ocsigen_cookie_map.Map_inner.fold
-    (fun name c h ->
-       let open Ocsigen_cookie_map in
-       let exp, v, secure =
-         match c with
-         | OUnset -> Some 0., "", false
-         | OSet (t, v, secure) -> t, v, secure
-       in
-       Cohttp.Header.add h
-         Ocsigen_header.Name.(to_string set_cookie)
-         (make_cookies_header path exp name v secure))
-    t hds
-
 let handler ~ssl ~address ~port ~connector (flow, conn) request body =
   let filenames = ref [] in
   let edn = Conduit_lwt_unix.endp_of_flow flow in
@@ -130,7 +104,7 @@ let handler ~ssl ~address ~port ~connector (flow, conn) request body =
       | `Not_found -> "Not Found"
       | _ -> Printexc.to_string exn
     in
-    Cohttp_lwt_unix.Server.respond_error ?headers
+    Ocsigen_response.respond_error ?headers
       ~status:(ret_code :> Cohttp.Code.status_code)
       ~body ()
   in
@@ -155,32 +129,18 @@ let handler ~ssl ~address ~port ~connector (flow, conn) request body =
                   Ocsigen_header.Name.x_forwarded_for))
             (Uri.path (Ocsigen_request.uri request)));
        Lwt.catch
-         (fun () ->
-            connector request >>= fun response ->
-            let response, body = Ocsigen_response.to_cohttp response
-            and cookies = Ocsigen_response.cookies response in
-            let response =
-              let headers =
-                Cohttp.Header.add_unless_exists
-                  (Cohttp.Header.add_unless_exists
-                     (Ocsigen_cookie_map.Map_path.fold make_cookies_headers
-                        cookies
-                        (Cohttp.Response.headers response))
-                     "server" Ocsigen_config.server_name)
-                  "date"
-                  (Ocsigen_lib.Date.to_string (Unix.time ()))
-              in
-              {response with Cohttp.Response.headers}
-            in
-            Lwt.return (response, body))
+         (fun () -> connector request)
          (function
            | Ocsigen_is_dir fun_request ->
                let headers =
                  fun_request request |> Uri.to_string
                  |> Cohttp.Header.init_with "location"
                and status = `Moved_permanently in
-               Cohttp_lwt_unix.Server.respond ~headers ~status ~body:`Empty ()
-           | exn -> handle_error exn))
+               Lwt.return
+                 (Ocsigen_response.respond_string ~headers ~status ~body:"" ())
+           | exn -> Lwt.return (handle_error exn))
+       >>= fun response ->
+       Lwt.return (Ocsigen_response.to_response_expert response))
     (fun () ->
        if !filenames <> []
        then
@@ -236,7 +196,7 @@ let service ?ssl ~address ~port ~connector () =
     let ssl = match ssl with Some _ -> true | None -> false in
     handler ~ssl ~address ~port ~connector
   in
-  let config = Cohttp_lwt_unix.Server.make ~conn_closed ~callback () in
+  let config = Cohttp_lwt_unix.Server.make_expert ~conn_closed ~callback () in
   let mode =
     match address, tls_own_key with
     | `Unix f, _ -> `Unix_domain_socket (`File f)