Source file Tiny_httpd.ml

type byte_stream = {
  bs_fill_buf: unit -> (bytes * int * int);
  bs_consume: int -> unit;
  bs_close: unit -> unit;
}
(** A buffer input stream, with a view into the current buffer (or refill if empty),
    and a function to consume [n] bytes *)

let _debug_on = ref (
  match String.trim @@ Sys.getenv "HTTP_DBG" with
  | "" -> false | _ -> true | exception _ -> false
)
let _enable_debug b = _debug_on := b
let _debug k =
  if !_debug_on then (
    k (fun fmt->
       Printf.fprintf stdout "[http.thread %d]: " Thread.(id @@ self());
       Printf.kfprintf (fun oc -> Printf.fprintf oc "\n%!") stdout fmt)
  )

module Buf_ = struct
  type t = {
    mutable bytes: bytes;
    mutable i: int;
  }

  let create ?(size=4_096) () : t =
    { bytes=Bytes.make size ' '; i=0 }

  let size self = self.i
  let clear self : unit =
    if Bytes.length self.bytes > 4_096 * 1_024 then (
      self.bytes <- Bytes.make 4096 ' '; (* free big buffer *)
    );
    self.i <- 0

  let resize self new_size : unit =
    let new_buf = Bytes.make new_size ' ' in
    Bytes.blit self.bytes 0 new_buf 0 self.i;
    self.bytes <- new_buf

  let add_bytes (self:t) s i len : unit =
    if self.i + len >= Bytes.length self.bytes then (
      resize self (self.i + self.i / 2 + len + 10);
    );
    Bytes.blit s i self.bytes self.i len;
    self.i <- self.i + len

  let contents (self:t) : string = Bytes.sub_string self.bytes 0 self.i

  let contents_and_clear (self:t) : string =
    let x = contents self in
    clear self;
    x
end

