Class ProtoMessageConsumerBase

Namespace: ProtoTest.Messaging

Assembly: ProtoTest.Messaging.dll

The base every messaging adapter's consumer derives from. It owns the one await queue: waits on a consumer serialize in call order, only deliveries at or after the consumer's position can match, a matched delivery is consumed exactly once, a delivery that matched no awaited predicate stays for a later await, and the source is scanned once more when the wait returns at the deadline, so a delivery assigned at the same instant as the timeout still wins. An adapter implements AwaitAsync(string, Func<ProtoMessage, bool>, TimeSpan, CancellationToken) by resolving its IProtoMessageAwaitSource and calling the protected AwaitAsync(…, sourceFactory) overload: it supplies a source (snapshot + wait), never its own await loop, gate, position or consumed set.

public abstract class ProtoMessageConsumerBase : IProtoMessageConsumer, IAsyncDisposable

Inheritance: object

ProtoMessageConsumerBase

Implements: IProtoMessageConsumer

IAsyncDisposable

Inherited Members: object.Equals(object)

object.Equals(object, object)

object.GetHashCode()

object.GetType()

object.MemberwiseClone()

object.ReferenceEquals(object, object)

object.ToString()

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

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

Constructors

ProtoMessageConsumerBase(long)

Creates the consumer with the first position an await may consume. The queue never matches a source entry before it, so a source the consumer shares with other tests can keep its history.

protected ProtoMessageConsumerBase(long position)

Parameters

position long: The first position an await may consume.

Methods

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

Waits for the first message on destination matching predicate within timeout. Only messages that reach this consumer count, so another test's traffic on a shared broker can never satisfy the await. Awaits on one consumer are serialized in call order, and a delivery that matches no active predicate is not consumed: it stays available to a later await on the same consumer, so concurrent awaits neither lose nor steal each other's messages and every matched message is consumed exactly once.

public abstract ValueTask<ProtoMessage> AwaitAsync(string destination, Func<ProtoMessage, bool> predicate, TimeSpan timeout, CancellationToken cancellationToken = default)

Parameters

destination string
predicate Func<ProtoMessage, bool>
timeout TimeSpan
cancellationToken CancellationToken

Returns

ValueTask<ProtoMessage>

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

Waits through the one await queue: the source factory is resolved inside the queue's await gate, so a consumer that re-baselines while awaits are already queued cannot race an in-flight scan. An adapter's public await resolves its source here.

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

Parameters

destination string: The destination the predicate matches on.
predicate Func<ProtoMessage, bool>: The match the first message at or after the position must satisfy.
timeout TimeSpan: How long the await may take before it fails.
cancellationToken CancellationToken: Cancels the wait and releases the queue.
sourceFactory Func<IProtoMessageAwaitSource>: Resolves the adapter's source for this await.

Returns

ValueTask<ProtoMessage>

Exceptions

TimeoutException: No message matched within timeout.

DisposeAsync()

The default disposal: a no-op, because the adapter owns nothing outside the consumer. An adapter that declared destinations for the test overrides this to remove them.

public virtual ValueTask DisposeAsync()

Returns

ValueTask

MatchRoutingKey(Func<ProtoMessage, bool>, string?)

Composes predicate with an ordinal match of RoutingKey against routingKey; a null routing key returns the predicate unchanged. An adapter whose transport has no routing-key binding supplies the composed predicate through the base's await, so routing-key matching is one rule beside the await machinery rather than one per adapter.

protected static Func<ProtoMessage, bool> MatchRoutingKey(Func<ProtoMessage, bool> predicate, string? routingKey)

Parameters

predicate Func<ProtoMessage, bool>
routingKey string

Returns

Func<ProtoMessage, bool>

PrepareAsync(IReadOnlyCollection<string>, CancellationToken)

The default preparation: a no-op, because the adapter's destinations already exist or its history covers them. An adapter that declares per-test destinations overrides this.

public virtual ValueTask PrepareAsync(IReadOnlyCollection<string> destinations, CancellationToken cancellationToken = default)

Parameters

destinations IReadOnlyCollection<string>
cancellationToken CancellationToken

Returns

ValueTask

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 resets before it scans the new source, whose positions are unrelated to the old one's.

protected void Reset(long position)

Parameters

position long: The first position an await may consume from the new source.