aboutsummaryrefslogtreecommitdiffstats
path: root/lib/listener.ml
diff options
context:
space:
mode:
Diffstat (limited to 'lib/listener.ml')
-rw-r--r--lib/listener.ml325
1 files changed, 116 insertions, 209 deletions
diff --git a/lib/listener.ml b/lib/listener.ml
index 5e8f55d..262a2b9 100644
--- a/lib/listener.ml
+++ b/lib/listener.ml
@@ -1,25 +1,51 @@
-module type LISTENER = functor
- (T : Transport.TRANSPORT)
- (S : Management.Physical.S with type error = string)
- -> sig
- val run : T.t -> S.t -> unit
-end
-
-module Make : LISTENER =
+module Make =
functor
(T : Transport.TRANSPORT)
- (S : Management.Physical.S with type error = string)
+ (NT : Nt.S)
->
struct
- module type SubS = Sublanguage.S with type storage = S.t
-
- module AlgebraOps = Algebra.Make (S)
-
- (* Per-connection state: the client must issue (use "name") before
- any sublanguage command is accepted. *)
- type conn_state =
- | NoMultigroup
- | Active of Nt.handle * Management.Multigroup.multigroup
+ module type SubS = Sublanguage.S
+
+ module Sess = Session.Make (NT)
+ module B = Branch.Make (NT)
+
+ (** 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 (state : conn_state) : Sublanguage_context.t =
+ let br = state.session#branch in
+ let claims = state.claims in
+ let session = state.session in
+ {
+ Sublanguage_context.write_handle = br#branch_handle;
+ branch = (br :> Sublanguage_context.branch_view);
+ resolve = br#path;
+ switch_branch = fun name ->
+ match br#close () with
+ | Error e -> Error e
+ | Ok () ->
+ begin match B.open_branch claims name with
+ | Error e ->
+ begin match B.open_branch claims "master" with
+ | Ok mbr -> session#set_branch mbr
+ | Error _ -> ()
+ end;
+ Error e
+ | Ok new_br ->
+ session#set_branch new_br;
+ Ok new_br#multigroups
+ end
+ }
let read_command input =
try Ok (Sexplib.Sexp.input_sexp input)
@@ -31,245 +57,126 @@ functor
(fun (module Language : SubS) ->
Utilities.StringMap.add Language.name (module Language : SubS))
[
- (module Drl.Sublanguage.Make (S) : SubS);
- (module Ddl.Sublanguage.Make (S) : SubS);
- (module Dml.Sublanguage.Make (S) : SubS);
- (module Icl.Sublanguage.Make (S) : SubS);
- (* (module Dcl.Sublanguage.Make (S) : SubS); *)
- (module Prl.Sublanguage.Make (S) : SubS);
- (module Scl.Sublanguage.Make (S) : 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 fmap f m = Result.bind m f
-
let find_language tag =
Utilities.StringMap.find_opt tag sublanguages
|> Option.to_result ~none:(Error.UnrecognizedSublanguage tag)
- let execute_sublanguage storage db expr (module Language : SubS) =
+ let execute_sublanguage ctx expr (module Language : SubS) =
Language.parse_sexp expr
- |> fmap (Language.execute storage db)
+ |> Utilities.Result.fmap (Language.execute ctx)
|> Result.map_error (fun e ->
Error.SublanguageError (Language.sexp_of_error e))
- let execute_command storage db = function
+ let execute_command ctx = function
| Sexplib.Sexp.(List [ Atom tag; expr ]) ->
- find_language tag |> fmap (execute_sublanguage storage db expr)
+ find_language tag |> Utilities.Result.fmap (execute_sublanguage ctx expr)
| s -> Error (Error.MalformedExpression s)
- (* Thread handle and db through the connection after a sublanguage result. *)
- let perform handle db result =
- match result with
- | Sublanguage_types.Transition new_db -> Ok (handle, new_db, result)
- | Sublanguage_types.SessionSwitch _ ->
- Error (Error.SyntaxError "SessionSwitch not supported")
- | Sublanguage_types.CreateMultigroup _ ->
- Error (Error.SyntaxError "CreateMultigroup not supported")
- | _ -> Ok (handle, db, result)
-
- let current_limit = 16
-
- let materialize_generator gen limit =
- let rec go gen pos acc count =
- if count >= limit then (List.rev acc, true)
- else
- match gen (Some pos) with
- | Generator.Done -> (List.rev acc, false)
- | Generator.Error _ -> (List.rev acc, false)
- | Generator.Value (t, next) -> (
- let mat =
- match t with
- | Tuple.Materialized m -> Some m
- | Tuple.NonMaterialized _ -> None
- in
- match mat with
- | None -> go next (pos + 1) acc count
- | Some m -> go next (pos + 1) (m :: acc) (count + 1))
- in
- go gen 0 [] 0
-
- let materialize_relation storage (rel : Relation.relation) limit =
- let gen = AlgebraOps.to_generator storage rel in
- materialize_generator gen limit
-
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 print_with_time str =
- (* TODO: a proper logger *)
- let now = Unix.gettimeofday () in
- let tm = Unix.localtime now in
- let orange = "\027[38;5;208m" in
- let reset = "\027[0m" in
- let formatted_time =
- Printf.sprintf "%04d-%02d-%02d %02d:%02d:%02d" (tm.Unix.tm_year + 1900)
- (tm.Unix.tm_mon + 1) tm.Unix.tm_mday tm.Unix.tm_hour tm.Unix.tm_min
- tm.Unix.tm_sec
- in
- print_endline
- @@ Printf.sprintf "%s[%s]%s %s" orange formatted_time reset str
+ 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;
- print_with_time response
+ Utilities.print_with_time response
let send_error out_ch e =
output_response out_ch
Sexplib.Sexp.(List [ Atom "error"; Error.sexp_of_error e ])
- let send_ok_message out_ch msg =
- output_response out_ch
- Sexplib.Sexp.(List [ Atom "ok"; List [ Atom "message"; Atom msg ] ])
-
- let serialize storage (db : Management.Multigroup.multigroup) =
+ (* 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"; Error.sexp_of_error e ]
| Ok (Sublanguage_types.Cursor { cursor_id; rows; has_more }) ->
let row_sexps = List.map tuple_to_sexp rows in
- let rows_sexp = List row_sexps in
List
[
Atom "cursor";
- List [ Atom "id"; Atom cursor_id ];
- List [ Atom "rows"; rows_sexp ];
- List
- [
- Atom "row_count"; Atom (string_of_int (List.length row_sexps));
- ];
- List [ Atom "has_more"; Atom (string_of_bool has_more) ];
- List [ Atom "db_hash"; Atom db#hash ];
- List [ Atom "db_name"; Atom db#name ];
- (* TODO: forsake the global cursor register in favour of per-connection bookkeeping *)
- ]
- | Ok (Sublanguage_types.Query rel) ->
- let tuples, truncated =
- materialize_relation storage rel current_limit
- in
- let schema_sexp =
- List
- (List.map
- (fun (a, d) -> List [ Atom a; Atom d ])
- rel#schema)
- in
- let rows_sexp = List (List.map tuple_to_sexp tuples) in
- List
- [
- Atom "relation";
- List [ Atom "name"; Atom rel#name ];
- List [ Atom "schema"; schema_sexp ];
- List [ Atom "rows"; rows_sexp ];
- List
- [ Atom "row_count"; Atom (string_of_int (List.length tuples)) ];
- List [ Atom "truncated"; Atom (string_of_bool truncated) ];
- List [ Atom "db_hash"; Atom db#hash ];
- List [ Atom "db_name"; Atom db#name ];
+ 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.Transition new_db) ->
+ | 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 "db_hash"; Atom new_db#hash ];
- List [ Atom "db_name"; Atom new_db#name ];
- ]
- | Ok (Sublanguage_types.SessionSwitch multigroup) ->
- List
- [
- Atom "ok";
- List
- [
- Atom "message"; Atom ("Switched to multigroup " ^ multigroup);
- ];
+ 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") ];
- ]
+ List [ Atom "ok"; List [ Atom "message"; Atom ("Multigroup " ^ name ^ " created") ] ]
- (* Open a handle to a multigroup by name via the NT object layer. The claims
- from Firewall become the connection_context for access checks. *)
- let open_multigroup (_storage : S.t) claims name =
- match
- Nt.open_handle () () ~path:[ "multigroups"; name ]
- ~connection_context:claims
- with
- | Error e -> Error (Error.SyntaxError e)
- | Ok handle -> (
- match Nt.deserialize () () Nt.Multigroup handle.Nt.obj with
- | Error e -> Error (Error.SyntaxError e)
- | Ok (Nt.Multigroup mg) -> Ok (handle, mg)
- | Ok _ -> Error (Error.SyntaxError "unexpected object type at path"))
-
- let handle_use storage claims output state name =
- match open_multigroup storage claims name with
- | Error e -> send_error output e
- | Ok (handle, mg) ->
- state := Active (handle, mg);
- send_ok_message output ("using multigroup: " ^ name)
-
- let handle_sublanguage storage output state sexp handle db =
- match
- Ok sexp
- |> fmap (execute_command storage db)
- |> fmap (perform handle db)
- with
+ 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 (new_handle, new_db, r) ->
- state := Active (new_handle, new_db);
- serialize storage new_db (Ok r) |> output_response output
-
- let dispatch_command storage claims output state sexp =
- match sexp with
- | Sexplib.Sexp.(List [ Atom "use"; Atom name ]) ->
- handle_use storage claims output state name
- | _ -> (
- match !state with
- | NoMultigroup ->
- send_error output
- (Error.SyntaxError "no multigroup selected; send (use \"name\")")
- | Active (handle, db) ->
- handle_sublanguage storage output state sexp handle db)
-
- let handle_client connection (storage : S.t) =
- let input = T.input connection in
+ | 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
- (* Authenticate via NT PermissionsManager::Firewall before anything else. *)
- let claims =
- match Nt.firewall () with
- | Ok c -> c
- | Error e -> Printf.eprintf "Auth failed: %s\n%!" e; ""
- in
- let state = ref NoMultigroup in
- (try
- while true do
- match read_command input with
- | Error e -> send_error output e
- | Ok sexp -> dispatch_command storage claims output state sexp
- done
- with
- | End_of_file -> ()
- | e ->
- Printf.eprintf "Error handling connection: %s" (Printexc.to_string e))
-
- let spawn_handler storage connection =
- Stdlib.Domain.spawn (fun () -> handle_client connection storage)
-
- let run transport storage =
+ match NT.authenticate Nt.PlainText with
+ | Error e ->
+ send_error output (Error.SyntaxError (Nt.string_of_error e))
+ | Ok claims ->
+ (match Sess.open_session claims ~branch_name:"master" with
+ | Error e ->
+ send_error output (Error.SyntaxError (Nt.string_of_error 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 storage |> ignore
+ T.accept transport |> spawn_handler |> ignore
done
end
generated by cgit v1.2.3 (git 2.25.1) at 2026-11-04 01:53:45 +0000