module Byte_stream = struct
  type t = byte_stream

  let close self = self.bs_close()

  let empty = {
    bs_fill_buf=(fun () -> Bytes.empty, 0, 0);
    bs_consume=(fun _ -> ());
    bs_close=(fun () -> ());
  }

  let of_chan_ ~close ic : t =
    let i = ref 0 in
    let len = ref 0 in
    let buf = Bytes.make 4096 ' ' in
    { bs_fill_buf=(fun () ->
      if !i >= !len then (
        i := 0;
        len := input ic buf 0 (Bytes.length buf);
      );
      buf, !i,!len - !i);
      bs_consume=(fun n -> i := !i + n);
      bs_close=(fun () -> close ic)
    }

  let of_chan = of_chan_ ~close:close_in
  let of_chan_close_noerr = of_chan_ ~close:close_in_noerr

  let rec iter f (self:t) : unit =
    let s, i, len = self.bs_fill_buf () in
    if len=0 then (
      self.bs_close();
    ) else (
      f s i len;
      self.bs_consume len;
      (iter [@tailcall]) f self
    )

  let to_chan (oc:out_channel) (self:t) =
    iter (fun s i len -> output oc s i len) self

  let of_bytes ?(i=0) ?len s : t =
    (* invariant: !i+!len is constant *)
    let len =
      ref (
        match len with
        | Some n ->
          if n > Bytes.length s -i then invalid_arg "Byte_stream.of_bytes";
          n
        | None -> Bytes.length s - i
      )
    in
    let i = ref i in
    { bs_fill_buf=(fun () -> s, !i, !len);
      bs_close=(fun () -> len := 0);
      bs_consume=(fun n -> assert (n<= !len); i := !i + n; len := !len - n);
    }

  let of_string s : t =
    of_bytes (Bytes.unsafe_of_string s)

  let with_file file f =
    let ic = open_in file in
    try
      let x = f (of_chan ic) in
      close_in ic;
      x
    with e ->
      close_in_noerr ic;
      raise e

  let read_all ?(buf=Buf_.create()) (self:t) : string =
    let continue = ref true in
    while !continue do
      let s, i, len = self.bs_fill_buf () in
      _debug (fun k->k "read-all: got i=%d, len=%d, bufsize %d" i len (Buf_.size buf));
      if len > 0 then (
        Buf_.add_bytes buf s i len;
        self.bs_consume len;
      );
      assert (len >= 0);
      if len = 0 then (
        continue := false
      )
    done;
    Buf_.contents_and_clear buf

  (* put [n] bytes from the input into bytes *)
  let read_exactly_ ~too_short (self:t) (bytes:bytes) (n:int) : unit =
    assert (Bytes.length bytes >= n);
    let offset = ref 0 in
    while !offset < n do
      let s, i, len = self.bs_fill_buf () in
      let n_read = min len (n- !offset) in
      Bytes.blit s i bytes !offset n_read;
      offset := !offset + n_read;
      self.bs_consume n_read;
      if n_read=0 then too_short();
    done

  (* read a line into the buffer, after clearing it. *)
  let read_line_into (self:t) ~buf : unit =
    Buf_.clear buf;
    let continue = ref true in
    while !continue do
      let s, i, len = self.bs_fill_buf () in
      if len=0 then (
        continue := false;
        if Buf_.size buf = 0 then raise End_of_file;
      );
      let j = ref i in
      while !j < i+len && Bytes.get s !j <> '\n' do
        incr j
      done;
      if !j-i < len then (
        assert (Bytes.get s !j = '\n');
        Buf_.add_bytes buf s i (!j-i); (* without \n *)
        self.bs_consume (!j-i+1); (* remove \n *)
        continue := false
      ) else (
        Buf_.add_bytes buf s i len;
        self.bs_consume len;
      )
    done

  (* new stream with maximum size [max_size].
     @param close_rec if true, closing this will also close the input stream
     @param too_big called with read size if the max size is reached *)
  let limit_size_to ~close_rec ~max_size ~too_big (self:t) : t =
    let size = ref 0 in
    let continue = ref true in
    { bs_fill_buf =
        (fun () ->
          if !continue then self.bs_fill_buf() else Bytes.empty, 0, 0);
      bs_close=(fun () ->
          if close_rec then self.bs_close ());
      bs_consume = (fun n ->
          size := !size + n;
          if !size > max_size then (
            continue := false;
            too_big !size
          ) else (
            self.bs_consume n
          ));
    }

  (* read exactly [size] bytes from the stream *)
  let read_exactly ~close_rec ~size ~too_short (self:t) : t =
    if size=0 then (
      empty
    ) else (
      let size = ref size in
      { bs_fill_buf = (fun () ->
            (* must not block on [self] if we're done *)
            if !size = 0 then Bytes.empty, 0, 0
            else (
              let buf, i, len = self.bs_fill_buf () in
              let len = min len !size in
              if len = 0 && !size > 0 then (
                too_short !size;
              );
              buf, i, len
            )
          );
        bs_close=(fun () ->
            (* close underlying stream if [close_rec] *)
            if close_rec then self.bs_close();
            size := 0);
        bs_consume = (fun n ->
            let n = min n !size in
            size := !size - n;
            self.bs_consume n);
      }
    )

  let read_line ?(buf=Buf_.create()) self : string =
    read_line_into self ~buf;
    Buf_.contents buf
end

exception Bad_req of int * string
let bad_reqf c fmt = Printf.ksprintf (fun s ->raise (Bad_req (c,s))) fmt

module Response_code = struct
  type t = int

  let ok = 200
  let not_found = 404
  let descr = function
    | 100 -> "Continue"
    | 200 -> "OK"
    | 201 -> "Created"
    | 202 -> "Accepted"
    | 204 -> "No content"
    | 300 -> "Multiple choices"
    | 301 -> "Moved permanently"
    | 302 -> "Found"
    | 304 -> "Not Modified"
    | 400 -> "Bad request"
    | 403 -> "Forbidden"
    | 404 -> "Not found"
    | 405 -> "Method not allowed"
    | 408 -> "Request timeout"
    | 409 -> "Conflict"
    | 410 -> "Gone"
    | 411 -> "Length required"
    | 413 -> "Payload too large"
    | 417 -> "Expectation failed"
    | 500 -> "Internal server error"
    | 501 -> "Not implemented"
    | 503 -> "Service unavailable"
    | n -> "Unknown response code " ^ string_of_int n (* TODO *)
end

type 'a resp_result = ('a, Response_code.t * string) result
let unwrap_resp_result = function
  | Ok x -> x
  | Error (c,s) -> raise (Bad_req (c,s))

