this repo has no description
Add client library
+88
+6
idl/dune
+6
idl/dune
+56
idl/worker_rpc.ml
+56
idl/worker_rpc.ml
···
1
+
(** Worker rpc *)
2
+
3
+
(** Functions to facilitate RPC calls to web workers. *)
4
+
5
+
module Worker = Brr_webworkers.Worker
6
+
open Brr_io
7
+
open Js_top_worker_rpc
8
+
9
+
(** The assumption made in this module is that RPCs are answered in the order
10
+
they are made. *)
11
+
12
+
type context =
13
+
{ worker : Worker.t
14
+
; timeout : int
15
+
; timeout_fn : unit -> unit
16
+
; waiting : ((Rpc.response, exn) Result.t Lwt_mvar.t * int) Queue.t
17
+
}
18
+
19
+
exception Timeout
20
+
21
+
let demux context msg =
22
+
Lwt.async (fun () ->
23
+
match Queue.take_opt context.waiting with
24
+
| None ->
25
+
Lwt.return ()
26
+
| Some (mv, outstanding_execution) ->
27
+
Brr.G.stop_timer outstanding_execution;
28
+
let msg : string = Message.Ev.data (Brr.Ev.as_type msg) in
29
+
Lwt_mvar.put mv (Ok (Marshal.from_string msg 0)))
30
+
31
+
let start worker timeout timeout_fn =
32
+
let context = { worker; timeout; timeout_fn; waiting = Queue.create () } in
33
+
let () =
34
+
Brr.Ev.listen Message.Ev.message (demux context) (Worker.as_target worker)
35
+
in
36
+
context
37
+
38
+
let rpc : context -> Rpc.call -> Rpc.response Lwt.t =
39
+
fun context call ->
40
+
let open Lwt in
41
+
let jv = Marshal.to_bytes call [] in
42
+
let mv = Lwt_mvar.create_empty () in
43
+
let outstanding_execution =
44
+
Brr.G.set_timeout ~ms:1000000 (fun () ->
45
+
Lwt.async (fun () -> Lwt_mvar.put mv (Error Timeout));
46
+
context.timeout_fn ())
47
+
in
48
+
Queue.push (mv, outstanding_execution) context.waiting;
49
+
Worker.post context.worker jv;
50
+
Lwt_mvar.take mv >>= fun r ->
51
+
match r with
52
+
| Ok jv ->
53
+
let response = jv in
54
+
Lwt.return response
55
+
| Error exn ->
56
+
Lwt.fail exn
+26
idl/worker_rpc.mli
+26
idl/worker_rpc.mli
···
1
+
(* Worker_rpc *)
2
+
3
+
(** Functions to facilitate RPC calls to web workers.
4
+
5
+
The assumption made in this module is that RPCs are answered in the order
6
+
they are made. *)
7
+
8
+
(** Represents the channel used to communicate with the worker *)
9
+
type context
10
+
11
+
(** When RPC calls take too long, the Lwt promise is set to failed state with
12
+
this exception. *)
13
+
exception Timeout
14
+
15
+
val start : Brr_webworkers.Worker.t -> int -> (unit -> unit) -> context
16
+
(** [start worker timeout timeout_fn] initialises communications with a web
17
+
worker. [timeout] is the number of seconds to wait for a response from any
18
+
RPC before raising an error, and [timeout_fn] is called when a timeout
19
+
occurs. *)
20
+
21
+
open Js_top_worker_rpc
22
+
23
+
val rpc : context -> Rpc.call -> Rpc.response Lwt.t
24
+
(** [rpc context call] returns a promise containing the result from the worker.
25
+
If we wait longer than the timeout specified in [context] for a response,
26
+
the Lwt promise will fail with exception {!Timeout}. *)