module Id : sig ... end
module Metadata : sig ... end
val implement : ('query, 'response, 'error) t -> ('connection_state -> 'query -> ('response Async_kernel.Pipe.Reader.t, 'error) Core_kernel.Result.t Async_kernel.Deferred.t) -> 'connection_state Implementation.t
The pipe returned by the implementation function will be closed automatically when either the connection to the client is closed or the client closes their pipe.
module Direct_stream_writer : sig ... end
Direct_stream_writer.t is a simple object for responding to a
val implement_direct : ('query, 'response, 'error) t -> ('connection_state -> 'query -> 'response Direct_stream_writer.t -> (unit, 'error) Core_kernel.Result.t Async_kernel.Deferred.t) -> 'connection_state Implementation.t
implement, but you are given the writer instead of providing a writer and the writer is a
Direct_stream_writer.t instead of a
The main advantage of this interface is that it consumes far less memory per open query.
Though the implementation function is given a writer immediately, the result of the client's call to
dispatch will not be determined until after the implementation function returns. Elements written before the function returns will be queued up to be written after the function returns.
val dispatch : ('query, 'response, 'error) t -> Connection.t -> 'query -> ('response Async_kernel.Pipe.Reader.t * Metadata.t, 'error) Core_kernel.Result.t Core_kernel.Or_error.t Async_kernel.Deferred.t
(..., 'error) Result.t as its return type to represent the possibility of the call itself being somehow erroneous (but understood - the outer
Or_error.t encompasses failures of that nature). Note that this cannot be done simply by making
'response a result type, since
('response Pipe.Reader.t, 'error) Result.t is distinct from
('response, 'error) Result.t Pipe.Reader.t.
Closing the pipe has the effect of calling
module Pipe_message : sig ... end
The input type of the
f passed to
module Pipe_response : sig ... end
The output type of the
f passed to
dispatch_iter. This is analagous to a simple
unit Deferred.t, with
Continue being like
Deferred.unit, but it is made explicit when no waiting should occur.
val dispatch_iter : ('query, 'response, 'error) t -> Connection.t -> 'query -> f:('response Pipe_message.t -> Pipe_response.t) -> (Id.t, 'error) Core_kernel.Result.t Core_kernel.Or_error.t Async_kernel.Deferred.t
dispatch_iter t conn query ~f is similar to calling
dispatch t conn
query and then iterating over the result pipe with
f. The main advantage it offers is that its memory usage is much lower, making it more suitable for situations where many queries are open at once.
f may be fed any number of
Update _ messages, followed by a single
Closed _ message.
f can cause the connection to stop reading messages off of its underlying
Reader.t by returning
Wait _. This is the same as what happens when a client stops reading from the pipe returned by
dispatch when the
dispatch_iter returns an
Id.t after the subscription is started. This may be fed to
abort with the same
Connection.t as the call to
dispatch_iter to cancel the subscription, which will close the pipe on the implementation side. Calling it with a different
Connection.t has undefined behavior.
abort rpc connection id given an RPC and the id returned as part of a call to dispatch, abort requests that the other side of the connection stop sending updates.
If you are using
dispatch rather than
dispatch_iter, you are encouraged to close the pipe you receive rather than calling
abort -- both of these have the same effect.
close_reason metadata will be determined sometime after the pipe associated with
metadata is closed. Its value will indicate what caused the pipe to be closed.
val client_pushes_back : (_, _, _) t -> bool
val name : (_, _, _) t -> string
val version : (_, _, _) t -> int