module Meth = struct
  type t = [
    | `GET
    | `PUT
    | `POST
    | `HEAD
    | `DELETE
  ]

  let to_string = function
    | `GET -> "GET"
    | `PUT -> "PUT"
    | `HEAD -> "HEAD"
    | `POST -> "POST"
    | `DELETE -> "DELETE"
  let pp out s = Format.pp_print_string out (to_string s)

  let of_string = function
    | "GET" -> `GET
    | "PUT" -> `PUT
    | "POST" -> `POST
    | "HEAD" -> `HEAD
    | "DELETE" -> `DELETE
    | s -> bad_reqf 400 "unknown method %S" s
end

module Headers = struct
  type t = (string * string) list
  let contains name headers =
    let name' = String.lowercase_ascii name in
    List.exists (fun (n, _) -> name'=n) headers
  let get_exn ?(f=fun x->x) x h =
    let x' = String.lowercase_ascii x in
    List.assoc x' h |> f
  let get ?(f=fun x -> x) x h =
    try Some (get_exn ~f x h) with Not_found -> None
  let remove x h =
    let x' = String.lowercase_ascii x in
    List.filter (fun (k,_) -> k<>x') h
  let set x y h =
    let x' = String.lowercase_ascii x in
    (x',y) :: List.filter (fun (k,_) -> k<>x') h
  let pp out l =
    let pp_pair out (k,v) = Format.fprintf out "@[<h>%s: %s@]" k v in
    Format.fprintf out "@[<v>%a@]" (Format.pp_print_list pp_pair) l

  let parse_ ~buf (bs:byte_stream) : t =
    let rec loop acc =
      let line = Byte_stream.read_line ~buf bs in
      _debug (fun k->k  "parsed header line %S" line);
      if line = "\r" then (
        acc
      ) else (
        let k,v =
          try Scanf.sscanf line "%s@: %s@\r" (fun k v->k,v)
          with _ -> bad_reqf 400 "invalid header line: %S" line
        in
        loop ((String.lowercase_ascii k,v)::acc)
      )
    in
    loop []
end

module Request = struct
  type 'body t = {
    meth: Meth.t;
    host: string;
    headers: Headers.t;
    path: string;
    body: 'body;
  }

  let headers self = self.headers
  let host self = self.host
  let meth self = self.meth
  let path self = self.path
  let body self = self.body

  let query self =
    match Tiny_httpd_util.(parse_query @@ get_query self.path) with
    | Ok l -> l
    | Error e -> bad_reqf 400 "invalid query: %s" e

  let get_header ?f self h = Headers.get ?f h self.headers
  let get_header_int self h = match get_header self h with
    | Some x -> (try Some (int_of_string x) with _ -> None)
    | None -> None
  let set_header self k v = {self with headers=Headers.set k v self.headers}

  let pp_ out self : unit =
    Format.fprintf out "{@[meth=%s;@ host=%s;@ headers=%a;@ path=%S;@ body=?@]}"
      (Meth.to_string self.meth) self.host Headers.pp self.headers self.path
  let pp out self : unit =
    Format.fprintf out "{@[meth=%s;@ host=%s;@ headers=%a;@ path=%S;@ body=%S@]}"
      (Meth.to_string self.meth) self.host Headers.pp self.headers
      self.path self.body

  (* decode a "chunked" stream into a normal stream *)
  let read_stream_chunked_ ?(buf=Buf_.create()) (bs:byte_stream) : byte_stream =
    _debug (fun k->k "body: start reading chunked stream...");
    let first = ref true in
    let read_next_chunk_len () : int =
      if !first then (
        first := false
      ) else (
        let line = Byte_stream.read_line ~buf bs in
        if String.trim line <> "" then bad_reqf 400 "expected crlf between chunks";
      );
      let line = Byte_stream.read_line ~buf bs in
      (* parse chunk length, ignore extensions *)
      let chunk_size = (
        if String.trim line = "" then 0
        else
          try Scanf.sscanf line "%x %s@\r" (fun n _ext -> n)
          with _ -> bad_reqf 400 "cannot read chunk size from line %S" line
      ) in
      chunk_size
    in
    let refill = ref true in
    let bytes = Bytes.make (16 * 4096)  ' ' in (* internal buffer, 16kb *)
    let offset = ref 0 in
    let len = ref 0 in
    let chunk_size = ref 0 in
    { bs_fill_buf=
        (fun () ->
           (* do we need to refill? *)
           if !offset >= !len then (
             if !chunk_size = 0 && !refill then (
               chunk_size := read_next_chunk_len();
               (* _debug (fun k->k"read next chunk of size %d" !chunk_size); *)
             );
             offset := 0;
             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 bytes) in
               Byte_stream.read_exactly_
                 ~too_short:(fun () -> bad_reqf 400 "chunk is too short")
                 bs bytes to_read;
               len := to_read;
               chunk_size := !chunk_size - to_read;
             ) else (
               refill := false; (* stream is finished *)
             )
           );
           bytes, !offset, !len
        );
      bs_consume=(fun n -> offset := !offset + n);
      bs_close=(fun () ->
          (* close this overlay, do not close underlying stream *)
          len := 0; refill:= false);
    }

  let limit_body_size_ ~max_size (bs:byte_stream) : byte_stream =
    _debug (fun k->k "limit size of body to max-size=%d" max_size);
    Byte_stream.limit_size_to ~max_size ~close_rec:false bs
      ~too_big:(fun size ->
          (* read too much *)
          bad_reqf 413
            "body size was supposed to be %d, but at least %d bytes received"
            max_size size
        )

  let limit_body_size ~max_size (req:byte_stream t) : byte_stream t =
    { req with body=limit_body_size_ ~max_size req.body }

  (* read exactly [size] bytes from the stream *)
  let read_exactly ~size (bs:byte_stream) : byte_stream =
    _debug (fun k->k "body: must read exactly %d bytes" size);
    Byte_stream.read_exactly bs ~close_rec:false
      ~size ~too_short:(fun size ->
          bad_reqf 400 "body is too short by %d bytes" size
        )

  (* parse request, but not body (yet) *)
  let parse_req_start ~buf (bs:byte_stream) : unit t option resp_result =
    try
      let line = Byte_stream.read_line ~buf bs in
      let meth, path =
        try Scanf.sscanf line "%s %s HTTP/1.1\r" (fun x y->x,y)
        with _ -> raise (Bad_req (400, "Invalid request line"))
      in
      let meth = Meth.of_string meth in
      _debug (fun k->k "got meth: %s, path %S" (Meth.to_string meth) path);
      let headers = Headers.parse_ ~buf bs in
      let host =
        match Headers.get "Host" headers with
        | None -> bad_reqf 400 "No 'Host' header in request"
        | Some h -> h
      in
      Ok (Some {meth; host; path; headers; body=()})
    with
    | End_of_file | Sys_error _ -> Ok None
    | Bad_req (c,s) -> Error (c,s)
    | e ->
      Error (400, Printexc.to_string e)

  (* parse body, given the headers.
     @param tr_stream a transformation of the input stream. *)
  let parse_body_ ~tr_stream ~buf (req:byte_stream t) : byte_stream t resp_result =
    try
      let size =
        match Headers.get_exn "Content-Length" req.headers |> int_of_string with
        | n -> n (* body of fixed size *)
        | exception Not_found -> 0
        | exception _ -> bad_reqf 400 "invalid content-length"
      in
      let body =
        match get_header ~f:String.trim req "Transfer-Encoding" with
        | None -> read_exactly ~size @@ tr_stream req.body
        | Some "chunked" ->
          let bs =
            read_stream_chunked_ ~buf @@ tr_stream req.body (* body sent by chunks *)
          in
          if size>0 then limit_body_size_ ~max_size:size bs else bs
        | Some s -> bad_reqf 500 "cannot handle transfer encoding: %s" s
      in
      Ok {req with body}
    with
    | End_of_file -> Error (400, "unexpected end of file")
    | Bad_req (c,s) -> Error (c,s)
    | e ->
      Error (400, Printexc.to_string e)

  let read_body_full (self:byte_stream t) : string t =
    try
      let body = Byte_stream.read_all self.body in
      { self with body }
    with
    | Bad_req _ as e -> raise e
    | e -> bad_reqf 500 "failed to read body: %s" (Printexc.to_string e)

  module Internal_ = struct
    let parse_req_start ?(buf=Buf_.create()) bs =
      parse_req_start ~buf bs |> unwrap_resp_result

    let parse_body ?(buf=Buf_.create()) req bs : _ t =
      parse_body_ ~tr_stream:(fun s->s) ~buf {req with body=bs} |> unwrap_resp_result
  end
end

(*$R
  let q = "GET hello HTTP/1.1\r\nHost: coucou\r\nContent-Length: 11\r\n\r\nsalutationsSOMEJUNK" in
  let str = Byte_stream.of_string q in
  let r = Request.Internal_.parse_req_start str in
  match r with
  | None -> assert_failure "should parse"
  | Some req ->
    assert_equal (Some "coucou") (Headers.get "Host" req.Request.headers);
    assert_equal (Some "coucou") (Headers.get "host" req.Request.headers);
    assert_equal (Some "11") (Headers.get "content-length" req.Request.headers);
    assert_equal "hello" req.Request.path;
    let req = Request.Internal_.parse_body req str |> Request.read_body_full in
    assert_equal ~printer:(fun s->s) "salutations" req.Request.body;
    ()
*)

module Response = struct
  type body = [`String of string | `Stream of byte_stream]
  type t = {
    code: Response_code.t;
    headers: Headers.t;
    body: body;
  }

  let make_raw ?(headers=[]) ~code body : t =
    (* add content length to response *)
    let headers =
      Headers.set "Content-Length" (string_of_int (String.length body)) headers
    in
    { code; headers; body=`String body; }

  let make_raw_stream ?(headers=[]) ~code body : t =
    (* add content length to response *)
    let headers = Headers.set "Transfer-Encoding" "chunked" headers in
    { code; headers; body=`Stream body; }

  let make_string ?headers r = match r with
    | Ok body -> make_raw ?headers ~code:200 body
    | Error (code,msg) -> make_raw ?headers ~code msg

  let make_stream ?headers r = match r with
    | Ok body -> make_raw_stream ?headers ~code:200 body
    | Error (code,msg) -> make_raw ?headers ~code msg

  let make ?headers r : t = match r with
    | Ok (`String body) -> make_raw ?headers ~code:200 body
    | Ok (`Stream body) -> make_raw_stream ?headers ~code:200 body
    | Error (code,msg) -> make_raw ?headers ~code msg

  let fail ?headers ~code fmt =
    Printf.ksprintf (fun msg -> make_raw ?headers ~code msg) fmt
  let fail_raise ~code fmt =
    Printf.ksprintf (fun msg -> raise (Bad_req (code,msg))) fmt

  let pp out self : unit =
    let pp_body out = function
      | `String s -> Format.fprintf out "%S" s
      | `Stream _ -> Format.pp_print_string out "<stream>"
    in
    Format.fprintf out "{@[code=%d;@ headers=%a;@ body=%a@]}"
      self.code Headers.pp self.headers pp_body self.body

  (* print a stream as a series of chunks *)
  let output_stream_chunked_ (oc:out_channel) (str:byte_stream) : unit =
    let continue = ref true in
    while !continue do
      (* next chunk *)
      let s, i, len = str.bs_fill_buf () in
      Printf.fprintf oc "%x\r\n" len;
      output oc s i len;
      str.bs_consume len;
      if len = 0 then (
        continue := false;
      );
      output_string oc "\r\n";
    done;
    ()

  let output_ (oc:out_channel) (self:t) : unit =
    Printf.fprintf oc "HTTP/1.1 %d %s\r\n" self.code (Response_code.descr self.code);
    List.iter (fun (k,v) -> Printf.fprintf oc "%s: %s\r\n" k v) self.headers;
    output_string oc "\r\n";
    begin match self.body with
      | `String "" -> ()
      | `String s -> output_string oc s;
      | `Stream str -> output_stream_chunked_ oc str;
    end;
    flush oc
end

(* semaphore, for limiting concurrency. *)
module Sem_ = struct
  type t = {
    mutable n : int;
    mutex : Mutex.t;
    cond : Condition.t;
  }

  let create n =
    if n <= 0 then invalid_arg "Semaphore.create";
    { n; mutex=Mutex.create(); cond=Condition.create(); }

  let acquire m t =
    Mutex.lock t.mutex;
    while t.n < m do
      Condition.wait t.cond t.mutex;
    done;
    assert (t.n >= m);
    t.n <- t.n - m;
    Condition.broadcast t.cond;
    Mutex.unlock t.mutex

  let release m t =
    Mutex.lock t.mutex;
    t.n <- t.n + m;
    Condition.broadcast t.cond;
    Mutex.unlock t.mutex
end

type cb_path_handler = byte_stream Request.t -> Response.t

type t = {
  addr: string;
  port: int;
  sem_max_connections: Sem_.t;
  new_thread: (unit -> unit) -> unit;
  masksigpipe: bool;
  mutable handler: (string Request.t -> Response.t);
  mutable path_handlers : (unit Request.t -> cb_path_handler resp_result option) list;
  mutable cb_decode_req:
    (unit Request.t -> (unit Request.t * (byte_stream -> byte_stream)) option) list;
  mutable cb_encode_resp: (unit Request.t -> Response.t -> Response.t option) list;
  mutable running: bool;
}

let addr self = self.addr
let port self = self.port

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

let add_path_handler_
    ?(accept=fun _req -> Ok ())
    ?meth ~tr_req self fmt f =
  let ph req: cb_path_handler resp_result option =
    match meth with
    | Some m when m <> req.Request.meth -> None (* ignore *)
    | _ ->
      begin match Scanf.sscanf req.Request.path fmt f with
        | handler ->
          (* we have a handler, do we accept the request based on its headers? *)
          begin match accept req with
            | Ok () -> Some (Ok (fun req -> handler @@ tr_req req))
            | Error _ as e -> Some e
          end
        | exception _ ->
          None (* path didn't match *)
      end
  in
  self.path_handlers <- ph :: self.path_handlers

let add_path_handler ?accept ?meth self fmt f=
  add_path_handler_ ?accept ?meth ~tr_req:Request.read_body_full self fmt f

let add_path_handler_stream ?accept ?meth self fmt f=
  add_path_handler_ ?accept ?meth ~tr_req:(fun x->x) self fmt f

let create
    ?(masksigpipe=true)
    ?(max_connections=32)
    ?(new_thread=(fun f -> ignore (Thread.create f () : Thread.t)))
    ?(addr="127.0.0.1") ?(port=8080) () : t =
  let handler _req = Response.fail ~code:404 "no top handler" in
  let max_connections = max 4 max_connections in
  { new_thread; addr; port; masksigpipe; handler;
    running= true; sem_max_connections=Sem_.create max_connections;
    path_handlers=[];
    cb_encode_resp=[]; cb_decode_req=[];
  }

let stop s = s.running <- false

let find_map f l =
  let rec aux f = function
    | [] -> None
    | x::l' ->
      match f x with
        | Some _ as res -> res
        | None -> aux f l'
  in aux f l

let handle_client_ (self:t) (client_sock:Unix.file_descr) : unit =
  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_chan ic in
  let continue = ref true in
  while !continue && self.running do
    _debug (fun k->k "read next request");
    match Request.parse_req_start ~buf is with
    | Ok None -> continue := false
    | Error (c,s) ->
      let res = Response.make_raw ~code:c s in
      begin
        try Response.output_ oc res
        with Sys_error _ -> ()
      end;
      continue := false
    | Ok (Some req) ->
      let res =
        try
          (* is there a handler for this path? *)
          let handler =
            match find_map (fun ph -> ph req) self.path_handlers with
            | Some f -> unwrap_resp_result f
            | None -> (fun req -> self.handler @@ Request.read_body_full req)
          in
          (* handle expectations *)
          begin match Request.get_header ~f:String.trim req "Expect" with
            | Some "100-continue" ->
              _debug (fun k->k "send back: 100 CONTINUE");
              Response.output_ oc (Response.make_raw ~code:100 "");
            | Some s -> bad_reqf 417 "unknown expectation %s" s
            | None -> ()
          end;
          (* preprocess request's input stream *)
          let req0, tr_stream =
            List.fold_left
              (fun (req,tr) cb ->
                 match cb req with
                 | None -> req, tr
                 | Some (r',f) -> r', (fun is -> tr is |> f))
              (req, (fun is->is)) self.cb_decode_req
          in
          (* now actually read request's body *)
          let req =
            Request.parse_body_ ~tr_stream ~buf {req0 with body=is}
            |> unwrap_resp_result
          in
          let resp = handler req in
          (* post-process response *)
          List.fold_left
            (fun resp cb -> match cb req0 resp with None -> resp | Some r' -> r')
            resp self.cb_encode_resp
        with
        | Bad_req (code,s) ->
          continue := false;
          Response.make_raw ~code s
        | e ->
          Response.fail ~code:500 "server error: %s" (Printexc.to_string e)
      in
      begin
        try Response.output_ oc res
        with Sys_error _ -> continue := false
      end
    | exception Bad_req (code,s) ->
      Response.output_ oc (Response.make_raw ~code s);
      continue := false
    | exception Sys_error _ ->
      continue := false; (* connection broken somehow *)
  done;
  _debug (fun k->k "done with client, exiting");
  (try Unix.close client_sock
   with e -> _debug (fun k->k "error when closing sock: %s" (Printexc.to_string e)));
  ()

let is_ipv6 self = String.contains self.addr ':'

let run (self:t) : (unit,_) result =
  try
    if self.masksigpipe then (
      ignore (Unix.sigprocmask Unix.SIG_BLOCK [Sys.sigpipe] : _ list);
    );
    let sock =
      Unix.socket (if is_ipv6 self then Unix.PF_INET6 else Unix.PF_INET)
        Unix.SOCK_STREAM 0
    in
    Unix.clear_nonblock sock;
    Unix.setsockopt sock Unix.SO_REUSEADDR true;
    Unix.setsockopt_optint sock Unix.SO_LINGER None;
    let inet_addr = Unix.inet_addr_of_string self.addr in
    Unix.bind sock (Unix.ADDR_INET (inet_addr, self.port));
    Unix.listen sock (2 * self.sem_max_connections.Sem_.n);
    while self.running do
      (* limit concurrency *)
      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;
           with e ->
             (try Unix.close client_sock with _ -> ());
             Sem_.release 1 self.sem_max_connections;
             raise e
        );
    done;
    Ok ()
  with e -> Error e