Module `Rpc.State_rpc`Source

A state rpc is an easy way for two processes to synchronize a data structure by sending updates over the wire. It's basically a pipe rpc that sends/receives an initial state of the data structure, and then updates, and applies the updates under the covers.

Sourcetype ('query, 'state, 'update, 'error) t

Sourcemodule Id : sig ... end

Sourcemodule Metadata : sig ... end

Source

val create : 
  ?client_pushes_back:unit ->
  name:string ->
  version:int ->
  bin_query:'query Core.Bin_prot.Type_class.t ->
  bin_state:'state Core.Bin_prot.Type_class.t ->
  bin_update:'update Core.Bin_prot.Type_class.t ->
  bin_error:'error Core.Bin_prot.Type_class.t ->
  unit ->
  ('query, 'state, 'update, 'error) t

Sourceval bin_query : ('query, _, _, _) t -> 'query Core.Bin_prot.Type_class.t

Sourceval bin_state : (_, 'state, _, _) t -> 'state Core.Bin_prot.Type_class.t

Sourceval bin_update : (_, _, 'update, _) t -> 'update Core.Bin_prot.Type_class.t

Sourceval bin_error : (_, _, _, 'error) t -> 'error Core.Bin_prot.Type_class.t

Sourceval shapes : (_, _, _, _) t -> Rpc_shapes.t

Source

val implement : 
  ?on_exception:On_exception.t ->
  ('query, 'state, 'update, 'error) t ->
  ('connection_state ->
    'query ->
    ('state * 'update Async_kernel.Pipe.Reader.t, 'error) Core.Result.t
      Async_kernel.Deferred.t) ->
  'connection_state Implementation.t

Source

val implement_with_auth : 
  ?on_exception:On_exception.t ->
  ('query, 'state, 'update, 'error) t ->
  ('connection_state ->
    'query ->
    ('state * 'update Async_kernel.Pipe.Reader.t, 'error) Core.Result.t
      Or_not_authorized.t
      Async_kernel.Deferred.t) ->
  'connection_state Implementation.t

Source

val implement_direct : 
  ?on_exception:On_exception.t ->
  ('query, 'state, 'update, 'error) t ->
  ('connection_state ->
    'query ->
    'update Pipe_rpc.Direct_stream_writer.t ->
    ('state, 'error) Core.Result.t Async_kernel.Deferred.t) ->
  'connection_state Implementation.t

Source

val implement_direct_with_auth : 
  ?on_exception:On_exception.t ->
  ('query, 'state, 'update, 'error) t ->
  ('connection_state ->
    'query ->
    'update Pipe_rpc.Direct_stream_writer.t ->
    ('state, 'error) Core.Result.t Or_not_authorized.t Async_kernel.Deferred.t) ->
  'connection_state Implementation.t

Source

val dispatch : 
  ('query, 'state, 'update, 'error) t ->
  Connection.t ->
  'query ->
  ('state * 'update Async_kernel.Pipe.Reader.t * Metadata.t, 'error)
    Core.Result.t
    Core.Or_error.t
    Async_kernel.Deferred.t

Sourcemodule Pipe_message = Pipe_message

Sourcemodule Pipe_response = Pipe_response

Source

val dispatch_fold : 
  ('query, 'state, 'update, 'error) t ->
  Connection.t ->
  'query ->
  init:('state -> 'acc) ->
  f:('acc -> 'update -> 'acc * Pipe_response.t) ->
  closed:('acc -> [ `By_remote_side | `Error of Core.Error.t ] -> 'result) ->
  (Id.t * 'result Async_kernel.Deferred.t, 'error) Core.Result.t
    Core.Or_error.t
    Async_kernel.Deferred.t

dispatch_fold is similar to Pipe_rpc.dispatch_iter. init will be called with the initial state, and then f will be called on each update message. If the update pipe is closed either by the implementer or an error, closed will be invoked on the last accumulated value to produce the result. The update stream can be stopped with abort, in which case closed will not be called and the 'result Deferred.t will never become determined.

Source

val dispatch' : 
  ('query, 'state, 'update, 'error) t ->
  Connection.t ->
  'query ->
  ('state * 'update Async_kernel.Pipe.Reader.t * Metadata.t, 'error)
    Core.Result.t
    Rpc_result.t
    Async_kernel.Deferred.t

Sourceval abort : (_, _, _, _) t -> Connection.t -> Id.t -> unit

Sourceval close_reason : Metadata.t -> Pipe_close_reason.t Async_kernel.Deferred.t

Sourceval client_pushes_back : (_, _, _, _) t -> bool

Sourceval name : (_, _, _, _) t -> string

Sourceval version : (_, _, _, _) t -> int

Sourceval description : (_, _, _, _) t -> Description.t

Sourceval query_type_id : ('query, _, _, _) t -> 'query Core.Type_equal.Id.t

Sourceval state_type_id : (_, 'state, _, _) t -> 'state Core.Type_equal.Id.t

Sourceval update_type_id : (_, _, 'update, _) t -> 'update Core.Type_equal.Id.t

Sourceval error_type_id : (_, _, _, 'error) t -> 'error Core.Type_equal.Id.t

package async_rpc_kernel

Module Rpc.State_rpcSource

Module `Rpc.State_rpc`Source