with type conn = Rpc.Connection.t
type conn = Rpc.Connection.t
A connection, perhaps embellished with additional information upon connection.
module Event : sig ... end
val create : server_name:string -> ?on_event:( 'address Event.t -> unit Async_kernel.Deferred.t ) -> ?retry_delay:( unit -> Core.Time_ns.Span.t ) -> ?random_state:[ `Non_random | `State of Core.Random.State.t ] -> ?time_source:Async_kernel.Time_source.t -> connect:( 'address -> conn Core.Or_error.t Async_kernel.Deferred.t ) -> address: (module Persistent_connection_kernel__Persistent_connection_kernel_intf.Address with type t = 'address) -> ( unit -> 'address Core.Or_error.t Async_kernel.Deferred.t ) -> t
create ~server_name ~on_event ~retry_delay get_address returns a persistent connection to a server whose host and port are obtained via
get_address every time we try to connect. For example,
get_address might look up a server's host and port in catalog at a particular path to which multiple redundant copies of a service are publishing their location. If one copy dies, we get the address of the another one when looking up the address afterwards.
All connection events (see the type above) are passed to the
on_event callback, if given. When this callback becomes determined, we move on to the next step in our connection attempt (e.g. we won't actually attempt to connect until
Attempting_to_connect is finished). Note that
on_event Disconnected will only be called once
on_event (Connected conn) finishes even if the connection goes down during that callback.
`Failed_to_connect error and
`Obtained_address addr events are only reported if they are distinct from the most recent event of the same type that has taken place since the most recent
Connection is by default retried after
~percent:(Percent.of_mult 0.3) (retry_delay ()). The default for
const (sec 10.). Note that what this retry delay actually throttles is the delay between two connection attempts, so when a long-lived connection dies, connection is usually immediately retried, and if that failed, wait for another retry delay and retry.
time_source arguments are there to make persistent connection code more deterministically testable. They default to
`State Random.State.default and
Time_source.wall_clock (), respectively. If random_state is set to
`Non_random, retry_delay will be used directly.
connected returns the first available connection from the time it is called. When currently connected, the returned deferred is already determined. If
closed has been called, then the returned deferred is never determined.
connected_or_failed_to_connect is immediately determined as
Ok _ if
t is already connected. Otherwise it becomes determined the next time
t becomes connected or fails to connect or when
t is closed.
val server_name : t -> string
close t closes the current connection and stops it from trying to reconnect. After the deferred it returns becomes determined, the last connection has been closed and no others will be attempted.
close calls are ever generated internally in response to the connection being closed by the other side.
close t closes the connection. The returned deferred becomes determined once any resources needed to maintain the connection have been released.
val is_closed : t -> bool
is_closed t returns true if
close has ever been called (even if the returned deferred has not yet been fulfilled).
Note that some modules implementing
Closable may call close internally upon noticing that the connection was closed by the other side. The interface of such a module ought to say that this is the case.
val close_when_current_connection_is_closed : t -> unit
close_when_current_connection_is_closed t causes the persistent connection to not reconnect if the current connection closes or if it is not currently connected. It does not close any active connection.
type 'a create := server_name:string -> ?on_event:( 'a Event.t -> unit Async_kernel.Deferred.t ) -> ?retry_delay:( unit -> Core.Time_ns.Span.t ) -> ?random_state:[ `Non_random | `State of Core.Random.State.t ] -> ?time_source:Async_kernel.Time_source.t -> connect:( 'a -> conn Core.Or_error.t Async_kernel.Deferred.t ) -> ( unit -> 'a Core.Or_error.t Async_kernel.Deferred.t ) -> t