Class FileOutboxStore

Namespace

Xapi.Client.Outbox

Assembly

Xapi.Client.Outbox.dll

An OutboxStore that keeps the outbox in a directory, as an append-only journal that is flushed to disk (fsync) before each change is acknowledged.

public sealed class FileOutboxStore : OutboxStore, IDisposable

Inheritance

object

OutboxStore

FileOutboxStore

Implements

IDisposable

Inherited Members

object.Equals(object)

object.Equals(object, object)

object.GetHashCode()

object.GetType()

object.ReferenceEquals(object, object)

object.ToString()

Remarks

The directory holds outbox.v1.jsonl, the journal: one JSON object per line, each with a CRC-32 of its record, and outbox.lock, held open while the store is open so that a second store (in this or another process) cannot use the same directory. When most of the journal describes Statements that are gone, it is compacted: the live state is written to outbox.v1.jsonl.tmp, flushed, and renamed over the journal.

If the process dies in the middle of a write, the journal ends with an incomplete line; the next open cuts it off. That write was never acknowledged, so nothing reported as stored is lost. A rename interrupted by a crash leaves either the old or the new journal, each complete. Lines damaged on disk are skipped, and a copy of the damaged file is kept; see Recovery.

The whole queue is also kept in memory, and the journal is read once, by OpenAsync(string, CancellationToken). The store suits queues of up to tens of thousands of Statements. All methods are thread-safe; they complete synchronously, blocking on the disk flush.

Properties

Directory

The full path of the directory.

public string Directory { get; }

Property Value

string

Recovery

What had to be repaired when the journal was opened, or null when it was intact.

public FileOutboxRecovery? Recovery { get; }

Property Value

FileOutboxRecovery

Methods

AddAsync(OutboxEntry, CancellationToken)

Adds an entry at the end of the queue, unless an entry with the same Statement id is pending.

public override ValueTask<bool> AddAsync(OutboxEntry entry, CancellationToken cancellationToken)

Parameters

entry OutboxEntry

The entry.

cancellationToken CancellationToken

Cancels the operation.

Returns

ValueTask<bool>

true when the entry was added; false when its id was already pending.

Dispose()

Closes the journal and releases the directory lock.

public void Dispose()

GetDeadLetterCountAsync(CancellationToken)

Returns the number of dead letters.

public override ValueTask<int> GetDeadLetterCountAsync(CancellationToken cancellationToken)

Parameters

cancellationToken CancellationToken

Cancels the operation.

Returns

ValueTask<int>

The number of dead letters.

GetDeadLettersAsync(CancellationToken)

Returns the dead letters, oldest first.

public override ValueTask<IReadOnlyList<OutboxDeadLetter>> GetDeadLettersAsync(CancellationToken cancellationToken)

Parameters

cancellationToken CancellationToken

Cancels the operation.

Returns

ValueTask<IReadOnlyList<OutboxDeadLetter>>

The dead letters.

GetPendingAsync(int, CancellationToken)

Returns the oldest pending entries, in queue order, without removing them.

public override ValueTask<IReadOnlyList<OutboxEntry>> GetPendingAsync(int maxCount, CancellationToken cancellationToken)

Parameters

maxCount int

The largest number of entries to return.

cancellationToken CancellationToken

Cancels the operation.

Returns

ValueTask<IReadOnlyList<OutboxEntry>>

Up to maxCount entries; empty when nothing is pending.

GetPendingCountAsync(CancellationToken)

Returns the number of pending entries.

public override ValueTask<int> GetPendingCountAsync(CancellationToken cancellationToken)

Parameters

cancellationToken CancellationToken

Cancels the operation.

Returns

ValueTask<int>

The number of pending entries.

MoveToDeadLettersAsync(OutboxDeadLetter, CancellationToken)

Removes the pending entry with the dead letter's Statement id, if there is one, and stores the dead letter, replacing any dead letter with the same id.

public override ValueTask MoveToDeadLettersAsync(OutboxDeadLetter deadLetter, CancellationToken cancellationToken)

Parameters

deadLetter OutboxDeadLetter

The dead letter.

cancellationToken CancellationToken

Cancels the operation.

Returns

ValueTask

A task that completes when the change is durable.

OpenAsync(string, CancellationToken)

Opens the store in a directory, creating the directory if needed, and reads the journal.

public static Task<FileOutboxStore> OpenAsync(string directory, CancellationToken cancellationToken = default)

Parameters

directory string

The directory. It should be used by this store only.

cancellationToken CancellationToken

Cancels reading the journal.

Returns

Task<FileOutboxStore>

The open store. Dispose it to release the directory.

Exceptions

IOException

Another FileOutboxStore has the directory open.

RecordFailedAttemptAsync(IReadOnlyCollection<Guid>, string, CancellationToken)

Records a failed send of pending entries: increments Attempts and sets LastError. Ids that are not pending are ignored.

public override ValueTask RecordFailedAttemptAsync(IReadOnlyCollection<Guid> statementIds, string error, CancellationToken cancellationToken)

Parameters

statementIds IReadOnlyCollection<Guid>

The Statement ids.

error string

A description of the error.

cancellationToken CancellationToken

Cancels the operation.

Returns

ValueTask

A task that completes when the change is durable.

RemoveAsync(IReadOnlyCollection<Guid>, CancellationToken)

Removes delivered entries. Ids that are not pending are ignored.

public override ValueTask RemoveAsync(IReadOnlyCollection<Guid> statementIds, CancellationToken cancellationToken)

Parameters

statementIds IReadOnlyCollection<Guid>

The Statement ids.

cancellationToken CancellationToken

Cancels the operation.

Returns

ValueTask

A task that completes when the removal is durable.

RemoveDeadLetterAsync(Guid, CancellationToken)

Deletes a dead letter.

public override ValueTask<bool> RemoveDeadLetterAsync(Guid statementId, CancellationToken cancellationToken)

Parameters

statementId Guid

The Statement id.

cancellationToken CancellationToken

Cancels the operation.

Returns

ValueTask<bool>

false when there is no dead letter with that id.

RequeueAsync(Guid, CancellationToken)

Moves a dead letter back to the end of the queue as a pending entry with no failed attempts. If an entry with the same id is already pending, the dead letter is only removed.

public override ValueTask<bool> RequeueAsync(Guid statementId, CancellationToken cancellationToken)

Parameters

statementId Guid

The Statement id.

cancellationToken CancellationToken

Cancels the operation.

Returns

ValueTask<bool>

false when there is no dead letter with that id.