Joseph Doherty
0c6e5b3ace
The gateway emits a ReplayGap sentinel MxEvent at the head of a StreamEvents stream resumed via after_worker_sequence when the requested cursor predates the oldest retained event. Clients previously ignored it, silently mis-treating a lossy resume as continuous. Each client now surfaces the sentinel as a distinct, typed, non-terminal signal (never synthesized, never swallowed) so a consumer can detect the gap and re-snapshot; resume contract is after_worker_sequence = oldest_available_sequence - 1. - .NET: MxEventStreamItem (IsReplayGap/ReplayGap/Event) via new StreamEventItemsAsync + AsStreamItemsAsync extension. Build clean, 87 passed. - Go: EventResult.ReplayGap field + IsReplayGap(); ReplayGap type alias. build/vet/test clean. - Rust: EventItem enum (Event/ReplayGap); EventStream now yields Result<EventItem, Error>; CLI renders REPLAY_GAP line / replayGap JSON. fmt/check/test/clippy clean. - Python: ReplayGap dataclass; stream_events yields pb.MxEvent | ReplayGap. 131 passed. - Shared docs: ClientLibrariesDesign non-goals reframed (reconnect-replay protocol is consumable; auto-reconnect stays a non-goal); CrossLanguageSmokeMatrix resume-gap note. Java client is deferred to the windev batch (no local JRE); CLI-15 stays open until it lands. Claude-Session: https://claude.ai/code/session_01DMXXvNuPekkkrTEyPNxEkW
83 lines
3.6 KiB
C#
83 lines
3.6 KiB
C#
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
|
|
|
namespace ZB.MOM.WW.MxGateway.Client;
|
|
|
|
/// <summary>
|
|
/// One item yielded by the typed event stream. It is either a normal MXAccess
|
|
/// <see cref="MxEvent"/> or a reconnect-replay <em>gap sentinel</em>.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// <para>
|
|
/// The gateway emits a single sentinel <see cref="MxEvent"/> at the head of a
|
|
/// <c>StreamEvents</c> stream that was resumed via
|
|
/// <c>StreamEventsRequest.after_worker_sequence</c> when the requested sequence
|
|
/// predates the oldest event still retained in the session replay ring — i.e.
|
|
/// events were evicted and cannot be replayed. On that sentinel the
|
|
/// <c>MxEvent.replay_gap</c> field is set, <see cref="MxEvent.Family"/> is
|
|
/// <see cref="MxEventFamily.Unspecified"/>, the <c>body</c> oneof is unset, and
|
|
/// no per-item fields are populated.
|
|
/// </para>
|
|
/// <para>
|
|
/// This wrapper makes that sentinel observable instead of forcing the consumer
|
|
/// to inspect the raw <see cref="MxEvent"/>. It never synthesizes or swallows an
|
|
/// event: the gap is exactly the gateway's own sentinel, exposed with
|
|
/// <see cref="IsReplayGap"/> set and <see cref="ReplayGap"/> populated. Every
|
|
/// other event flows through unchanged with <see cref="IsReplayGap"/> false.
|
|
/// </para>
|
|
/// <para>
|
|
/// When <see cref="IsReplayGap"/> is <see langword="true"/> the consumer has
|
|
/// missed events and MUST discard local state and re-snapshot. To resume without
|
|
/// incurring another gap, reconnect with
|
|
/// <c>after_worker_sequence = ReplayGap.OldestAvailableSequence - 1</c>.
|
|
/// </para>
|
|
/// </remarks>
|
|
public sealed class MxEventStreamItem
|
|
{
|
|
private MxEventStreamItem(MxEvent gatewayEvent, ReplayGap? replayGap)
|
|
{
|
|
Event = gatewayEvent;
|
|
ReplayGap = replayGap;
|
|
}
|
|
|
|
/// <summary>
|
|
/// The underlying raw <see cref="MxEvent"/>. For a normal event this is the
|
|
/// MXAccess event itself; for a replay-gap item this is the gateway's
|
|
/// sentinel event whose only meaningful payload is <see cref="ReplayGap"/>.
|
|
/// Never <see langword="null"/>.
|
|
/// </summary>
|
|
public MxEvent Event { get; }
|
|
|
|
/// <summary>
|
|
/// The reconnect-replay gap payload when this item is a gap sentinel;
|
|
/// otherwise <see langword="null"/>. Read
|
|
/// <see cref="Contracts.Proto.ReplayGap.RequestedAfterSequence"/> and
|
|
/// <see cref="Contracts.Proto.ReplayGap.OldestAvailableSequence"/> to learn
|
|
/// which events were lost.
|
|
/// </summary>
|
|
public ReplayGap? ReplayGap { get; }
|
|
|
|
/// <summary>
|
|
/// <see langword="true"/> when this item is a reconnect-replay gap sentinel:
|
|
/// the consumer missed events and must discard local state and re-snapshot.
|
|
/// Resume without another gap by reconnecting with
|
|
/// <c>after_worker_sequence = ReplayGap.OldestAvailableSequence - 1</c>.
|
|
/// </summary>
|
|
public bool IsReplayGap => ReplayGap is not null;
|
|
|
|
/// <summary>
|
|
/// Wraps a raw stream <see cref="MxEvent"/> as a typed item, classifying it
|
|
/// as a replay-gap sentinel when <c>MxEvent.replay_gap</c> is present.
|
|
/// </summary>
|
|
/// <param name="gatewayEvent">The raw event from the <c>StreamEvents</c> stream.</param>
|
|
/// <returns>A typed item exposing either the normal event or the replay gap.</returns>
|
|
internal static MxEventStreamItem From(MxEvent gatewayEvent)
|
|
{
|
|
ArgumentNullException.ThrowIfNull(gatewayEvent);
|
|
|
|
// For a message-typed proto3 field, presence is a non-null reference.
|
|
return gatewayEvent.ReplayGap is { } replayGap
|
|
? new MxEventStreamItem(gatewayEvent, replayGap)
|
|
: new MxEventStreamItem(gatewayEvent, replayGap: null);
|
|
}
|
|
}
|