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.
module Id : sig ... endmodule Metadata : sig ... endval 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) tval bin_query : ('query, _, _, _) t -> 'query Core.Bin_prot.Type_class.tval bin_state : (_, 'state, _, _) t -> 'state Core.Bin_prot.Type_class.tval bin_update : (_, _, 'update, _) t -> 'update Core.Bin_prot.Type_class.tval bin_error : (_, _, _, 'error) t -> 'error Core.Bin_prot.Type_class.tval shapes : (_, _, _, _) t -> Rpc_shapes.tval 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.tval 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.tval 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.tval 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.tval 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.tmodule Pipe_message = Pipe_messagemodule Pipe_response = Pipe_responseval 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.tdispatch_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.tval abort : (_, _, _, _) t -> Connection.t -> Id.t -> unitval close_reason : Metadata.t -> Pipe_close_reason.t Async_kernel.Deferred.tval client_pushes_back : (_, _, _, _) t -> boolval name : (_, _, _, _) t -> stringval version : (_, _, _, _) t -> intval description : (_, _, _, _) t -> Description.tval query_type_id : ('query, _, _, _) t -> 'query Core.Type_equal.Id.tval state_type_id : (_, 'state, _, _) t -> 'state Core.Type_equal.Id.tval update_type_id : (_, _, 'update, _) t -> 'update Core.Type_equal.Id.tval error_type_id : (_, _, _, 'error) t -> 'error Core.Type_equal.Id.t