async_rpc_kernel v0.17.0 · OCaml Package

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.

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

module Id : sig ... end

module Metadata : sig ... end

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

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

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

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

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

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

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

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

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

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

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

module Pipe_message = Pipe_message

module Pipe_response = Pipe_response

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.

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

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

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

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

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

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

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

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

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

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

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

package async_rpc_kernel