Source file persistent_connection.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
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
open! Core
open Async_kernel
open Async_log_kernel
open Async_rpc_kernel
open Async_unix
open! Import
open! Require_explicit_time_source
include Persistent_connection_intf
module Connect_context = Persistent_connection_kernel.Connect_context

module Make' (Conn_err : Connection_error) (Conn : Closable) = struct
  include Persistent_connection_kernel.Make' (Conn_err) (Conn)

  let make_on_event
    (type address)
    ~address:(module Address : Address with type t = address)
    ~created_at
    ~server_name
    ~log
    ~event_log_level
    ~on_event
    event
    =
    Option.iter log ~f:(fun log ->
      let open Ppx_log_syntax in
      let created_at =
        if Source_code_position.equal created_at Lexing.dummy_pos
        then None
        else Some created_at
      in
      [%log.t
        log
          (Persistent_connection_kernel.Event.Variants.to_name event)
          (event : Address.t Event.t)
          ~persistent_connection_to:(server_name : string)
          (created_at
           : (Source_code_position.t Sexp_hidden_in_test.t option[@sexp.option]))
        [@@level Some (event_log_level event)]]);
    on_event event
  ;;

  let create
    ~(created_at : [%call_pos])
    ~server_name
    ?log
    ?(event_log_level = Persistent_connection_kernel.Event.log_level)
    ?(on_event = fun _ -> Deferred.unit)
    ?retry_delay
    ?random_state
    ?time_source
    ~connect
    ~address
    get_address
    =
    let retry_delay =
      Option.map retry_delay ~f:(fun f () ->
        f () |> Time_ns.Span.of_span_float_round_nearest)
    in
    let on_event =
      make_on_event ~address ~created_at ~server_name ~log ~event_log_level ~on_event
    in
    create
      ~server_name
      ~on_event
      ?retry_delay
      ?random_state
      ?time_source
      ~connect
      ~address
      get_address
  ;;

  let create_with_connect_context
    ~(created_at : [%call_pos])
    ~server_name
    ?log
    ?(event_log_level = Persistent_connection_kernel.Event.log_level)
    ?(on_event = fun _ -> Deferred.unit)
    ?retry_delay
    ?random_state
    ?time_source
    ~connect
    ~address
    get_address
    =
    let retry_delay =
      Option.map retry_delay ~f:(fun f () ->
        f () |> Time_ns.Span.of_span_float_round_nearest)
    in
    let on_event =
      make_on_event ~address ~created_at ~server_name ~log ~event_log_level ~on_event
    in
    create_with_connect_context
      ~server_name
      ~on_event
      ?retry_delay
      ?random_state
      ?time_source
      ~connect
      ~address
      get_address
  ;;
end

module Make (Conn : Closable) = struct
  include Make' (Persistent_connection_kernel.Default_connection_error) (Conn)
end

let create_convenience_wrapper
  ~(create : created_at:[%call_pos] -> server_name:string -> _)
  ~connection_of_rpc_connection
  ~server_name
  ~created_at
  ?log
  ?event_log_level
  ?on_event
  ?retry_delay
  ?random_state
  ?time_source
  ?bind_to_address
  ?implementations
  ?max_message_size
  ?make_transport
  ?handshake_timeout
  ?heartbeat_config
  get_address
  =
  let connect host_and_port =
    let%bind.Deferred.Or_error conn =
      Async_rpc.Rpc.Connection.client
        (Tcp.Where_to_connect.of_host_and_port ?bind_to_address host_and_port)
        ?implementations
        ?max_message_size
        ?make_transport
        ?handshake_timeout
        ?heartbeat_config
        ~description:(Info.Portable.of_string ("persistent connection to " ^ server_name))
      >>| Or_error.of_exn_result
    in
    connection_of_rpc_connection conn
  in
  create
    ~server_name
    ?log
    ?event_log_level
    ?on_event
    ?retry_delay
    ?random_state
    ?time_source
    ~connect
    ~address:(module Host_and_port : Address with type t = Host_and_port.t)
    ~created_at
    get_address
;;

module Versioned_rpc = struct
  include Make (Async_rpc_kernel.Persistent_connection.Versioned_rpc_conn)

  let create' ~(created_at : [%call_pos]) ~server_name =
    create_convenience_wrapper
      ~server_name
      ~create
      ~created_at
      ~connection_of_rpc_connection:Versioned_rpc.Connection_with_menu.create
  ;;
end

module Rpc = struct
  include Make (Async_rpc_kernel.Persistent_connection.Rpc_conn)

  let create' ~(created_at : [%call_pos]) ~server_name =
    create_convenience_wrapper
      ~server_name
      ~create
      ~created_at
      ~connection_of_rpc_connection:Deferred.Or_error.return
  ;;
end