diff options
| -rw-r--r-- | bin/server.ml | 2 | ||||
| -rw-r--r-- | lib/RNT.ml | 45 | ||||
| -rw-r--r-- | lib/RNT.mli | 4 | ||||
| -rw-r--r-- | lib/configuration.ml | 29 | ||||
| -rw-r--r-- | lib/drl/executor.ml | 40 | ||||
| -rw-r--r-- | lib/listener.ml | 254 | ||||
| -rw-r--r-- | lib/session.ml | 36 | ||||
| -rw-r--r-- | lib/sublanguage_types.ml | 2 | ||||
| -rw-r--r-- | lib/system.ml | 86 |
9 files changed, 204 insertions, 294 deletions
diff --git a/bin/server.ml b/bin/server.ml index 64c4594..2a47cdb 100644 --- a/bin/server.ml +++ b/bin/server.ml @@ -7,6 +7,6 @@ let () = end; match System.run_from_config Sys.argv.(1) with | Error e -> - Printf.eprintf "Couldn't initialize: %s\n%!" (Condition.to_string e); + Printf.eprintf "Couldn't initialize: %s\n%!" (Condition.to_string_hum e); exit 1 | Ok run -> run () @@ -10,13 +10,21 @@ module M = K.Multigroup.Make (S) module SI = K.Storage.Make (S) module SR = K.SubstantialRelation.Make (S) module T = K.Tuple.Make (S) +module Sess = K.Session.Make (S) module Error = struct open Condition let of_kernel c = - condition "kernel-condition" "RNT signalled a condition" - ("report" |=| Value.String (C.Condition.to_string_hum c)) + let report = C.Condition.to_string_hum c in + let report = + match BatString.Exceptionless.split report ~by:"\n\nBacktrace:" with + | Some (report, _) -> report + | None -> report + in + condition "kernel-condition" + (BatString.nreplace ~str:(BatString.trim report) ~sub:"\n\t" ~by:"; ") + empty let invalid_path path = condition "invalid-path" "A path must start at \\ and have no empty segments" @@ -55,8 +63,10 @@ type blueprint = | Branch | Multigroup | Relation of {heading: string Utilities.StringMap.t; tuples: Value.t Utilities.StringMap.t list} + | Session of string type plan = Base of path | Project of plan * string list +type session = {branch: string; snapshot: string} let lift r = Result.map_error Error.of_kernel r let value_of : C.Value.value -> Value.t = function String s -> String s | Integer n -> Integer n @@ -139,7 +149,12 @@ let conforms heading attributes = Utilities.StringMap.( equal String.equal heading (map (fun v -> T.Attribute.domain (kernel_value_of v)) attributes) ) -let make {connection; _} path name = function +let pinned kernel branch = + using kernel ["branch"; branch] (fun handle -> + let* addressable = P.Handle.require P.Addressable.from handle in + Ok {P.Session.branch; snapshot= P.Addressable.address addressable} ) + +let make ({connection; _} as kernel) path name = function | Namespace -> Ok (K.Namespace.make ()) | Branch -> B.make connection |> lift | Multigroup -> M.make connection |> lift @@ -147,12 +162,14 @@ let make {connection; _} path name = function if List.for_all (conforms heading) tuples then store_relation connection name heading tuples |> lift else Error (Error.heading_mismatch path) + | Session branch -> + let* state = pinned kernel branch |> lift in + Ok (Sess.make connection state) let initialize configuration = let* connection = S.connect (C.Configuration.term_of_sexp configuration) |> lift in - (let* root = I.initialize ~evaluators:["fol", FOL.make ()] connection in - let* _ = K.Path.(update root this "session" None (Some (K.Namespace.make ()))) in - Ok {connection; root} ) + I.initialize ~evaluators:["fol", FOL.make ()] connection + |> Result.map (fun root -> {connection; root}) |> Result.map_error (fun c -> S.disconnect connection; Error.of_kernel c) let shutdown {connection; _} = S.disconnect connection @@ -184,6 +201,22 @@ let remove kernel path = let* removed = lift removed in if removed then Ok () else Error (Error.conflict path) +let with_session kernel path f = + let* segments = segments_of path in + using kernel segments (fun handle -> + let* session = P.Session.require handle in + Ok (f session) ) + |> lift + +let session kernel path = + with_session kernel path P.Session.state + |> Result.map (fun {P.Session.branch; snapshot} -> + {branch; snapshot= C.Hash.to_hum_string snapshot} ) + +let pin kernel path branch = + let* state = pinned kernel branch |> lift in + with_session kernel path (fun session -> P.Session.pin session state) + let release = P.Handle.release let identity handle = P.Handle.hash handle |> C.Hash.to_hum_string diff --git a/lib/RNT.mli b/lib/RNT.mli index 841ab2d..db9e111 100644 --- a/lib/RNT.mli +++ b/lib/RNT.mli @@ -9,8 +9,10 @@ type blueprint = | Branch | Multigroup | Relation of {heading: string Utilities.StringMap.t; tuples: Value.t Utilities.StringMap.t list} + | Session of string type plan = Base of path | Project of plan * string list +type session = {branch: string; snapshot: string} val initialize : Sexplib.Sexp.t -> (kernel, Condition.condition) result val shutdown : kernel -> unit @@ -18,6 +20,8 @@ val list : kernel -> path -> (string list, Condition.condition) result val lookup : kernel -> path -> (handle, Condition.condition) result val create : kernel -> path -> blueprint -> (unit, Condition.condition) result val remove : kernel -> path -> (unit, Condition.condition) result +val session : kernel -> path -> (session, Condition.condition) result +val pin : kernel -> path -> string -> (unit, Condition.condition) result val release : handle -> unit val identity : handle -> string val describe : handle -> (string Utilities.StringMap.t, Condition.condition) result diff --git a/lib/configuration.ml b/lib/configuration.ml index 420dff6..6f45de0 100644 --- a/lib/configuration.ml +++ b/lib/configuration.ml @@ -5,14 +5,13 @@ module type CONFIGURABLE = sig end module Error = struct - (* open Condition let unknown_section key = - condition "unknown-section" "Unknown configuration section" ("key" |=| of_string key) + condition "unknown-section" "Unknown configuration section" ("key" |=| Value.String key) let duplicate_section key = - condition "duplicate-section" "Duplicate configuration section" ("key" |=| of_string key) + condition "duplicate-section" "Duplicate configuration section" ("key" |=| Value.String key) let empty_section key = condition "empty-section" "Empty configuration section (expected a tagged value)" @@ -25,11 +24,11 @@ module Error = struct let invalid_toplevel toplevel = condition "invalid-toplevel" "Configuration must be a (server ...) s-expression" - ("expression" |=| of_sexp toplevel) + ("expression" |=| Value.String (Sexplib.Sexp.to_string toplevel)) let malformed_section expr = condition "malformed-section" "Section must be a (tag ...) s-expression" - ("expression" |=| of_sexp expr) + ("expression" |=| Value.String (Sexplib.Sexp.to_string expr)) let missing_section name = condition "missing-section" "Missing section in configuration" ("name" |=| Value.String name) @@ -38,7 +37,7 @@ module Error = struct condition "invalid-tag" "Invalid tag for section" ( "section" |=| Value.String section & "tag" |=| Value.String tag - & "valid-options" |=| of_list Value.String valid_options ) + & "valid-options" |=| Value.String (String.concat " " valid_options) ) let failed_to_load_file path msg = condition "failed-to-load-file" "Failed to load configuration file" @@ -46,19 +45,15 @@ module Error = struct let syntax_error path msg = condition "syntax-error" "Failed to parse configuration file" - ("path" |=| Value.String path & "message" |=| Value.String msg) - *) - + ("path" |=| Value.String path & "message" |=| Value.String msg) end -(* - (** Section name -> raw sexp subtree. *) type t = Sexplib.Sexp.t Utilities.StringMap.t let find_section (key : string) (config : t) : Sexplib.Sexp.t option = Utilities.StringMap.find_opt key config -( + (** Check that [key] is expected and not duplicate, then add it. *) let insert_section ~expected acc key body = let ( let* ) = Result.bind in @@ -74,7 +69,7 @@ let insert_section ~expected acc key body = | _ -> Error (Error.multiple_values_in_section key) (** [(server ...)] sexp -> section map. Rejects unknown/duplicate keys. *) -let parse_server ~expected_keys (sexp : Sexplib.Sexp.t) : (t, Condition.t) result = +let parse_server ~expected_keys (sexp : Sexplib.Sexp.t) : (t, Condition.condition) result = let open Sexplib.Sexp in match sexp with | List (Atom "server" :: sections) -> @@ -91,14 +86,15 @@ let parse_server ~expected_keys (sexp : Sexplib.Sexp.t) : (t, Condition.t) resul | _ -> Error (Error.invalid_toplevel sexp) (** Read a file from disk and run [parse_server] on it. *) -let load ~expected_keys (path : string) : (t, Condition.t) result = +let load ~expected_keys (path : string) : (t, Condition.condition) result = match Sexplib.Sexp.load_sexp path with | sexp -> parse_server ~expected_keys sexp | exception Sys_error msg -> Error (Error.failed_to_load_file path msg) | exception Failure msg -> Error (Error.syntax_error path msg) (** [(tag field1 ...)] -> [(tag, List [field1; ...])]. *) -let extract_tagged_section (sexp : Sexplib.Sexp.t) : (string * Sexplib.Sexp.t, Condition.t) result = +let extract_tagged_section (sexp : Sexplib.Sexp.t) : + (string * Sexplib.Sexp.t, Condition.condition) result = let open Sexplib.Sexp in match sexp with | List (Atom tag :: body) -> Ok (tag, List body) @@ -106,7 +102,7 @@ let extract_tagged_section (sexp : Sexplib.Sexp.t) : (string * Sexplib.Sexp.t, C (** Look up a section by name, extract its tag, and check the tag is allowed. *) let require_section ~(name : string) ~(valid_tags : string list) (config : t) : - (string * Sexplib.Sexp.t, Condition.t) result = + (string * Sexplib.Sexp.t, Condition.condition) result = let ( let* ) = Result.bind in let* sexp = find_section name config |> Option.to_result ~none:(Error.missing_section name) in let* tag, body = extract_tagged_section sexp in @@ -114,4 +110,3 @@ let require_section ~(name : string) ~(valid_tags : string list) (config : t) : if List.mem tag valid_tags then Ok () else Error (Error.invalid_tag name tag valid_tags) in Ok (tag, body) - *) diff --git a/lib/drl/executor.ml b/lib/drl/executor.ml index e472ec5..dcfe716 100644 --- a/lib/drl/executor.ml +++ b/lib/drl/executor.ml @@ -1,27 +1,35 @@ open Utilities.Result +module Error = struct + open Condition + + let unqualified_name name = + condition "unqualified-name" "A relation must be named as multigroup:relation" + ("name" |=| Value.String name) +end + let page_limit = 16 -(* [mg:rel] names a relation on the session's branch; a bare name, one the session created. *) -let path_of (ctx : Sublanguage_context.t) name = +let relation_path branch name = match String.index_opt name ':' with - | None -> ctx.session#namespace ^ "\\" ^ name - | Some i -> - String.concat "\\" - [ ""; - "branch"; - ctx.session#branch; - "multigroup"; - String.sub name 0 i; - "relation"; - String.sub name (i + 1) (String.length name - i - 1) ] + | Some i when i > 0 && i < String.length name - 1 -> + Ok + (String.concat "\\" + [ branch; + "multigroup"; + String.sub name 0 i; + "relation"; + String.sub name (i + 1) (String.length name - i - 1) ] ) + | _ -> Error (Error.unqualified_name name) -let rec compile ctx = function - | Ast.Base name -> RNT.Base (path_of ctx name) - | Ast.Project (attributes, query) -> RNT.Project (compile ctx query, attributes) +let rec compile branch = function + | Ast.Base name -> relation_path branch name |> Result.map (fun path -> RNT.Base path) + | Ast.Project (attributes, query) -> + compile branch query |> Result.map (fun plan -> RNT.Project (plan, attributes)) let execute (ctx : Sublanguage_context.t) query = - let* relation = RNT.execute ctx.kernel (compile ctx query) in + let* plan = compile (ctx.session#path ^ "\\branch") query in + let* relation = RNT.execute ctx.kernel plan in Fun.protect ~finally:(fun () -> RNT.release relation) (fun () -> diff --git a/lib/listener.ml b/lib/listener.ml index 5a7560f..b4405a9 100644 --- a/lib/listener.ml +++ b/lib/listener.ml @@ -1,155 +1,101 @@ -module Make = -functor - (T : Transport.TRANSPORT) - (NT : Nt.S) - -> - struct - module type SubS = Sublanguage.S - - module Sess = Session.Make (NT) - module B = Branch.Make (NT) - module SC = Sublanguage_context.Make (NT) - - module Error = struct - open Condition - - let syntax_error msg = condition "syntax-error" msg empty - - let unrecognized_sublanguage tag = - condition "unrecognized-sublanguage" "Unrecognized sublanguage" ("tag" |=| of_string tag) - - let malformed_expression expr = - condition "malformed-expression" "Malformed expression" ("expression" |=| of_sexp expr) - end - - (** Per-connection state: session owns the active branch. The branch's - [relation_path] becomes the context resolver, so all sublanguages - address the full RNT namespace rather than a single multigroup. **) - type conn_state = {claims: Nt.claims; session: Sess.session} - - (** Build a fresh execution context from the current session state. Called - at the top of every handle_sublanguage so it always reflects the live - branch after any prior VCL switch. **) - let make_ctx ({session; claims} : conn_state) : Sublanguage_context.t = - SC.make_ctx session claims - - let read_command input = - try Ok (Sexplib.Sexp.input_sexp input) - with Sexplib.Sexp.Parse_error {err_msg; _} -> Error (Error.syntax_error err_msg) - - let sublanguages = - List.fold_right - (fun (module Language : SubS) -> - Utilities.StringMap.add Language.name (module Language : SubS) ) - [ (module Drl.Sublanguage.Make (NT) : SubS); - (module Ddl.Sublanguage.Make (NT) : SubS); - (module Dml.Sublanguage.Make (NT) : SubS); - (module Icl.Sublanguage.Make (NT) : SubS); - (module Prl.Sublanguage.Make (NT) : SubS); - (module Scl.Sublanguage.Make (NT) : SubS); - (module Vcl.Sublanguage.Make (NT) : SubS) ] - Utilities.StringMap.empty - - let find_language tag = - Utilities.StringMap.find_opt tag sublanguages - |> Option.to_result ~none:(Error.unrecognized_sublanguage tag) - - let execute_sublanguage ctx expr (module Language : SubS) = - Language.parse_sexp expr |> Utilities.Result.fmap (Language.execute ctx) - - let execute_command ctx = function - | Sexplib.Sexp.(List [Atom tag; expr]) -> - find_language tag |> Utilities.Result.fmap (execute_sublanguage ctx expr) - | s -> Error (Error.malformed_expression s) - - let tuple_to_sexp (t : Tuple.materialized) = - let open Sexplib.Sexp in - List - ( Tuple.AttributeMap.bindings t.Tuple.attributes - |> List.map (fun (k, attr) -> - List [Atom k; Conventions.AbstractValue.sexp_of_t attr.Attribute.value] ) ) - - let output_response out_ch sexp = - let response = Sexplib.Sexp.to_string sexp in - output_string out_ch response; - output_string out_ch "\n"; - flush out_ch; - Utilities.print_with_time response - - let send_error out_ch e = - output_response out_ch Sexplib.Sexp.(List [Atom "error"; Condition.to_sexp e]) - - (* Serializes a sublanguage result. [branch] is the write target whose - tip and name surface in Cursor / Transition responses. *) - let serialize (branch : B.branch) = - let open Sexplib.Sexp in - function - | Error e -> List [Atom "error"; Condition.to_sexp e] - | Ok (Sublanguage_types.Cursor {cursor_id; rows; has_more}) -> - let row_sexps = List.map tuple_to_sexp rows in - List - [ Atom "cursor"; - List [Atom "id"; Atom cursor_id]; - List [Atom "rows"; List row_sexps]; - List [Atom "row_count"; Atom (string_of_int (List.length row_sexps))]; - List [Atom "has_more"; Atom (string_of_bool has_more)]; - List [Atom "snapshot"; Atom branch#tip]; - List [Atom "branch"; Atom branch#name] ] - | Ok (Sublanguage_types.Query _rel) -> - List [Atom "error"; Atom "unexpected Query result: use Cursor path"] - | Ok (Sublanguage_types.Transition _new_cache) -> - List - [ Atom "ok"; - List [Atom "snapshot"; Atom branch#tip]; - List [Atom "branch"; Atom branch#name] ] - | Ok (Sublanguage_types.SessionSwitch name) -> - List [Atom "ok"; List [Atom "message"; Atom ("Switched to branch " ^ name)]] - | Ok (Sublanguage_types.CreateMultigroup name) -> - List [Atom "ok"; List [Atom "message"; Atom ("Multigroup " ^ name ^ " created")]] - - let handle_sublanguage output state sexp = - let ctx = make_ctx !state in - match execute_command ctx sexp with - | Error e -> send_error output e - | Ok result -> - (* Sync updated schema_cache back into the branch mirror on any - state-advancing result (DDL, ICL, VCL). For VCL the session - already holds the new branch; refresh_mg is a no-op on it but - correct for DDL/ICL which stay on the same branch. *) - ( match result with - | Sublanguage_types.Transition delta -> !state.session#branch#apply_delta delta - | _ -> () ); - serialize !state.session#branch (Ok result) |> output_response output - - let handle_client connection = - let input = T.input connection in - let output = T.output connection in - match NT.authenticate Nt.PlainText with - | Error e -> - (* FIXME: why is this a syntax error? *) - send_error output e - | Ok claims -> ( - match Sess.open_session claims ~branch_name:"master" with - | Error e -> send_error output e - | Ok sess -> ( - let state = ref {claims; session= sess} in - try - while true do - match read_command input with - | Error e -> send_error output e - | Ok sexp -> handle_sublanguage output state sexp - done - with - | End_of_file -> ignore (!state.session#close ()) - | e -> - ignore (!state.session#close ()); - Printf.eprintf "Error handling connection: %s" (Printexc.to_string e) ) ) - - let spawn_handler connection = Stdlib.Domain.spawn (fun () -> handle_client connection) - - let run transport = - T.listen transport; - while true do - T.accept transport |> spawn_handler |> ignore - done +module Make (T : Transport.TRANSPORT) = struct + module type SubS = Sublanguage.S + + module Error = struct + open Condition + + let syntax_error msg = condition "syntax-error" msg empty + + let unrecognized_sublanguage tag = + condition "unrecognized-sublanguage" "Unrecognized sublanguage" ("tag" |=| Value.String tag) + + let malformed_expression expr = + condition "malformed-expression" "Malformed expression" + ("expression" |=| Value.String (Sexplib.Sexp.to_string expr)) end + + let sublanguages = + List.fold_right + (fun (module Language : SubS) -> + Utilities.StringMap.add Language.name (module Language : SubS) ) + [(module Drl.Sublanguage : SubS); (module Dml.Sublanguage : SubS)] + Utilities.StringMap.empty + + let find_language tag = + Utilities.StringMap.find_opt tag sublanguages + |> Option.to_result ~none:(Error.unrecognized_sublanguage tag) + + let read_command input = + try Ok (Sexplib.Sexp.input_sexp input) + with Sexplib.Sexp.Parse_error {err_msg; _} -> Error (Error.syntax_error err_msg) + + let execute_sublanguage ctx expr (module Language : SubS) = + Language.parse_sexp expr |> Utilities.Result.fmap (Language.execute ctx) + + let execute_command ctx = function + | Sexplib.Sexp.(List [Atom tag; expr]) -> + find_language tag |> Utilities.Result.fmap (execute_sublanguage ctx expr) + | s -> Error (Error.malformed_expression s) + + let field key value = Sexplib.Sexp.(List [Atom key; Atom value]) + + let sexp_of_tuple ({attributes; _} : RNT.tuple) = + Sexplib.Sexp.List + ( Utilities.StringMap.bindings attributes + |> List.map (fun (k, v) -> field k (Value.to_string v)) ) + + let meta session = + match session#state with + | Ok {RNT.branch; snapshot} -> [field "branch" branch; field "db_hash" snapshot] + | Error _ -> [] + + let serialize meta result = + let open Sexplib.Sexp in + let reply tag fields = List ((Atom tag :: fields) @ meta) in + match result with + | Error {Condition.name; message; properties; _} -> + reply "error" + ( field "message" (name ^ ": " ^ message) + :: ( BatMap.to_seq properties + |> List.of_seq + |> List.map (fun (k, v) -> field k (Value.to_string v)) ) ) + | Ok (Sublanguage_types.Cursor {rows; has_more}) -> + reply "cursor" + [ List [Atom "rows"; List (List.map sexp_of_tuple rows)]; + field "row_count" (string_of_int (List.length rows)); + field "has_more" (string_of_bool has_more) ] + | Ok (Sublanguage_types.Transition message) -> reply "ok" [field "message" message] + + let output_response out_ch sexp = + let response = Sexplib.Sexp.to_string sexp in + output_string out_ch response; + output_string out_ch "\n"; + flush out_ch; + Utilities.print_with_time response + + let handle_client kernel connection = + let input = T.input connection in + let output = T.output connection in + match Session.open_session kernel ~branch_name:"master" with + | Error e -> output_response output (serialize [] (Error e)) + | Ok session -> + let ctx = {Sublanguage_context.kernel; session} in + let rec serve () = + let result = read_command input |> Utilities.Result.fmap (execute_command ctx) in + serialize (meta session) result |> output_response output; + serve () + in + ( try serve () with + | End_of_file -> () + | e -> Printf.eprintf "Error handling connection: %s\n%!" (Printexc.to_string e) ); + ignore session#close; close_in_noerr input + + let spawn_handler kernel connection = + Stdlib.Domain.spawn (fun () -> handle_client kernel connection) + + let run kernel transport = + T.listen transport; + while true do + T.accept transport |> spawn_handler kernel |> ignore + done +end diff --git a/lib/session.ml b/lib/session.ml index ba4d96d..ca129d1 100644 --- a/lib/session.ml +++ b/lib/session.ml @@ -1,30 +1,15 @@ -type t = {identifier: string; branch: string; snapshot: string; commit: unit} +let path_of identifier = "\\session\\" ^ identifier -let namespace identifier = "\\session\\" ^ identifier -let branch_path name = "\\branch\\" ^ name - -let snapshot_of kernel branch = - let open Utilities.Result in - let* handle = RNT.lookup kernel (branch_path branch) in - let snapshot = RNT.identity handle in - RNT.release handle; Ok snapshot - -class session ~kernel ~identifier ~branch ~snapshot ~commit = +class session ~kernel ~identifier = object - val mutable session : t = {identifier; branch; snapshot; commit} - method identifier = session.identifier - method branch = session.branch - method snapshot = session.snapshot - method namespace = namespace session.identifier + method identifier = identifier + method path = path_of identifier + method state = RNT.session kernel (path_of identifier) (** Switch to a different branch within the same session. *) - method set_branch_to branch = - let open Utilities.Result in - let* snapshot = snapshot_of kernel branch in - session <- {session with branch; snapshot}; - Ok () + method set_branch_to branch = RNT.pin kernel (path_of identifier) branch - method close = RNT.remove kernel (namespace session.identifier) + method close = RNT.remove kernel (path_of identifier) end (* Open a session on a live (mutable) branch. @@ -34,12 +19,11 @@ let open_session kernel ~branch_name = let* branches = RNT.list kernel "\\branch" in let* () = if List.mem branch_name branches then Ok () - else RNT.create kernel (branch_path branch_name) RNT.Branch + else RNT.create kernel ("\\branch\\" ^ branch_name) RNT.Branch in - let* snapshot = snapshot_of kernel branch_name in let identifier = Printf.sprintf "%016Lx" (Random.State.bits64 (Random.State.make_self_init ())) in - let* () = RNT.create kernel (namespace identifier) RNT.Namespace in - Ok (new session ~kernel ~identifier ~branch:branch_name ~snapshot ~commit:()) + let* () = RNT.create kernel (path_of identifier) (RNT.Session branch_name) in + Ok (new session ~kernel ~identifier) (* Open a session pinned to a specific snapshot hash (detached / read-only). Mutations through the returned branch are refused. *) diff --git a/lib/sublanguage_types.ml b/lib/sublanguage_types.ml index 73d213a..e676d9a 100644 --- a/lib/sublanguage_types.ml +++ b/lib/sublanguage_types.ml @@ -1 +1 @@ -type result = Cursor of {rows: RNT.tuple list; has_more: bool} +type result = Cursor of {rows: RNT.tuple list; has_more: bool} | Transition of string diff --git a/lib/system.ml b/lib/system.ml index 16c3b42..1848682 100644 --- a/lib/system.ml +++ b/lib/system.ml @@ -1,76 +1,16 @@ -(** System assembly: registry-based configuration dispatch. This module is the - composition root. It owns the mapping from configuration tags to concrete - implementations and assembles a server from a config file. *) +open Utilities.Result -type nt_parcel = NtParcel : (module Nt.S) -> nt_parcel - -type transport_parcel = - | TransportParcel : (module Transport.TRANSPORT with type t = 't) * 't -> transport_parcel - -type nt_provider = Sexplib.Sexp.t -> (nt_parcel, Condition.t) result -type transport_provider = Sexplib.Sexp.t -> (transport_parcel, Condition.t) result - -type registry = - {nt: nt_provider Utilities.StringMap.t; transport: transport_provider Utilities.StringMap.t} - -module Error = struct - open Condition - - let unknown_backend component tag = - condition "unknown-backend" "Unknown backend for component" - ("component" |=| of_string component & "tag" |=| of_string tag) -end - -let registry : registry = - let open Utilities.StringMap in - { nt= - empty - |> add "memory" (fun _sexp -> Ok (NtParcel (module Nt.Memory : Nt.S))) - |> add "sqlite" (fun sexp -> - (* Accept (sqlite) for :memory: or (sqlite "/path/to/db") for a file. *) - let path = match sexp with Sexplib.Sexp.(List [Atom p]) -> p | _ -> ":memory:" in - let module M = Nt.Make (struct - let driver = "sqlite" - let init_arg = path - end) in - Ok (NtParcel (module M : Nt.S)) ); - transport= - empty - |> add "tcp" (fun sexp -> - let ( let* ) = Result.bind in - let* config = Transport.TCP.parse sexp in - let transport = Transport.TCP.create config in - Ok (TransportParcel ((module Transport.TCP), transport)) ) } - -let assemble (config : Configuration.t) : (unit -> unit, Condition.t) result = - let open Utilities.Result in - let* nt_tag, nt_body = - Configuration.require_section ~name:"nt" - ~valid_tags:(Utilities.StringMap.bindings registry.nt |> List.map fst) - config - in - let* nt_provider = - Utilities.StringMap.find_opt nt_tag registry.nt - |> Option.to_result ~none:(Error.unknown_backend "nt" nt_tag) - in - let* packed_nt = nt_provider nt_body in - let* transport_tag, transport_body = - Configuration.require_section ~name:"transport" - ~valid_tags:(Utilities.StringMap.bindings registry.transport |> List.map fst) - config +let assemble config = + let* storage = + Configuration.find_section "storage" config + |> Option.to_result ~none:(Configuration.Error.missing_section "storage") in - let* transport_provider = - Utilities.StringMap.find_opt transport_tag registry.transport - |> Option.to_result ~none:(Error.unknown_backend "transport" transport_tag) - in - let* packed_transport = transport_provider transport_body in - let (NtParcel (module NT)) = packed_nt in - let (TransportParcel ((module T), transport)) = packed_transport in - let* () = NT.initialize () in - let module L = Listener.Make (T) (NT) in - Ok (fun () -> L.run transport) - -let run_from_config (path : string) : (unit -> unit, Condition.t) result = - let ( let* ) = Result.bind in - let* config = Configuration.load ~expected_keys:["nt"; "transport"] path in + let* _, transport = Configuration.require_section ~name:"transport" ~valid_tags:["tcp"] config in + let* transport = Transport.TCP.parse transport in + let* kernel = RNT.initialize storage in + let module L = Listener.Make (Transport.TCP) in + Ok (fun () -> L.run kernel (Transport.TCP.create transport)) + +let run_from_config path = + let* config = Configuration.load ~expected_keys:["storage"; "transport"] path in assemble config |