blob: 4a99b29ddbc4f37778fbfeed169df63f7b928e6b (
plain) (
blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
|
open Lwt
open Unix
let host = Unix.inet_addr_loopback
let port = 7524
let create_socket () : Lwt_unix.file_descr t =
let sock = Lwt_unix.socket PF_INET SOCK_STREAM 0 in
Lwt_unix.set_close_on_exec sock;
Lwt_unix.setsockopt sock Unix.SO_REUSEADDR true;
let () = Lwt_unix.set_blocking sock false in
Lwt_io.write_line Lwt_io.stdout "🔍 Relational Engine Server Started"
>>= fun () ->
Lwt_unix.bind sock @@ ADDR_INET (host, port) >>= fun () ->
Lwt_unix.listen sock 10;
return sock
let client_read sock maxlen =
let str = Bytes.create maxlen in
let rec _read sock acc =
Lwt_unix.read sock str 0 maxlen >>= fun recvlen ->
Lwt_io.write_line Lwt_io.stdout
(Printf.sprintf "[LOG] Received %d bytes from client" recvlen)
>>= fun () ->
let acc = acc ^ Bytes.to_string (Bytes.sub str 0 recvlen) in
if recvlen <= maxlen then Lwt.return acc else _read sock acc
in
_read sock String.empty >>= fun _ ->
match Relation.write_and_retrieve () with
| Ok (last_commit, locations) ->
(* let (first_names, last_names) = (Planner.Scan.execute last_commit locations "user/first-name", Planner.Scan.execute last_commit locations "user/last-name") in *)
(* let tuples = Planner.Join.execute first_names last_names locations in *)
let fname, lname = List.split [] in
(* tuples in *)
let relation_result : Protocol.facts =
[
{
attribute_name = "user/first-name";
attribute_type = "string";
tuples = List.map Bytes.to_string fname;
};
{
attribute_name = "user/last-name";
attribute_type = "string";
tuples = List.map Bytes.to_string lname;
};
]
in
return @@ Xml.to_string (Protocol.facts_to_xml_light relation_result)
| Error err -> return err
[@@warning "-27"]
let rec socket_read sock =
(* (Int32.of_int (String.length results))) *)
Lwt_unix.accept sock >>= fun (client, _) ->
client_read client 512
(* >>= fun results -> Lwt_unix.send client (Data_encoding.Binary.to_bytes_exn Data_encoding.Encoding.int32 5l) 0 1 [] *)
>>= fun results ->
Lwt_unix.send client (Bytes.of_string results) 0 (String.length results) []
>>= fun _ ->
Lwt_unix.wait_write client >>= fun () ->
Lwt_unix.close client >>= fun () -> socket_read sock
let start () =
Stdlib.exit
@@ Lwt_main.run
( Lwt_exit.wrap_and_exit (create_socket ()) >>= fun sock ->
let clean_callback =
Lwt_exit.register_clean_up_callback ~loc:__LOC__ (fun _ ->
Lwt_io.write_line Lwt_io.stdout "💥 Shutting down" >>= fun () ->
Lwt_unix.close sock)
in
Lwt_exit.wrap_and_exit (socket_read sock) >>= fun () ->
let () = Lwt_exit.unregister_clean_up_callback clean_callback in
Lwt_exit.exit_and_wait 0 )
|
