Source file persistent_connection_intf.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
open! Core
open Async_kernel
open Async_log_kernel
open Async_rpc_kernel
open Async_unix
open! Import

module type Address = Persistent_connection_kernel.Address
module type Connection_error = Persistent_connection_kernel.Connection_error
module type Closable = Persistent_connection_kernel.Closable

(** The subset of [S'] that excludes the [create] functions. Useful for modules that wrap
    a persistent connection but don't want to expose the create functions. *)
module type S'_without_create = Persistent_connection_kernel.S'_without_create

module type S_without_create = S'_without_create with type conn_error := Error.t

module type S' = sig
  include S'_without_create

  val create
    :  created_at:[%call_pos]
    -> server_name:string
    -> ?log:Log.t
         (** If [~log] is supplied then all events that would be passed to [on_event] will
             be written there as well, with a "persistent-connection-to" tag value of
             [server_name], which should be the name of the server we are connecting to. *)
    -> ?event_log_level:('address Event.t -> [ `Info | `Debug | `Warn | `Error ])
         (** If [~event_log_level] is supplied, it overrides the default mapping from
             events to log levels. The default is
             [Persistent_connection_kernel.Event.log_level]. *)
    -> ?on_event:('address Event.t -> unit Deferred.t)
    -> ?retry_delay:(unit -> Time_float.Span.t)
    -> ?random_state:[ `Non_random | `State of Random.State.t ]
         (** If a [~random_state] is supplied, randomization is applied to the result of
             [retry_delay] after each call; if not, no randomization will be applied. The
             default is [`State Random.State.default]. *)
    -> ?time_source:Time_source.t
    -> connect:('address -> (conn, conn_error) Result.t Deferred.t)
    -> address:(module Address with type t = 'address)
    -> (unit -> ('address, conn_error) Result.t Deferred.t)
    -> t

  (** Like [create], but passes a [Connect_context.t] to the [connect] callback. This
      allows the callback to abort the persistent connection from within the connect
      logic. *)
  val create_with_connect_context
    :  created_at:[%call_pos]
    -> server_name:string
    -> ?log:Log.t
    -> ?event_log_level:('address Event.t -> [ `Info | `Debug | `Warn | `Error ])
    -> ?on_event:('address Event.t -> unit Deferred.t)
    -> ?retry_delay:(unit -> Time_float.Span.t)
    -> ?random_state:[ `Non_random | `State of Random.State.t ]
    -> ?time_source:Time_source.t
    -> connect:
         (Persistent_connection_kernel.Connect_context.t
          -> 'address
          -> (conn, conn_error) Result.t Deferred.t)
    -> address:(module Address with type t = 'address)
    -> (unit -> ('address, conn_error) Result.t Deferred.t)
    -> t
end

module type S = S' with type conn_error := Error.t

module type S_rpc = sig
  include S

  (** Like [create] but for Rpc-like connections (i.e. Async.Rpc and Async.Versioned_rpc)
      where there is an obvious default for [connect] -- with a handful of extra optional
      parameters to pass to the [Rpc.Connection] functions. *)
  val create'
    :  created_at:[%call_pos]
    -> server_name:string
    -> ?log:Log.t
    -> ?event_log_level:(Host_and_port.t Event.t -> [ `Info | `Debug | `Warn | `Error ])
    -> ?on_event:(Host_and_port.t Event.t -> unit Deferred.t)
    -> ?retry_delay:(unit -> Time_float.Span.t)
    -> ?random_state:[ `Non_random | `State of Random.State.t ]
    -> ?time_source:Time_source.t
    -> ?bind_to_address:Unix.Inet_addr.t
    -> ?implementations:Rpc.Connection.Client_implementations.t
    -> ?max_message_size:int
    -> ?make_transport:Async_rpc.Rpc.Connection.transport_maker
    -> ?handshake_timeout:Time_float.Span.t
    -> ?heartbeat_config:Rpc.Connection.Heartbeat_config.t
    -> (unit -> Host_and_port.t Or_error.t Deferred.t)
    -> t
end

module type Persistent_connection = sig
  (** A persistent connection is one that is automatically reestablished whenever lost. *)

  module type Address = Address
  module type Closable = Closable
  module type Connection_error = Connection_error
  module type S'_without_create = S'_without_create
  module type S_without_create = S_without_create
  module type S = S
  module type S' = S'
  module type S_rpc = S_rpc

  module Connect_context = Persistent_connection_kernel.Connect_context

  module Make (Conn : Closable) :
    S with type conn = Conn.t and type t = Persistent_connection_kernel.Make(Conn).t

  module Make' (Conn_err : Connection_error) (Conn : Closable) :
    S'
    with type conn = Conn.t
     and type conn_error = Conn_err.t
     and type t = Persistent_connection_kernel.Make'(Conn_err)(Conn).t

  module Rpc :
    S_rpc
    with type conn = Rpc.Connection.t
     and type t =
      Persistent_connection_kernel.Make(Async_rpc_kernel.Persistent_connection.Rpc_conn).t

  module Versioned_rpc :
    S_rpc
    with type conn = Versioned_rpc.Connection_with_menu.t
     and type t =
      Persistent_connection_kernel.Make
        (Async_rpc_kernel.Persistent_connection.Versioned_rpc_conn)
      .t
end