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
open! Oxbow_core
module Error = struct
type t =
| Connection_failed of string
| Protocol of string
end
let send ~env ?seat ?socket body =
let path = Socket_path.resolve ?override:socket () in
let net = Eio.Stdenv.net env in
let addr = `Unix path in
let req = Request.{ body; seat } in
let req_str = Yojson.Safe.to_string (Request.yojson_of_t req) ^ "\n" in
try
Eio.Switch.run
@@ fun sw ->
let flow = Eio.Net.connect ~sw net addr in
Eio.Flow.copy_string req_str flow;
Eio.Flow.shutdown flow `Send;
let buf = Eio.Buf_read.of_flow flow ~max_size:65536 in
let line = Eio.Buf_read.line buf in
let resp = Response.t_of_yojson @@ Yojson.Safe.from_string line in
if resp.ok
then Ok resp.data
else Error (Error.Protocol (Option.value ~default:"unspecified" resp.err))
with
| Eio.Io _ as ex -> Error (Connection_failed (Format.asprintf "%a" Eio.Exn.pp ex))
;;
let subscribe ~env ?socket ?output ~kinds f =
let path = Socket_path.resolve ?override:socket () in
let net = Eio.Stdenv.net env in
let addr = `Unix path in
let body = Request.Body.Subscribe { kinds; output } in
let req = Request.{ body; seat = None } in
let req_str = Yojson.Safe.to_string (Request.yojson_of_t req) ^ "\n" in
try
Eio.Switch.run
@@ fun sw ->
let flow = Eio.Net.connect ~sw net addr in
Eio.Flow.copy_string req_str flow;
let buf = Eio.Buf_read.of_flow flow ~max_size:65536 in
let line = Eio.Buf_read.line buf in
let resp = Response.t_of_yojson @@ Yojson.Safe.from_string line in
if not resp.ok
then Error (Error.Protocol (Option.value ~default:"unspecified" resp.err))
else (
try
while true do
f (Eio.Buf_read.line buf)
done;
Ok ()
with
| End_of_file -> Ok ())
with
| Eio.Io _ as ex -> Error (Connection_failed (Format.asprintf "%a" Eio.Exn.pp ex))
;;