Class ProtoMessageAwaitQueue

Namespace

ProtoTest.Messaging

Assembly

ProtoTest.Messaging.dll

One consumer's await machinery, shared by the in-memory, RabbitMQ and MassTransit adapters: awaits on the consumer serialize in call order, only messages at or after the consumer's position can match, a matched message is consumed exactly once per destination, and a message that matched no awaited predicate stays in the source for a later await. Consumption is scoped by destination: each destination's deliveries form their own position space, so a position alone does not identify a delivery across destinations. The scan runs before every wait and again when the wait returns at the deadline, so a delivery assigned at the same instant as the timeout still wins. An adapter contributes an IProtoMessageAwaitSource over its own storage and owns one queue per consumer; it never implements its own await loop, gate, position or consumed set.

public sealed class ProtoMessageAwaitQueue

Inheritance

object

ProtoMessageAwaitQueue

Inherited Members

object.Equals(object)

object.Equals(object, object)

object.GetHashCode()

object.GetType()

object.ReferenceEquals(object, object)

object.ToString()

Extension Methods

SheetModelAssertions.ShouldMatchShape<TRow>(TRow, object, bool, JsonSerializerOptions?)

SheetModelAssertions.ShouldMatchShape<TRow>(TRow, object, JsonSerializerOptions?)

Constructors

ProtoMessageAwaitQueue(long)

One consumer's await machinery, shared by the in-memory, RabbitMQ and MassTransit adapters: awaits on the consumer serialize in call order, only messages at or after the consumer's position can match, a matched message is consumed exactly once per destination, and a message that matched no awaited predicate stays in the source for a later await. Consumption is scoped by destination: each destination's deliveries form their own position space, so a position alone does not identify a delivery across destinations. The scan runs before every wait and again when the wait returns at the deadline, so a delivery assigned at the same instant as the timeout still wins. An adapter contributes an IProtoMessageAwaitSource over its own storage and owns one queue per consumer; it never implements its own await loop, gate, position or consumed set.

public ProtoMessageAwaitQueue(long position)

Parameters

position long

The first position an await may consume, on every destination; a source's older messages can never match.

Methods

AwaitAsync(string, Func<ProtoMessage, bool>, TimeSpan, CancellationToken, Func<IProtoMessageAwaitSource>)

Waits for the first message on destination that matches predicate within timeout. The source is resolved inside the await gate, so a consumer that re-baselines while awaits are already queued cannot race an in-flight scan.

public ValueTask<ProtoMessage> AwaitAsync(string destination, Func<ProtoMessage, bool> predicate, TimeSpan timeout, CancellationToken cancellationToken, Func<IProtoMessageAwaitSource> sourceFactory)

Parameters

destination string

predicate Func<ProtoMessage, bool>

timeout TimeSpan

cancellationToken CancellationToken

sourceFactory Func<IProtoMessageAwaitSource>

Returns

ValueTask<ProtoMessage>

Exceptions

TimeoutException

No message matched within the timeout.

Reset(long)

Re-baselines the queue: position replaces the current position and the consumed set is dropped. A consumer whose source is replaced mid-run - the MassTransit bridge when a substitution gives the test its own harness, whose history starts empty - resets before it scans the new source.

public void Reset(long position)

Parameters

position long