Amall_http_service.Service

(IO : IO_Type)

  (I : It_type.IT with type 'a It_IO.m = 'a IO.m

    and type It_IO.input_channel = IO.input_channel) :

  sig

    module H : module type of Amall_http.Make(IO)(I)

      

    module HS : module type of Http_server(IO)(I)

      

    (* Service receives path components relative to "mount point".

       For example, when service is registered on

       "http://host:port/a/b/c"

       ( = [""; "http://host:port"; "a"; "b"; "c"]) and when



       request uri is:                 segpath is:

       http://host:port/a/b/c          [""]

       http://host:port/a/b/c/         [""]

       http://host:port/a/b/c/d/e      ["d"; "e"]

       http://host:port/a/b/c/d?e#f    ["d"]

     *)

    (* Abstract type.  Will be used to control listening and

       accepting connections. *)

    type listener

    

    type 's mount_point

    

    open Amall_http

      

    (* where to install service/fallback. *)

    type 's endpoint =

      (('s mount_point) *

       [ | `Service of (segpath * seg) | `Fallback of segpath ])

    

    (* returns listener and mount point for its root *)

    val listener_create :

      Amall_http_server.listen_addr ->

        (listener * (H.http_service_func mount_point) *

         (H.websocket_service_func mount_point))

      

    (* IO value for waiting listener *)

    val io_listener_run : listener -> IO.server

      

    (* really runs listening/accepting *)

    val listener_run : listener -> unit

      

    val listener_stop : listener -> unit

      

    (* creates a destination mount point in the specified segpath

       below the source mount point. *)

    val mount_rel_http :

      H.http_service_func mount_point ->

        segpath -> H.http_service_func mount_point

      

    val mount_rel_ws :

      H.websocket_service_func mount_point ->

        segpath -> H.websocket_service_func mount_point

      

    val mount_http :

      H.http_service_func endpoint -> H.http_service_func -> unit

      

    (* receives websocket messages *)

    type websocket_service_func_worker = H.websocket_service_func_worker

    

    (* returns websocket_service_func_worker, run once per connection *)

    type websocket_service_func = H.websocket_service_func

    

    val mount_websocket :

      H.websocket_service_func endpoint -> websocket_service_func -> unit

      

    (*

  todo:

    value umount : endpoint -> unit

    ;

*)

    val it_post_vars : (char, (string * string) list) I.iteratee

      

  end =

  struct

    open Uri_type

      

    open Amall_http

      

    module H = Amall_http.Make(IO)(I)

      

    module HS = Http_server(IO)(I)

      

    type websocket_service_func = H.websocket_service_func

    

    type websocket_service_func_worker = H.websocket_service_func_worker

    

    let it_post_vars = HS.it_post_vars

      

    let default_http_fallback _segpath rq = (* todo: escape chars for html *)

      let uri = rq.rq_uri in

      let au_opt = uri.authority in

      let (host, port_opt) =

        match au_opt with

        | None -> ("", "")

        | Some au ->

            ((au.host),

             (match au.port with | None -> "" | Some x -> sprintf ":%i" x)) in

      let txt =

        sprintf

          "Can't find path \"%s\" on server \"%s%s\". URI dump follows:<br/>%s"

          uri.path host port_opt (Uri.dump_uri uri)

      in

        I.return

          {

            rs_status_code = 404;

            rs_reason_phrase = "Not found";

            rs_headers = { rs_all = []; };

            rs_body =

              Body_string

                ("<html><body><p>" ^ (txt ^ "</p></body></html>\n"));

          }

      

    let default_ws_fallback =

      Partapp3.make

        (fun _segpath _rq _ws_out_socket ->

           failwith "websocket service not found")

      

    type 's disp_level =

      { seg_map : 's seg_map; mutable fallback : 's

      }

      and 's disp_handler =

      | Level of 's disp_level | Service of 's

      and 's seg_map =

      (seg, 's disp_handler) Timp.map_rws

    

    type listener =

      { io_server_lazy : IO.server Lazy.t;

        root_http_disp_level : H.http_service_func disp_level;

        (* contains just '"" => root' binding when there exists

             at least one service, empty otherwise; and

             fallback with error "no such proto://host:port".

           *)

        root_ws_disp_level : websocket_service_func disp_level

      }

    

    type 's mount_point = 's disp_level

    

    module Tr = Simp.Tree(String)

      

    class ['s] c_seg_map = ['s disp_handler] Tr.map_rws_tree ()

      

    let segpath_of_uri uri =

      let host_port_txt =

        match uri.authority with

        | None -> ""

        | Some au ->

            let port_txt =

              (match au.port with | None -> "" | Some i -> string_of_int i)

            in sprintf "%s:%s" au.host port_txt in

      let pre = host_port_txt in

      let path = Uri.normseg_of_uri uri

      in

        (* из http://host:port/a/b/c?d#e делать

         [""; "host:port"; "a"; "b"; "c"],

         первая пустая строка -- чтобы можно было

         сделать сервис, слушающий [""], то есть,

         все протоколы-хосты-порты.

       *)

        "" :: pre :: path

      

    let rec try_find_handler disp_level segpath :

      [

        | `Ok of (segpath * 's)

        | `Segpath_finished of 's disp_level

        | `Seg_not_found of (seg * segpath * ('s disp_level))

      ] =

      match segpath with

      | [] -> `Segpath_finished disp_level

      | seg :: segs ->

          let () = dbg "S.try_find_handler: %S" seg

          in

            (match disp_level.seg_map#get_opt seg with

             | None ->

                 let () = dbg "S.try_find_handler: .. not found"

                 in `Seg_not_found ((seg, segs, disp_level))

             | Some (Level disp_level) -> try_find_handler disp_level segs

             | Some (Service f) -> `Ok ((segs, f)))

      

    let find_handler (type s) (disp_level : s disp_level) segpath

      (wrap : s -> H.service_desc) : (segpath * H.service_desc) =

      match try_find_handler disp_level segpath with

      | `Ok ((p, f)) -> (p, (wrap f))

      | `Segpath_finished dl -> ([], (wrap dl.fallback))

      | `Seg_not_found ((seg, segs, disp_level)) ->

          ((seg :: segs), (wrap disp_level.fallback))

      

    exception Already_handled

      

    let go_or_create_smallstep ~fb ~disp ~seg : 's disp_level =

      let () = dbg "S.go_or_create_smallstep: seg=%S" seg

      in

        match disp.seg_map#get_opt seg with

        | None ->

            let lev = { seg_map = new c_seg_map; fallback = fb; }

            in

              (disp.seg_map#replace seg (Level lev);

               let () = dbg "S.go_or_create_smallstep: .. created." in lev)

        | Some (Level lev) ->

            let () = dbg "S.go_or_create_smallstep: .. found." in lev

        | Some (Service _f) -> raise Already_handled

      

    let rec go_or_create_bigstep ~fb ~disp ~segs : 's disp_level =

      match segs with

      | [] -> disp

      | seg :: segs ->

          go_or_create_bigstep ~fb

            ~disp: (go_or_create_smallstep ~fb ~disp ~seg) ~segs

      

    type 's endpoint =

      (('s mount_point) *

       [ | `Service of (segpath * seg) | `Fallback of segpath ])

    

    let mount_rel ~fb mount_point segpath =

      go_or_create_bigstep ~disp: mount_point ~fb ~segs: segpath

      

    let mount_rel_http x = mount_rel ~fb: default_http_fallback x

    and mount_rel_ws x = mount_rel ~fb: default_ws_fallback x

      

    let (install_handler :

         endpoint: ('s endpoint) -> fb: 's -> what: 's -> unit) =

      fun ~endpoint ~fb ~what ->

        let (mount_point, how_where) = endpoint in

        let segpath =

          match how_where with | `Service ((s, _)) -> s | `Fallback s -> s in

        let disp_level =

          go_or_create_bigstep ~fb ~disp: mount_point ~segs: segpath

        in

          match how_where with

          | `Service ((_, p)) ->

              (match disp_level.seg_map#get_opt p with

               | None -> disp_level.seg_map#replace p (Service what)

               | Some _ -> raise Already_handled)

          | `Fallback _ -> disp_level.fallback <- what

      

    (*

    value remove_handler ~endpoint =

      (ignore endpoint; raise E_xit)

    ;

*)

(**********)



    let mount_http endpoint f =

      let () = dbg "S.mount"

      in install_handler ~fb: default_http_fallback ~endpoint ~what: f

      

    let mount_websocket endpoint f =

      let () = dbg "S.mount"

      in install_handler ~fb: default_ws_fallback ~endpoint ~what: f

      

    (*

    value umount endpoint =

      remove_handler ~endpoint

    ;

*)

(**********)



    let (server_func :

         H.http_service_func disp_level ->

           H.websocket_service_func disp_level ->

             request -> (segpath * H.service_desc)) =

      fun root_http_disp_level root_ws_disp_level rq -> let open Amall_http

        in

          let uri = rq.rq_uri

          in

            match uri.scheme with

            | None -> invalid_arg "S.server_func: no scheme in uri"

            | Some sch ->

                let segpath = segpath_of_uri uri in

                let () =

                  dbg "S.server_func: proto=%s segpath=[%s]" sch

                    ((String.concat ";") & (List.map (sprintf "%S") segpath))

                in

                  (match sch with

                   | "http" ->

                       find_handler root_http_disp_level segpath

                         (fun s -> `Service_http s)

                   | "ws" ->

                       find_handler root_ws_disp_level segpath

                         (fun s -> `Service_ws s)

                   | _ -> invalid_arg "S.server_func: no scheme %s" sch)

      

    let listener_create addr =

      let root_http_disp_level =

        { seg_map = new c_seg_map; fallback = default_http_fallback; }

      and root_ws_disp_level =

        { seg_map = new c_seg_map; fallback = default_ws_fallback; } in

      let server_func =

        server_func root_http_disp_level root_ws_disp_level in

      let io_server_lazy = lazy (HS.run addr server_func) in

      let listener =

        {

          io_server_lazy = io_server_lazy;

          root_http_disp_level = root_http_disp_level;

          root_ws_disp_level = root_ws_disp_level;

        }

      in (listener, root_http_disp_level, root_ws_disp_level)

      

    let io_listener_run l : IO.server = Lazy.force l.io_server_lazy

      

    let listener_run l =

      let io_srv = io_listener_run l

      in

        match IO.runIO (IO.wait_server io_srv) with

        | `Ok () -> ()

        | `Error e -> raise e

      

    let listener_stop l = IO.shutdown_server (Lazy.force l.io_server_lazy)

      

  end