Source file plain_rpc.ml

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
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
open! Core
open! Async_kernel
open! Import
open Deferred.Or_error.Let_syntax
include Plain_rpc_intf

(* Implementation-wise, a [Streamable.Plain_rpc] is just a [Streamable.State_rpc] with the
   pipe unused. *)
let plain_impl_to_state f conn query =
  let%bind response = f conn query in
  let empty = Pipe.create_reader ~close_on_exception:false (fun _ -> Deferred.unit) in
  Deferred.Or_error.return (response, empty)
;;

module Update = Main.Of_atomic (Nothing)

type ('q, 'r) t = ('q, 'r, Nothing.t) State_rpc.t

module Direct_writer = struct
  type 'response_part t =
    ('response_part, Update.Intermediate.Part.t) State_rpc.Direct_writer.t

  let write_response_without_pushback_exn =
    State_rpc.Direct_writer.write_state_without_pushback_exn
  ;;

  let finalise_response_without_pushback_exn =
    State_rpc.Direct_writer.finalise_state_without_pushback_exn
  ;;

  let is_response_finalised = State_rpc.Direct_writer.is_state_finalised
  let response_finalised = State_rpc.Direct_writer.state_finalised
  let close = State_rpc.Direct_writer.close
  let closed = State_rpc.Direct_writer.closed
  let flushed = State_rpc.Direct_writer.flushed
  let is_closed = State_rpc.Direct_writer.is_closed

  module Expert = struct
    let create_response_part ~bin_writer response_part =
      State_rpc.Direct_writer.Expert.create_state_part
        ~state_bin_writer:bin_writer
        response_part
    ;;

    let finalise_response_message = State_rpc.Direct_writer.Expert.finalise_state_message
    let write_without_pushback = State_rpc.Direct_writer.Expert.write_without_pushback
  end
end

module Make (X : S) = struct
  module State_X = struct
    let name = X.name
    let version = X.version

    type query = X.query [@@deriving bin_io]
    type state = X.response

    module State = X.Response

    type update = Nothing.t

    module Update = Update

    let client_pushes_back = X.client_pushes_back
  end

  module M = State_rpc.Make (State_X)

  let rpc = M.rpc
  let implement' ?on_exception f = M.implement' ?on_exception (plain_impl_to_state f)
  let implement_direct ?on_exception f = M.implement_direct ?on_exception f
end

let description = State_rpc.description

let dispatch' rpc conn query =
  let%bind server_response = State_rpc.dispatch' rpc conn query in
  Or_error.map server_response ~f:(fun (response, pipe) ->
    Pipe.close_read pipe;
    response)
  |> return
;;

let dispatch rpc conn query = dispatch' rpc conn query |> Deferred.map ~f:Or_error.join

let implement ?on_exception rpc f =
  State_rpc.implement ?on_exception rpc (plain_impl_to_state f)
;;

let bin_query_shape = State_rpc.bin_query_shape
let bin_response_shape = State_rpc.bin_state_shape