A single-consumer single-producer queue on a block device


module Log : S.LOG
module Block : S.BLOCK
module Time : S.TIME
module Clock : S.CLOCK


type t

A journal kept on disk of work we need to perform

type error = [
| `Msg of string
type 'a result = ( 'a, error ) Result.result
val pp_error : Format.formatter -> error -> unit
val open_error : 'a result -> ( 'a, [> error ] ) Result.result
val error_to_msg : 'a result -> ( 'a, S.msg ) Result.result
val start : ?name:string -> ?client:string -> ?flush_interval:int64 -> ?retry_interval:int64 -> Block.t -> ( Operation.t list -> unit result Lwt.t ) -> t result Lwt.t

Start a journal replay thread on a given disk, with the given processing function which will be applied at-least-once to every item in the journal. If a flush_interval is provided then every push will start a timer and the items will not be processed until the timer expires (or the journal becomes full) to encourage batching. The retry_interval gives the delay between re'perform'ing journalled items that fail. Default is 5 seconds.

val shutdown : t -> unit Lwt.t

Shut down a journal replay thread

type waiter = {
flush : unit -> unit;
sync : unit -> unit Lwt.t;
val push : t -> Operation.t -> waiter result Lwt.t

Append an operation to the journal. When this returns, the operation will be performed at-least-once before any later items are performed. If a client needs to wait for the operation to be completed then call the returned thunk and wait for the resulting thread.