| Safe Haskell | Safe-Inferred |
|---|---|
| Language | GHC2021 |
Control.Concurrent.PersistentQueue
Description
A crash-safe, file-backed, bounded FIFO queue.
Each item is written to its own file (named by a monotonic index) before it
enters the in-memory TBQueue, and removed from disk only after it has been
popped, so a restart reloads exactly the items that were enqueued but not yet
consumed (at-least-once delivery). Items carry their CBOR serialization,
produced once on write and reused, and can be drained in byte- and
count-bounded batches with an in-flight pin for safe retries.
Synopsis
- data PersistentQueue m a
- data PersistentQueueLog
- = PersistentQueueLoadFailed {
- reason :: Text
- | PersistentQueueFull
- | PersistentQueueDeleteFailed { }
- = PersistentQueueLoadFailed {
- newPersistentQueue :: (MonadLabelledSTM m, MonadIO m, FromCBOR a, MonadCatch m, MonadFail m) => Tracer IO PersistentQueueLog -> FilePath -> Natural -> m (PersistentQueue m a)
- writePersistentQueue :: (ToCBOR a, MonadSTM m, MonadIO m) => Tracer IO PersistentQueueLog -> PersistentQueue m a -> a -> m ()
- peekPersistentQueue :: MonadSTM m => PersistentQueue m a -> m a
- tryPeekPersistentQueue :: MonadSTM m => PersistentQueue m a -> m (Maybe a)
- peekBatchPersistentQueue :: MonadSTM m => PersistentQueue m a -> Int -> Int -> m [(a, ByteString)]
- popBatchPersistentQueue :: (MonadSTM m, MonadIO m) => Tracer IO PersistentQueueLog -> PersistentQueue m a -> [(a, ByteString)] -> m ()
- nextPendingBatch :: MonadSTM m => TVar m (Maybe [(a, ByteString)]) -> PersistentQueue m a -> Int -> Int -> m (Maybe [(a, ByteString)])
Documentation
data PersistentQueue m a Source #
Queue elements carry the item's CBOR serialization, produced once on write and reused for both the on-disk file and downstream consumers, so consuming does not serialize twice.
data PersistentQueueLog Source #
Events emitted while operating the queue.
Constructors
| PersistentQueueLoadFailed | |
Fields
| |
| PersistentQueueFull | |
| PersistentQueueDeleteFailed | |
Instances
| Show PersistentQueueLog Source # | |
Defined in Control.Concurrent.PersistentQueue | |
| Eq PersistentQueueLog Source # | |
Defined in Control.Concurrent.PersistentQueue Methods (==) :: PersistentQueueLog -> PersistentQueueLog -> Bool Source # (/=) :: PersistentQueueLog -> PersistentQueueLog -> Bool Source # | |
newPersistentQueue :: (MonadLabelledSTM m, MonadIO m, FromCBOR a, MonadCatch m, MonadFail m) => Tracer IO PersistentQueueLog -> FilePath -> Natural -> m (PersistentQueue m a) Source #
Create a new persistent queue at file path and given capacity.
writePersistentQueue :: (ToCBOR a, MonadSTM m, MonadIO m) => Tracer IO PersistentQueueLog -> PersistentQueue m a -> a -> m () Source #
Write a value to the queue, blocking if the queue is full.
peekPersistentQueue :: MonadSTM m => PersistentQueue m a -> m a Source #
Get the next value from the queue without removing it, blocking if the queue is empty.
tryPeekPersistentQueue :: MonadSTM m => PersistentQueue m a -> m (Maybe a) Source #
Like peekPersistentQueue, but returns Nothing instead of blocking
when the queue is empty.
peekBatchPersistentQueue :: MonadSTM m => PersistentQueue m a -> Int -> Int -> m [(a, ByteString)] Source #
Get all pending values and their serializations, up to the given count
and total byte limits, blocking until at least one is available. Values
are not removed; use popBatchPersistentQueue after they were sent.
popBatchPersistentQueue :: (MonadSTM m, MonadIO m) => Tracer IO PersistentQueueLog -> PersistentQueue m a -> [(a, ByteString)] -> m () Source #
Remove a batch previously returned by peekBatchPersistentQueue. Pops
unconditionally, one item per batch entry: the caller is expected to be the
sole consumer, so the queue head still holds exactly the peeked items (an
item-matching guard could only silently no-op and wedge the queue).
nextPendingBatch :: MonadSTM m => TVar m (Maybe [(a, ByteString)]) -> PersistentQueue m a -> Int -> Int -> m (Maybe [(a, ByteString)]) Source #
Get the batch to broadcast next: the batch already in flight if there is
one, otherwise a fresh one peeked from the queue (and recorded as in
flight). Returns Nothing when nothing is pending. The caller must clear
the in-flight var after popping a successfully sent batch.
Pinning the in-flight batch across transient retries matters for consumers with an at-least-once send that can commit while the caller sees a transient failure: the retry must send (and afterwards pop) exactly the content of the committed attempt. Re-peeking on retry could pick up messages enqueued in the meantime, and a dedup that declared the grown batch delivered would pop and lose the never-sent tail.