simon/tiny_httpd
1
0
Fork 0
mirror of https://github.com/c-cube/tiny_httpd.git synced 2025-12-05 19:00:32 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions

finish refactor

Browse source
This commit is contained in:
Simon Cruanes 2024-02-26 15:48:10 -05:00
parent 22f158ccd8
commit adf4c6815f
No known key found for this signature in database
GPG key ID: EBFFF6F283F3A2B4
15 changed files with 125 additions and 226 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
dune-project
View file

@ -36,5 +36,6 @@
(depends
(tiny_httpd (= :version))
(camlzip (>= 1.06))
iostream-camlzip
(logs :with-test)
(odoc :with-doc)))

188
src/camlzip/Tiny_httpd_camlzip.ml
View file

@ -1,175 +1,59 @@
module S = Tiny_httpd_server
module BS = Tiny_httpd_stream
module W = Tiny_httpd_io.Writer
module Out = Tiny_httpd_io.Output
module Log = Tiny_httpd.Log
module W = IO.Writer
let decode_deflate_stream_ ~buf_size (is : S.byte_stream) : S.byte_stream =
(* TODO: just use iostream-camlzip? *)
let decode_deflate_stream_ ~buf_size (ic : IO.Input.t) : IO.Input.t =
Log.debug (fun k -> k "wrap stream with deflate.decode");
let zlib_str = Zlib.inflate_init false in
let is_done = ref false in
BS.make ~bs:(Bytes.create buf_size)
~close:(fun _ ->
Zlib.inflate_end zlib_str;
BS.close is)
~consume:(fun self len ->
if len > self.len then
S.Response.fail_raise ~code:400
"inflate: error during decompression: invalid consume len %d (max %d)"
len self.len;
self.off <- self.off + len;
self.len <- self.len - len)
~fill:(fun self ->
(* refill [buf] if needed *)
if self.len = 0 && not !is_done then (
is.fill_buf ();
(try
let finished, used_in, used_out =
Zlib.inflate zlib_str self.bs 0 (Bytes.length self.bs) is.bs is.off
is.len Zlib.Z_SYNC_FLUSH
in
is.consume used_in;
self.off <- 0;
self.len <- used_out;
if finished then is_done := true;
Log.debug (fun k ->
k "decode %d bytes as %d bytes from inflate (finished: %b)"
used_in used_out finished)
with Zlib.Error (e1, e2) ->
S.Response.fail_raise ~code:400
"inflate: error during decompression:\n%s %s" e1 e2);
Log.debug (fun k ->
k "inflate: refill %d bytes into internal buf" self.len)
))
()
Iostream_camlzip.decompress_in_buf ~buf_size ic
let encode_deflate_writer_ ~buf_size (w : W.t) : W.t =
Log.debug (fun k -> k "wrap writer with deflate.encode");
let zlib_str = Zlib.deflate_init 4 false in
let o_buf = Bytes.create buf_size in
let o_off = ref 0 in
let o_len = ref 0 in
(* write output buffer to out *)
let write_out (oc : Out.t) =
if !o_len > 0 then (
Out.output oc o_buf !o_off !o_len;
o_off := 0;
o_len := 0
)
let { IO.Writer.write } = w in
let write' (oc : IO.Output.t) =
let oc' = Iostream_camlzip.compressed_out ~buf_size ~level:4 oc in
write (oc' :> IO.Output.t)
in
IO.Writer.make ~write:write' ()
let flush_zlib ~flush (oc : Out.t) =
let continue = ref true in
while !continue do
let finished, used_in, used_out =
Zlib.deflate zlib_str Bytes.empty 0 0 o_buf 0 (Bytes.length o_buf) flush
in
assert (used_in = 0);
o_len := !o_len + used_out;
if finished then continue := false;
write_out oc
done
in
(* compress and consume input buffer *)
let write_zlib ~flush (oc : Out.t) buf i len =
let i = ref i in
let len = ref len in
while !len > 0 do
let _finished, used_in, used_out =
Zlib.deflate zlib_str buf !i !len o_buf 0 (Bytes.length o_buf) flush
in
i := !i + used_in;
len := !len - used_in;
o_len := !o_len + used_out;
write_out oc
done
in
let write (oc : Out.t) : unit =
let output buf i len = write_zlib ~flush:Zlib.Z_NO_FLUSH oc buf i len in
let bchar = Bytes.create 1 in
let output_char c =
Bytes.set bchar 0 c;
output bchar 0 1
in
let flush () =
flush_zlib oc ~flush:Zlib.Z_FINISH;
assert (!o_len = 0);
oc.flush ()
in
let close () =
flush ();
Zlib.deflate_end zlib_str;
oc.close ()
in
(* new output channel that compresses on the fly *)
let oc' = { Out.flush; close; output; output_char } in
w.write oc';
oc'.close ()
in
W.make ~write ()
let split_on_char ?(f = fun x -> x) c s : string list =
let rec loop acc i =
match String.index_from s i c with
| exception Not_found ->
let acc =
if i = String.length s then
acc
else
f (String.sub s i (String.length s - i)) :: acc
in
List.rev acc
| j ->
let acc = f (String.sub s i (j - i)) :: acc in
loop acc (j + 1)
in
loop [] 0
let accept_deflate (req : _ S.Request.t) =
match S.Request.get_header req "Accept-Encoding" with
| Some s -> List.mem "deflate" @@ split_on_char ~f:String.trim ',' s
let accept_deflate (req : _ Request.t) =
match Request.get_header req "Accept-Encoding" with
| Some s ->
List.mem "deflate" @@ List.rev_map String.trim @@ String.split_on_char ',' s
| None -> false
let has_deflate s =
try Scanf.sscanf s "deflate, %s" (fun _ -> true) with _ -> false
(* decompress [req]'s body if needed *)
let decompress_req_stream_ ~buf_size (req : BS.t S.Request.t) : _ S.Request.t =
match S.Request.get_header ~f:String.trim req "Transfer-Encoding" with
let decompress_req_stream_ ~buf_size (req : IO.Input.t Request.t) : _ Request.t
=
match Request.get_header ~f:String.trim req "Transfer-Encoding" with
(* TODO
| Some "gzip" ->
let req' = S.Request.set_header req "Transfer-Encoding" "chunked" in
Some (req', decode_gzip_stream_)
*)
| Some "deflate" ->
let body' = S.Request.body req |> decode_deflate_stream_ ~buf_size in
req
|> S.Request.remove_header "Transfer-Encoding"
|> S.Request.set_body body'
let body' = Request.body req |> decode_deflate_stream_ ~buf_size in
req |> Request.remove_header "Transfer-Encoding" |> Request.set_body body'
| Some s when has_deflate s ->
(match Scanf.sscanf s "deflate, %s" (fun s -> s) with
| tr' ->
let body' = S.Request.body req |> decode_deflate_stream_ ~buf_size in
let body' = Request.body req |> decode_deflate_stream_ ~buf_size in
req
|> S.Request.set_header "Transfer-Encoding" tr'
|> S.Request.set_body body'
|> Request.set_header "Transfer-Encoding" tr'
|> Request.set_body body'
| exception _ -> req)
| _ -> req
let compress_resp_stream_ ~compress_above ~buf_size (req : _ S.Request.t)
(resp : S.Response.t) : S.Response.t =
let compress_resp_stream_ ~compress_above ~buf_size (req : _ Request.t)
(resp : Response.t) : Response.t =
(* headers for compressed stream *)
let update_headers h =
h
|> S.Headers.remove "Content-Length"
|> S.Headers.set "Content-Encoding" "deflate"
|> Headers.remove "Content-Length"
|> Headers.set "Content-Encoding" "deflate"
in
if accept_deflate req then (
@ -181,25 +65,25 @@ let compress_resp_stream_ ~compress_above ~buf_size (req : _ S.Request.t)
(String.length s) compress_above);
let body = encode_deflate_writer_ ~buf_size @@ W.of_string s in
resp
|> S.Response.update_headers update_headers
|> S.Response.set_body (`Writer body)
| `Stream str ->
|> Response.update_headers update_headers
|> Response.set_body (`Writer body)
| `Stream ic ->
Log.debug (fun k -> k "encode stream response with deflate");
let w = BS.to_writer str in
let w = IO.Writer.of_input ic in
resp
|> S.Response.update_headers update_headers
|> S.Response.set_body (`Writer (encode_deflate_writer_ ~buf_size w))
|> Response.update_headers update_headers
|> Response.set_body (`Writer (encode_deflate_writer_ ~buf_size w))
| `Writer w ->
Log.debug (fun k -> k "encode writer response with deflate");
resp
|> S.Response.update_headers update_headers
|> S.Response.set_body (`Writer (encode_deflate_writer_ ~buf_size w))
|> Response.update_headers update_headers
|> Response.set_body (`Writer (encode_deflate_writer_ ~buf_size w))
| `String _ | `Void -> resp
) else
resp
let middleware ?(compress_above = 16 * 1024) ?(buf_size = 16 * 1_024) () :
S.Middleware.t =
Server.Middleware.t =
let buf_size = max buf_size 1_024 in
fun h req ~resp ->
let req = decompress_req_stream_ ~buf_size req in
@ -209,4 +93,4 @@ let middleware ?(compress_above = 16 * 1024) ?(buf_size = 16 * 1_024) () :
let setup ?compress_above ?buf_size server =
let m = middleware ?compress_above ?buf_size () in
Log.info (fun k -> k "setup gzip middleware");
S.add_middleware ~stage:`Encoding server m
Server.add_middleware ~stage:`Encoding server m

4
src/camlzip/Tiny_httpd_camlzip.mli
View file

@ -7,7 +7,7 @@
*)
val middleware :
?compress_above:int -> ?buf_size:int -> unit -> Tiny_httpd_server.Middleware.t
?compress_above:int -> ?buf_size:int -> unit -> Server.Middleware.t
(** Middleware responsible for deflate compression/decompression.
@param compress_above threshold, in bytes, above which a response body
that has a known content-length is compressed. Stream bodies
@ -15,7 +15,7 @@ val middleware :
@param buf_size size of the underlying buffer for compression/decompression
@since 0.11 *)
val setup : ?compress_above:int -> ?buf_size:int -> Tiny_httpd_server.t -> unit
val setup : ?compress_above:int -> ?buf_size:int -> Server.t -> unit
(** Install middleware for tiny_httpd to be able to encode/decode
compressed streams
@param compress_above threshold above with string responses are compressed

4
src/camlzip/dune
View file

@ -2,5 +2,5 @@
(name tiny_httpd_camlzip)
(public_name tiny_httpd_camlzip)
(synopsis "A wrapper around camlzip to bring compression to Tiny_httpd")
(flags :standard -safe-string -warn-error -a+8)
(libraries tiny_httpd camlzip))
(flags :standard -open Tiny_httpd_core -safe-string -warn-error -a+8)
(libraries tiny_httpd.core iostream-camlzip camlzip))

42
src/core/IO.ml
View file

@ -109,7 +109,7 @@ module Input = struct
(fd : Unix.file_descr) : t =
let eof = ref false in
object
inherit Iostream.In_buf.t_from_refill ~buf ()
inherit Iostream.In_buf.t_from_refill ~bytes:buf.bytes ()
method private refill (slice : Slice.t) =
if not !eof then (
@ -137,7 +137,7 @@ module Input = struct
if slice.len = 0 then eof := true
)
method! close () =
method close () =
if not !closed then (
closed := true;
eof := true;
@ -150,13 +150,13 @@ module Input = struct
let of_slice (slice : Slice.t) : t =
object
inherit Iostream.In_buf.t_from_refill ~buf:slice ()
inherit Iostream.In_buf.t_from_refill ~bytes:slice.bytes ()
method private refill (slice : Slice.t) =
slice.off <- 0;
slice.len <- 0
method! close () = ()
method close () = ()
end
(** Read into the given slice.
@ -198,7 +198,7 @@ module Input = struct
slice.off <- 0;
input_rec slice
method! close () =
method close () =
close i1;
close i2
end
@ -225,10 +225,10 @@ module Input = struct
Stdlib.output oc slice.bytes slice.off slice.len)
self
let to_chan' (oc : #Iostream.Out_buf.t) (self : #t) : unit =
let to_chan' (oc : #Iostream.Out.t) (self : #t) : unit =
iter_slice
(fun (slice : Slice.t) ->
Iostream.Out_buf.output oc slice.bytes slice.off slice.len)
Iostream.Out.output oc slice.bytes slice.off slice.len)
self
let read_all_using ~buf (self : #t) : string =
@ -299,11 +299,10 @@ module Input = struct
@param close_rec if true, closing this will also close the input stream *)
let limit_size_to ~close_rec ~max_size ~(bytes : bytes) (arg : t) : t =
let remaining_size = ref max_size in
let slice = Slice.of_bytes bytes in
object
inherit Iostream.In_buf.t_from_refill ~buf:slice ()
method! close () = if close_rec then close arg
inherit Iostream.In_buf.t_from_refill ~bytes ()
method close () = if close_rec then close arg
method private refill slice =
if slice.len = 0 then
@ -324,12 +323,11 @@ module Input = struct
@param close_rec if true, closing this will also close the input stream *)
let reading_exactly ~close_rec ~size ~(bytes : bytes) (arg : t) : t =
let remaining_size = ref size in
let slice = Slice.of_bytes bytes in
object
inherit Iostream.In_buf.t_from_refill ~buf:slice ()
inherit Iostream.In_buf.t_from_refill ~bytes ()
method! close () =
method close () =
if !remaining_size > 0 then skip arg !remaining_size;
if close_rec then close arg
@ -346,9 +344,11 @@ module Input = struct
)
end
let read_chunked ~(buf : Slice.t) ~fail (bs : #t) : t =
let read_chunked ~(bytes : bytes) ~fail (bs : #t) : t =
let first = ref true in
let line_buf = Buf.create ~size:128 () in
(* small buffer to read the chunk sizes *)
let line_buf = Buf.create ~size:32 () in
let read_next_chunk_len () : int =
if !first then
first := false
@ -377,7 +377,7 @@ module Input = struct
let chunk_size = ref 0 in
object
inherit t_from_refill ~buf ()
inherit t_from_refill ~bytes ()
method private refill (slice : Slice.t) : unit =
if !chunk_size = 0 && not !eof then chunk_size := read_next_chunk_len ();
@ -395,10 +395,7 @@ module Input = struct
(* stream is finished *)
eof := true
method! close () =
(* do not close underlying stream *)
eof := true;
()
method close () = eof := true (* do not close underlying stream *)
end
(** Output a stream using chunked encoding *)
@ -435,14 +432,15 @@ module Writer = struct
let[@inline] make ~write () : t = { write }
(** Write into the channel. *)
let[@inline] write (oc : Output.t) (self : t) : unit = self.write oc
let[@inline] write (oc : #Output.t) (self : t) : unit =
self.write (oc :> Output.t)
(** Empty writer, will output 0 bytes. *)
let empty : t = { write = ignore }
(** A writer that just emits the bytes from the given string. *)
let[@inline] of_string (str : string) : t =
let write oc = Output.output_string oc str in
let write oc = Iostream.Out.output_string oc str in
{ write }
let[@inline] of_input (ic : #Input.t) : t =

1
src/core/buf.ml
View file

@ -4,6 +4,7 @@ let create ?(size = 4_096) () : t =
let bytes = Bytes.make size ' ' in
{ bytes; i = 0; original = bytes }
let of_bytes bytes : t = { bytes; i = 0; original = bytes }
let[@inline] size self = self.i
let[@inline] bytes_slice self = self.bytes

1
src/core/buf.mli
View file

@ -11,6 +11,7 @@ type t
val size : t -> int
val clear : t -> unit
val create : ?size:int -> unit -> t
val of_bytes : bytes -> t
val contents : t -> string
val clear_and_zero : t -> unit

25
src/core/request.ml
View file

@ -24,6 +24,9 @@ let start_time self = self.start_time
let query self = self.query
let get_header ?f self h = Headers.get ?f h self.headers
let remove_header k self = { self with headers = Headers.remove k self.headers }
let add_meta self k v = self.meta <- Hmap.add k v self.meta
let get_meta self k = Hmap.find k self.meta
let get_meta_exn self k = Hmap.get k self.meta
let get_header_int self h =
match get_header self h with
@ -64,9 +67,9 @@ let pp out self : unit =
self.body pp_comp_ self.path_components pp_query self.query
(* decode a "chunked" stream into a normal stream *)
let read_stream_chunked_ ~buf (bs : #IO.Input.t) : IO.Input.t =
let read_stream_chunked_ ~bytes (bs : #IO.Input.t) : IO.Input.t =
Log.debug (fun k -> k "body: start reading chunked stream...");
IO.Input.read_chunked ~buf ~fail:(fun s -> Bad_req (400, s)) bs
IO.Input.read_chunked ~bytes ~fail:(fun s -> Bad_req (400, s)) bs
let limit_body_size_ ~max_size ~bytes (bs : #IO.Input.t) : IO.Input.t =
Log.debug (fun k -> k "limit size of body to max-size=%d" max_size);
@ -146,8 +149,8 @@ let parse_req_start ~client_addr ~get_time_s ~buf (bs : IO.Input.t) :
(* parse body, given the headers.
@param tr_stream a transformation of the input stream. *)
let parse_body_ ~tr_stream ~buf (req : IO.Input.t t) : IO.Input.t t resp_result
=
let parse_body_ ~tr_stream ~bytes (req : IO.Input.t t) :
IO.Input.t t resp_result =
try
let size =
match Headers.get_exn "Content-Length" req.headers |> int_of_string with
@ -162,7 +165,9 @@ let parse_body_ ~tr_stream ~buf (req : IO.Input.t t) : IO.Input.t t resp_result
read_exactly ~size ~bytes @@ tr_stream req.body
| Some "chunked" ->
(* body sent by chunks *)
let bs : IO.Input.t = read_stream_chunked_ ~buf @@ tr_stream req.body in
let bs : IO.Input.t =
read_stream_chunked_ ~bytes @@ tr_stream req.body
in
if size > 0 then (
let bytes = Bytes.create 4096 in
limit_body_size_ ~max_size:size ~bytes bs
@ -176,11 +181,11 @@ let parse_body_ ~tr_stream ~buf (req : IO.Input.t t) : IO.Input.t t resp_result
| Bad_req (c, s) -> Error (c, s)
| e -> Error (400, Printexc.to_string e)
let read_body_full ?buf ?buf_size (self : IO.Input.t t) : string t =
let read_body_full ?bytes ?buf_size (self : IO.Input.t t) : string t =
try
let buf =
match buf with
| Some b -> b
match bytes with
| Some b -> Buf.of_bytes b
| None -> Buf.create ?size:buf_size ()
in
let body = IO.Input.read_all_using ~buf self.body in
@ -196,8 +201,8 @@ module Private_ = struct
let parse_req_start_exn ?(buf = Buf.create ()) ~client_addr ~get_time_s bs =
parse_req_start ~client_addr ~get_time_s ~buf bs |> unwrap_resp_result
let parse_body ?(buf = IO.Slice.create 4096) req bs : _ t =
parse_body_ ~tr_stream:(fun s -> s) ~buf { req with body = bs }
let parse_body ?(bytes = Bytes.create 4096) req bs : _ t =
parse_body_ ~tr_stream:(fun s -> s) ~bytes { req with body = bs }
|> unwrap_resp_result
let[@inline] set_body body self = { self with body }

19
src/core/request.mli
View file

@ -36,6 +36,19 @@ type 'body t = private {
@since 0.11 the field [start_time] was added
*)
val add_meta : _ t -> 'a Hmap.key -> 'a -> unit
(** Add metadata
@since NEXT_RELEASE *)
val get_meta : _ t -> 'a Hmap.key -> 'a option
(** Get metadata
@since NEXT_RELEASE *)
val get_meta_exn : _ t -> 'a Hmap.key -> 'a
(** Like {!get_meta} but can fail
@raise Invalid_argument if not present
@since NEXT_RELEASE *)
val pp : Format.formatter -> string t -> unit
(** Pretty print the request and its body. The exact format of this printing
is not specified. *)
@ -102,11 +115,11 @@ val limit_body_size :
@since 0.3
*)
val read_body_full : ?buf:Buf.t -> ?buf_size:int -> IO.Input.t t -> string t
val read_body_full : ?bytes:bytes -> ?buf_size:int -> IO.Input.t t -> string t
(** Read the whole body into a string. Potentially blocking.
@param buf_size initial size of underlying buffer (since 0.11)
@param buf the initial buffer (since 0.14)
@param bytes the initial buffer (since 0.14)
*)
(**/**)
@ -128,7 +141,7 @@ module Private_ : sig
unit t option
val close_after_req : _ t -> bool
val parse_body : ?buf:IO.Slice.t -> unit t -> IO.Input.t -> IO.Input.t t
val parse_body : ?bytes:bytes -> unit t -> IO.Input.t -> IO.Input.t t
val set_body : 'a -> _ t -> 'a t
end

4
src/core/response.ml
View file

@ -76,12 +76,12 @@ let pp out self : unit =
Format.fprintf out "{@[code=%d;@ headers=[@[%a@]];@ body=%a@]}" self.code
Headers.pp self.headers pp_body self.body
let output_ ~buf (oc : IO.Output.t) (self : t) : unit =
let output_ ~bytes (oc : IO.Output.t) (self : t) : unit =
(* double indirection:
- print into [buffer] using [bprintf]
- transfer to [buf_] so we can output from there *)
let tmp_buffer = Buffer.create 32 in
Buf.clear buf;
let buf = Buf.of_bytes bytes in
(* write start of reply *)
Printf.bprintf tmp_buffer "HTTP/1.1 %d %s\r\n" self.code

2
src/core/response.mli
View file

@ -112,7 +112,7 @@ val pp : Format.formatter -> t -> unit
module Private_ : sig
val make_void_force_ : ?headers:Headers.t -> code:int -> unit -> t
val output_ : buf:Buf.t -> IO.Output.t -> t -> unit
val output_ : bytes:Bytes.t -> IO.Output.t -> t -> unit
end
(**/**)

47
src/core/server.ml
View file

@ -72,7 +72,6 @@ let unwrap_handler_result req = function
type t = {
backend: (module IO_BACKEND);
mutable tcp_server: IO.TCP_server.t option;
buf_size: int;
mutable handler: IO.Input.t Request.t -> Response.t;
(** toplevel handler, if any *)
mutable middlewares: (int * Middleware.t) list; (** Global middlewares *)
@ -80,7 +79,7 @@ type t = {
(** sorted version of {!middlewares} *)
mutable path_handlers: (unit Request.t -> handler_result option) list;
(** path handlers *)
buf_pool: Buf.t Pool.t;
bytes_pool: bytes Pool.t;
}
let addr (self : t) =
@ -169,8 +168,8 @@ let add_route_handler (type a) ?accept ?middlewares ?meth self
(route : (a, _) Route.t) (f : _) : unit =
let tr_req _oc req ~resp f =
let req =
Pool.with_resource self.buf_pool @@ fun buf ->
Request.read_body_full ~buf req
Pool.with_resource self.bytes_pool @@ fun bytes ->
Request.read_body_full ~bytes req
in
resp (f req)
in
@ -190,8 +189,8 @@ exception Exit_SSE
let add_route_server_sent_handler ?accept self route f =
let tr_req (oc : IO.Output.t) req ~resp f =
let req =
Pool.with_resource self.buf_pool @@ fun buf ->
Request.read_body_full ~buf req
Pool.with_resource self.bytes_pool @@ fun bytes ->
Request.read_body_full ~bytes req
in
let headers =
ref Headers.(empty |> set "content-type" "text/event-stream")
@ -256,6 +255,8 @@ let add_upgrade_handler ?(accept = fun _ -> Ok ()) (self : t) route f : unit =
in
self.path_handlers <- ph :: self.path_handlers
let clear_bytes_ bs = Bytes.fill bs 0 (Bytes.length bs) '\x00'
let create_from ?(buf_size = 16 * 1_024) ?(middlewares = []) ~backend () : t =
let handler _req = Response.fail ~code:404 "no top handler" in
let self =
@ -263,13 +264,12 @@ let create_from ?(buf_size = 16 * 1_024) ?(middlewares = []) ~backend () : t =
backend;
tcp_server = None;
handler;
buf_size;
path_handlers = [];
middlewares = [];
middlewares_sorted = lazy [];
buf_pool =
Pool.create ~clear:Buf.clear_and_zero
~mk_item:(fun () -> Buf.create ~size:buf_size ())
bytes_pool =
Pool.create ~clear:clear_bytes_
~mk_item:(fun () -> Bytes.create buf_size)
();
}
in
@ -302,8 +302,8 @@ let string_as_list_contains_ (s : string) (sub : string) : bool =
(* handle client on [ic] and [oc] *)
let client_handle_for (self : t) ~client_addr ic oc : unit =
Pool.with_resource self.buf_pool @@ fun buf ->
Pool.with_resource self.buf_pool @@ fun buf_res ->
Pool.with_resource self.bytes_pool @@ fun bytes_req ->
Pool.with_resource self.bytes_pool @@ fun bytes_res ->
let (module B) = self.backend in
(* how to log the response to this query *)
@ -336,7 +336,7 @@ let client_handle_for (self : t) ~client_addr ic oc : unit =
let msg = Printexc.to_string e in
let resp = Response.fail ~code:500 "server error: %s" msg in
if not Log.dummy then log_exn msg bt;
Response.Private_.output_ ~buf:buf_res oc resp
Response.Private_.output_ ~bytes:bytes_res oc resp
in
let handle_bad_req req e bt =
@ -346,7 +346,7 @@ let client_handle_for (self : t) ~client_addr ic oc : unit =
log_exn msg bt;
log_response req resp
);
Response.Private_.output_ ~buf:buf_res oc resp
Response.Private_.output_ ~bytes:bytes_res oc resp
in
let handle_upgrade req (module UP : UPGRADE_HANDLER) : unit =
@ -368,7 +368,7 @@ let client_handle_for (self : t) ~client_addr ic oc : unit =
Log.error (fun k -> k "upgrade failed: %s" msg);
let resp = Response.make_raw ~code:429 "upgrade required" in
log_response req resp;
Response.Private_.output_ ~buf:buf_res oc resp
Response.Private_.output_ ~bytes:bytes_res oc resp
| Ok (headers, handshake_st) ->
(* send the upgrade reply *)
let headers =
@ -376,7 +376,7 @@ let client_handle_for (self : t) ~client_addr ic oc : unit =
in
let resp = Response.make_string ~code:101 ~headers (Ok "") in
log_response req resp;
Response.Private_.output_ ~buf:buf_res oc resp;
Response.Private_.output_ ~bytes:bytes_res oc resp;
UP.handle_connection client_addr handshake_st ic oc
with e ->
@ -388,6 +388,7 @@ let client_handle_for (self : t) ~client_addr ic oc : unit =
let handle_one_req () =
match
let buf = Buf.of_bytes bytes_req in
Request.Private_.parse_req_start ~client_addr ~get_time_s:B.get_time_s
~buf ic
with
@ -395,7 +396,7 @@ let client_handle_for (self : t) ~client_addr ic oc : unit =
| Error (c, s) ->
(* connection error, close *)
let res = Response.make_raw ~code:c s in
(try Response.Private_.output_ ~buf:buf_res oc res
(try Response.Private_.output_ ~bytes:bytes_res oc res
with Sys_error _ -> ());
continue := false
| Ok (Some req) ->
@ -418,7 +419,7 @@ let client_handle_for (self : t) ~client_addr ic oc : unit =
(match Request.get_header ~f:String.trim req "Expect" with
| Some "100-continue" ->
Log.debug (fun k -> k "send back: 100 CONTINUE");
Response.Private_.output_ ~buf:buf_res oc
Response.Private_.output_ ~bytes:bytes_res oc
(Response.make_raw ~code:100 "")
| Some s -> bad_reqf 417 "unknown expectation %s" s
| None -> ());
@ -432,11 +433,7 @@ let client_handle_for (self : t) ~client_addr ic oc : unit =
in
(* now actually read request's body into a stream *)
let req =
Request.Private_.parse_body
~buf:(IO.Slice.of_bytes (Buf.bytes_slice buf))
req ic
in
let req = Request.Private_.parse_body ~bytes:bytes_req req ic in
(* how to reply *)
let resp r =
@ -444,7 +441,7 @@ let client_handle_for (self : t) ~client_addr ic oc : unit =
if Headers.get "connection" r.Response.headers = Some "close" then
continue := false;
log_response req r;
Response.Private_.output_ ~buf:buf_res oc r
Response.Private_.output_ ~bytes:bytes_res oc r
with Sys_error e ->
Log.debug (fun k ->
k "error when writing response: %s@.connection broken" e);
@ -466,7 +463,7 @@ let client_handle_for (self : t) ~client_addr ic oc : unit =
continue := false;
let resp = Response.make_raw ~code s in
log_response req resp;
Response.Private_.output_ ~buf:buf_res oc resp
Response.Private_.output_ ~bytes:bytes_res oc resp
| Upgrade _ as e -> raise e
| e ->
let bt = Printexc.get_raw_backtrace () in

7
src/html/Tiny_httpd_html.ml
View file

@ -14,12 +14,11 @@ include Html_
be a "html" tag.
@since 0.14
*)
let to_output ?(top = false) (self : elt) (out : IO.Output.t) : unit =
let to_output ?(top = false) (self : elt) (out : #IO.Output.t) : unit =
let out = Out.create_of_out out in
if top then Out.add_string out "<!DOCTYPE html>\n";
self out;
Out.add_format_nl out;
Out.flush out
Out.add_format_nl out
(** Convert a HTML element to a string.
@param top if true, add DOCTYPE at the beginning. The top element should then
@ -54,7 +53,7 @@ let to_out_channel_top = to_output ~top:true
@param top if true, add a DOCTYPE. See {!to_out_channel}.
@since 0.14 *)
let to_writer ?top (self : elt) : IO.Writer.t =
let write oc = to_output ?top self oc in
let write (oc : #IO.Output.t) = to_output ?top self oc in
IO.Writer.make ~write ()
(** Convert a HTML element to a stream. This might just convert

4
src/ws/tiny_httpd_ws.ml
View file

@ -380,13 +380,13 @@ let upgrade ic oc : _ * _ =
let reader = Reader.create ~ic ~writer () in
let ws_ic : IO.Input.t =
object
inherit IO.Input.t_from_refill ()
inherit IO.Input.t_from_refill ~bytes:(Bytes.create 4_096) ()
method private refill (slice : IO.Slice.t) =
slice.off <- 0;
slice.len <- Reader.read reader slice.bytes 0 (Bytes.length slice.bytes)
method! close () = Reader.close reader
method close () = Reader.close reader
end
in
let ws_oc : IO.Output.t =

2
vendor/iostream vendored

@ -1 +1 @@
Subproject commit bb03b78fab03e9eb7186c45d33d23735650351ec
Subproject commit 668a7c22c09d21293c9ce3fd8bc66b3080c525d2