mirror of
https://github.com/c-cube/tiny_httpd.git
synced 2025-12-07 03:35:34 -05:00
564 lines
18 KiB
OCaml
564 lines
18 KiB
OCaml
(** IO abstraction.
|
|
|
|
We abstract IO so we can support classic unix blocking IOs
|
|
with threads, and modern async IO with Eio.
|
|
|
|
{b NOTE}: experimental.
|
|
|
|
@since 0.14
|
|
*)
|
|
|
|
open Common_
|
|
module Buf = Buf
|
|
module Slice = Iostream.Slice
|
|
module A = Atomic_
|
|
|
|
(** Output channel (byte sink) *)
|
|
module Output = struct
|
|
include Iostream.Out_buf
|
|
|
|
class of_unix_fd ?(close_noerr = false) ~closed ~(buf : Slice.t)
|
|
(fd : Unix.file_descr) : t =
|
|
object
|
|
inherit t_from_output ~bytes:buf.bytes ()
|
|
|
|
method private output_underlying bs i len0 =
|
|
let i = ref i in
|
|
let len = ref len0 in
|
|
while !len > 0 do
|
|
match Unix.write fd bs !i !len with
|
|
| 0 -> failwith "write failed"
|
|
| n ->
|
|
i := !i + n;
|
|
len := !len - n
|
|
| exception
|
|
Unix.Unix_error
|
|
( ( Unix.EBADF | Unix.ENOTCONN | Unix.ESHUTDOWN
|
|
| Unix.ECONNRESET | Unix.EPIPE ),
|
|
_,
|
|
_ ) ->
|
|
failwith "write failed"
|
|
| exception
|
|
Unix.Unix_error
|
|
((Unix.EWOULDBLOCK | Unix.EAGAIN | Unix.EINTR), _, _) ->
|
|
ignore (Unix.select [] [ fd ] [] 1.)
|
|
done
|
|
|
|
method private close_underlying () =
|
|
if not (A.exchange closed true) then
|
|
if close_noerr then (
|
|
try Unix.close fd with _ -> ()
|
|
) else
|
|
Unix.close fd
|
|
end
|
|
|
|
let output_buf (self : t) (buf : Buf.t) : unit =
|
|
let b = Buf.bytes_slice buf in
|
|
output self b 0 (Buf.size buf)
|
|
|
|
(** [chunk_encoding oc] makes a new channel that outputs its content into [oc]
|
|
in chunk encoding form.
|
|
@param close_rec if true, closing the result will also close [oc]
|
|
@param buf a buffer used to accumulate data into chunks.
|
|
Chunks are emitted when [buf]'s size gets over a certain threshold,
|
|
or when [flush] is called.
|
|
*)
|
|
let chunk_encoding ?(buf = Buf.create ()) ~close_rec (oc : #t) : t =
|
|
(* write content of [buf] as a chunk if it's big enough.
|
|
If [force=true] then write content of [buf] if it's simply non empty. *)
|
|
let write_buf ~force () =
|
|
let n = Buf.size buf in
|
|
if (force && n > 0) || n >= 4_096 then (
|
|
output_string oc (Printf.sprintf "%x\r\n" n);
|
|
output oc (Buf.bytes_slice buf) 0 n;
|
|
output_string oc "\r\n";
|
|
Buf.clear buf
|
|
)
|
|
in
|
|
|
|
object
|
|
method flush () =
|
|
write_buf ~force:true ();
|
|
flush oc
|
|
|
|
method close () =
|
|
write_buf ~force:true ();
|
|
(* write an empty chunk to close the stream *)
|
|
output_string oc "0\r\n";
|
|
(* write another crlf after the stream (see #56) *)
|
|
output_string oc "\r\n";
|
|
flush oc;
|
|
if close_rec then close oc
|
|
|
|
method output b i n =
|
|
Buf.add_bytes buf b i n;
|
|
write_buf ~force:false ()
|
|
|
|
method output_char c =
|
|
Buf.add_char buf c;
|
|
write_buf ~force:false ()
|
|
end
|
|
end
|
|
|
|
(** Input channel (byte source) *)
|
|
module Input = struct
|
|
include Iostream.In_buf
|
|
|
|
let of_unix_fd ?(close_noerr = false) ~closed ~(buf : Slice.t)
|
|
(fd : Unix.file_descr) : t =
|
|
let eof = ref false in
|
|
let input buf i len : int =
|
|
let n = ref 0 in
|
|
if not !eof then (
|
|
n := Unix.read fd buf i len;
|
|
if !n = 0 then eof := true
|
|
);
|
|
!n
|
|
in
|
|
|
|
object
|
|
inherit Iostream.In_buf.t_from_refill ~bytes:buf.bytes ()
|
|
|
|
method private refill (slice : Slice.t) =
|
|
if not !eof then (
|
|
slice.off <- 0;
|
|
slice.len <- input slice.bytes 0 (Bytes.length slice.bytes);
|
|
(* Printf.eprintf "read returned %d B\n%!" !n; *)
|
|
if slice.len = 0 then eof := true
|
|
)
|
|
|
|
method close () =
|
|
if not (A.exchange closed true) then (
|
|
eof := true;
|
|
if close_noerr then (
|
|
try Unix.close fd with _ -> ()
|
|
) else
|
|
Unix.close fd
|
|
)
|
|
end
|
|
|
|
let[@inline] of_string s : t = (of_string s :> t)
|
|
|
|
let of_slice (slice : Slice.t) : t =
|
|
object
|
|
inherit Iostream.In_buf.t_from_refill ~bytes:slice.bytes ()
|
|
|
|
method private refill (slice : Slice.t) =
|
|
slice.off <- 0;
|
|
slice.len <- 0
|
|
|
|
method close () = ()
|
|
end
|
|
|
|
(** Read into the given slice.
|
|
@return the number of bytes read, [0] means end of input. *)
|
|
let[@inline] input (self : t) buf i len = self#input buf i len
|
|
|
|
(** Close the channel. *)
|
|
let[@inline] close self : unit = self#close ()
|
|
|
|
(** Read exactly [len] bytes.
|
|
@raise End_of_file if the input did not contain enough data. *)
|
|
let really_input (self : #t) buf i len : unit =
|
|
let i = ref i in
|
|
let len = ref len in
|
|
while !len > 0 do
|
|
let n = input self buf !i !len in
|
|
if n = 0 then raise End_of_file;
|
|
i := !i + n;
|
|
len := !len - n
|
|
done
|
|
|
|
let iter_slice (f : Slice.t -> unit) (self : #t) : unit =
|
|
let continue = ref true in
|
|
while !continue do
|
|
let slice = self#fill_buf () in
|
|
if slice.len = 0 then (
|
|
continue := false;
|
|
close self
|
|
) else (
|
|
f slice;
|
|
Slice.consume slice slice.len
|
|
)
|
|
done
|
|
|
|
let iter f self =
|
|
iter_slice (fun (slice : Slice.t) -> f slice.bytes slice.off slice.len) self
|
|
|
|
let to_chan oc (self : #t) =
|
|
iter_slice
|
|
(fun (slice : Slice.t) ->
|
|
Stdlib.output oc slice.bytes slice.off slice.len)
|
|
self
|
|
|
|
let to_chan' (oc : #Iostream.Out.t) (self : #t) : unit =
|
|
iter_slice
|
|
(fun (slice : Slice.t) ->
|
|
Iostream.Out.output oc slice.bytes slice.off slice.len)
|
|
self
|
|
|
|
(** Output a stream using chunked encoding *)
|
|
let output_chunked' ?buf (oc : #Iostream.Out_buf.t) (self : #t) : unit =
|
|
let oc' = Output.chunk_encoding ?buf oc ~close_rec:false in
|
|
match to_chan' oc' self with
|
|
| () -> Output.close oc'
|
|
| exception e ->
|
|
let bt = Printexc.get_raw_backtrace () in
|
|
Output.close oc';
|
|
Printexc.raise_with_backtrace e bt
|
|
|
|
(** print a stream as a series of chunks *)
|
|
let output_chunked ?buf (oc : out_channel) (self : #t) : unit =
|
|
output_chunked' ?buf (Output.of_out_channel oc) self
|
|
end
|
|
|
|
(** Input channel (byte source) with read-with-timeout *)
|
|
module Input_with_timeout = struct
|
|
include Iostream.In_buf
|
|
|
|
class type t = Iostream.In_buf.t_with_timeout
|
|
|
|
exception Timeout = Iostream.Timeout
|
|
(** Exception for timeouts *)
|
|
|
|
exception Timeout_partial_read of int
|
|
(** Exception for timeouts with a partial read *)
|
|
|
|
(** fill buffer, but stop at the deadline *)
|
|
let fill_buf_with_deadline (self : #t) ~(deadline : float) : Slice.t =
|
|
let timeout = deadline -. Time.now_s () in
|
|
if timeout <= 0. then raise Timeout;
|
|
fill_buf_with_timeout self timeout
|
|
|
|
(** fill buffer, but stop at the deadline if provided *)
|
|
let fill_buf_with_deadline_opt (self : #t) ~(deadline : float option) :
|
|
Slice.t =
|
|
match deadline with
|
|
| None -> fill_buf self
|
|
| Some d -> fill_buf_with_deadline self ~deadline:d
|
|
|
|
let of_unix_fd ?(close_noerr = false) ~closed ~(buf : Slice.t)
|
|
(fd : Unix.file_descr) : t =
|
|
let eof = ref false in
|
|
|
|
let input_with_timeout t buf i len : int =
|
|
let deadline = Time.now_s () +. t in
|
|
let n = ref 0 in
|
|
while
|
|
(not (Atomic.get closed))
|
|
&& (not !eof)
|
|
&&
|
|
try
|
|
n := Unix.read fd buf i len;
|
|
false
|
|
with
|
|
| Unix.Unix_error ((Unix.EAGAIN | Unix.EWOULDBLOCK), _, _) ->
|
|
(* sleep *)
|
|
true
|
|
| Unix.Unix_error ((Unix.ECONNRESET | Unix.ESHUTDOWN | Unix.EPIPE), _, _)
|
|
->
|
|
(* exit *)
|
|
false
|
|
do
|
|
let now = Time.now_s () in
|
|
if now >= deadline then raise Timeout;
|
|
ignore (Unix.select [ fd ] [] [] (deadline -. now) : _ * _ * _)
|
|
done;
|
|
!n
|
|
in
|
|
|
|
object
|
|
inherit Iostream.In_buf.t_with_timeout_from_refill ~bytes:buf.bytes ()
|
|
|
|
method private refill_with_timeout t (slice : Slice.t) =
|
|
if not !eof then (
|
|
slice.off <- 0;
|
|
slice.len <-
|
|
input_with_timeout t slice.bytes 0 (Bytes.length slice.bytes);
|
|
(* Printf.eprintf "read returned %d B\n%!" !n; *)
|
|
if slice.len = 0 then eof := true
|
|
)
|
|
|
|
method close () =
|
|
if not (A.exchange closed true) then (
|
|
eof := true;
|
|
if close_noerr then (
|
|
try Unix.close fd with _ -> ()
|
|
) else
|
|
Unix.close fd
|
|
)
|
|
end
|
|
|
|
let of_slice (slice : Slice.t) : t =
|
|
object
|
|
inherit Iostream.In_buf.t_with_timeout_from_refill ~bytes:slice.bytes ()
|
|
|
|
method private refill_with_timeout _t (slice : Slice.t) =
|
|
slice.off <- 0;
|
|
slice.len <- 0
|
|
|
|
method close () = ()
|
|
end
|
|
|
|
(** Read into the given slice.
|
|
@return the number of bytes read, [0] means end of input. *)
|
|
let[@inline] input (self : t) buf i len = self#input buf i len
|
|
|
|
(** Close the channel. *)
|
|
let[@inline] close self : unit = self#close ()
|
|
|
|
let iter_slice = Input.iter_slice
|
|
let iter = Input.iter
|
|
let to_chan = Input.to_chan
|
|
let to_chan' = Input.to_chan'
|
|
|
|
(** Read the whole body
|
|
@param deadline a deadline before which the operation must complete
|
|
@raise Timeout if deadline expires (leftovers are in [buf] *)
|
|
let read_all_using ~buf ~(deadline : float) (self : #t) : string =
|
|
Buf.clear buf;
|
|
let continue = ref true in
|
|
while !continue do
|
|
let timeout = deadline -. Time.now_s () in
|
|
if timeout <= 0. then raise Timeout;
|
|
let slice = fill_buf_with_timeout self timeout in
|
|
if slice.len = 0 then
|
|
continue := false
|
|
else (
|
|
assert (slice.len > 0);
|
|
Buf.add_bytes buf slice.bytes slice.off slice.len;
|
|
Slice.consume slice slice.len
|
|
)
|
|
done;
|
|
Buf.contents_and_clear buf
|
|
|
|
(** Read [n] bytes from the input into [bytes].
|
|
@raise Timeout_partial_read if timeout occurs before it's done *)
|
|
let read_exactly_ ?(off = 0) ~too_short ~(deadline : float) (self : #t)
|
|
(bytes : bytes) (n : int) : unit =
|
|
assert (Bytes.length bytes >= off + n);
|
|
let offset = ref off in
|
|
while !offset < n do
|
|
let slice =
|
|
try fill_buf_with_deadline self ~deadline
|
|
with Timeout -> raise (Timeout_partial_read (!offset - off))
|
|
in
|
|
let n_read = min slice.len (n - !offset) in
|
|
Bytes.blit slice.bytes slice.off bytes !offset n_read;
|
|
offset := !offset + n_read;
|
|
Slice.consume slice n_read;
|
|
if n_read = 0 then too_short ()
|
|
done
|
|
|
|
let[@inline] really_input (self : #t) ~deadline buf i len =
|
|
read_exactly_ ~off:i ~deadline self buf len ~too_short:(fun () ->
|
|
raise End_of_file)
|
|
|
|
(** read a line into the buffer, after clearing it. *)
|
|
let read_line_into (self : #t) ~(deadline : float) ~buf : unit =
|
|
Buf.clear buf;
|
|
let continue = ref true in
|
|
while !continue do
|
|
let slice = fill_buf_with_deadline self ~deadline in
|
|
if slice.len = 0 then (
|
|
continue := false;
|
|
if Buf.size buf = 0 then raise End_of_file
|
|
);
|
|
let j = ref slice.off in
|
|
while !j < slice.off + slice.len && Bytes.get slice.bytes !j <> '\n' do
|
|
incr j
|
|
done;
|
|
if !j - slice.off < slice.len then (
|
|
assert (Bytes.get slice.bytes !j = '\n');
|
|
(* line without '\n' *)
|
|
Buf.add_bytes buf slice.bytes slice.off (!j - slice.off);
|
|
(* consume line + '\n' *)
|
|
Slice.consume slice (!j - slice.off + 1);
|
|
continue := false
|
|
) else (
|
|
Buf.add_bytes buf slice.bytes slice.off slice.len;
|
|
Slice.consume slice slice.len
|
|
)
|
|
done
|
|
|
|
let read_line_using ~buf ~deadline (self : #t) : string =
|
|
read_line_into self ~deadline ~buf;
|
|
Buf.contents_and_clear buf
|
|
|
|
let read_line_using_opt ~buf ~deadline (self : #t) : string option =
|
|
match read_line_into self ~buf ~deadline with
|
|
| () -> Some (Buf.contents_and_clear buf)
|
|
| exception End_of_file -> None
|
|
|
|
(* helper for making a new input stream that either contains at most [size]
|
|
bytes, or contains exactly [size] bytes. *)
|
|
let reading_exactly_ ~skip_on_close ~close_rec ~size ~bytes (arg : #t) : t =
|
|
let remaining_size = ref size in
|
|
|
|
object
|
|
inherit t_with_timeout_from_refill ~bytes ()
|
|
|
|
method close () =
|
|
if !remaining_size > 0 && skip_on_close then skip arg !remaining_size;
|
|
if close_rec then close arg
|
|
|
|
method private refill_with_timeout t (slice : Slice.t) =
|
|
slice.off <- 0;
|
|
slice.len <- 0;
|
|
if !remaining_size > 0 then (
|
|
let sub = fill_buf_with_timeout arg t in
|
|
let n =
|
|
min !remaining_size (min sub.len (Bytes.length slice.bytes))
|
|
in
|
|
Bytes.blit sub.bytes sub.off slice.bytes 0 n;
|
|
Slice.consume sub n;
|
|
remaining_size := !remaining_size - n;
|
|
slice.len <- n
|
|
)
|
|
end
|
|
|
|
(** new stream with maximum size [max_size].
|
|
@param close_rec if true, closing this will also close the input stream *)
|
|
let limit_size_to ~close_rec ~max_size ~bytes (arg : #t) : t =
|
|
reading_exactly_ ~size:max_size ~skip_on_close:false ~bytes ~close_rec arg
|
|
|
|
(** New stream that consumes exactly [size] bytes from the input.
|
|
If fewer bytes are read before [close] is called, we read and discard
|
|
the remaining quota of bytes before [close] returns.
|
|
@param close_rec if true, closing this will also close the input stream *)
|
|
let reading_exactly ~close_rec ~size ~bytes (arg : t) : t =
|
|
reading_exactly_ ~size ~close_rec ~skip_on_close:true ~bytes arg
|
|
|
|
let read_chunked ~(bytes : bytes) ~fail (ic : #t) : t =
|
|
let first = ref true in
|
|
|
|
(* small buffer to read the chunk sizes *)
|
|
let line_buf = Buf.create ~size:32 () in
|
|
let read_next_chunk_len ~deadline () : int =
|
|
if !first then
|
|
first := false
|
|
else (
|
|
let line = read_line_using ~buf:line_buf ~deadline ic in
|
|
if String.trim line <> "" then
|
|
raise (fail "expected crlf between chunks")
|
|
);
|
|
let line = read_line_using ~buf:line_buf ~deadline ic in
|
|
(* parse chunk length, ignore extensions *)
|
|
let chunk_size =
|
|
if String.trim line = "" then
|
|
0
|
|
else (
|
|
try
|
|
let off = ref 0 in
|
|
let n = Parse_.pos_hex line off in
|
|
n
|
|
with _ ->
|
|
raise (fail (spf "cannot read chunk size from line %S" line))
|
|
)
|
|
in
|
|
chunk_size
|
|
in
|
|
let eof = ref false in
|
|
let chunk_size = ref 0 in
|
|
|
|
object
|
|
inherit t_with_timeout_from_refill ~bytes ()
|
|
|
|
method private refill_with_timeout t (slice : Slice.t) : unit =
|
|
let deadline = Time.now_s () +. t in
|
|
if !chunk_size = 0 && not !eof then (
|
|
chunk_size := read_next_chunk_len ~deadline ();
|
|
if !chunk_size = 0 then eof := true (* stream is finished *)
|
|
);
|
|
slice.off <- 0;
|
|
slice.len <- 0;
|
|
if !chunk_size > 0 then (
|
|
(* read the whole chunk, or [Bytes.length bytes] of it *)
|
|
let to_read = min !chunk_size (Bytes.length slice.bytes) in
|
|
read_exactly_ ~deadline
|
|
~too_short:(fun () -> raise (fail "chunk is too short"))
|
|
ic slice.bytes to_read;
|
|
slice.len <- to_read;
|
|
chunk_size := !chunk_size - to_read
|
|
)
|
|
|
|
method close () = eof := true (* do not close underlying stream *)
|
|
end
|
|
|
|
let output_chunked = Input.output_chunked
|
|
let output_chunked' = Input.output_chunked'
|
|
end
|
|
|
|
(** A writer abstraction. *)
|
|
module Writer = struct
|
|
type t = { write: Output.t -> unit } [@@unboxed]
|
|
(** Writer.
|
|
|
|
A writer is a push-based stream of bytes.
|
|
Give it an output channel and it will write the bytes in it.
|
|
|
|
This is useful for responses: an http endpoint can return a writer
|
|
as its response's body; the writer is given access to the connection
|
|
to the client and can write into it as if it were a regular
|
|
[out_channel], including controlling calls to [flush].
|
|
Tiny_httpd will convert these writes into valid HTTP chunks.
|
|
@since 0.14
|
|
*)
|
|
|
|
let[@inline] make ~write () : t = { write }
|
|
|
|
(** Write into the channel. *)
|
|
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 = Iostream.Out.output_string oc str in
|
|
{ write }
|
|
|
|
let[@inline] of_input (ic : #Input.t) : t =
|
|
{ write = (fun oc -> Input.to_chan' oc ic) }
|
|
end
|
|
|
|
(** A TCP server abstraction. *)
|
|
module TCP_server = struct
|
|
type conn_handler = {
|
|
handle:
|
|
client_addr:Unix.sockaddr -> Input_with_timeout.t -> Output.t -> unit;
|
|
(** Handle client connection *)
|
|
}
|
|
|
|
type t = {
|
|
endpoint: unit -> string * int;
|
|
(** Endpoint we listen on. This can only be called from within [serve]. *)
|
|
active_connections: unit -> int;
|
|
(** Number of connections currently active *)
|
|
running: unit -> bool; (** Is the server currently running? *)
|
|
stop: unit -> unit;
|
|
(** Ask the server to stop. This might not take effect immediately,
|
|
and is idempotent. After this [server.running()] must return [false]. *)
|
|
}
|
|
(** A running TCP server.
|
|
|
|
This contains some functions that provide information about the running
|
|
server, including whether it's active (as opposed to stopped), a function
|
|
to stop it, and statistics about the number of connections. *)
|
|
|
|
type builder = {
|
|
serve: after_init:(t -> unit) -> handle:conn_handler -> unit -> unit;
|
|
(** Blocking call to listen for incoming connections and handle them.
|
|
Uses the connection handler [handle] to handle individual client
|
|
connections in individual threads/fibers/tasks.
|
|
@param after_init is called once with the server after the server
|
|
has started. *)
|
|
}
|
|
(** A TCP server builder implementation.
|
|
|
|
Calling [builder.serve ~after_init ~handle ()] starts a new TCP server on
|
|
an unspecified endpoint
|
|
(most likely coming from the function returning this builder)
|
|
and returns the running server. *)
|
|
end
|