追溯事件

範例：使用倒轉的追溯貨物（C#）

我用一個範例說明了大多數關於事件溯源的範例，這個範例是基於將貨物在港口間透過船隻運送。我將繼續使用那個範例，這次來看看如何使用追溯事件。

首先，我們需要針對如何執行追溯事件做出一些策略性決策。對於這個範例，我將使用透過倒轉即時模型來達到分支點的方法。由於這個範例很簡單，我不會使用任何選擇性重播，而只會倒轉完整的事件清單。

由於我們使用的是倒轉，因此所有事件都必須是可逆的。我不會深入探討事件處理或反轉的詳細資訊，請參閱事件溯源中的範例，了解其運作方式。

追溯事件行為是由一種特殊類型的事件觸發的。

類別 ReplacementEvent...

  private DomainEvent original, replacement;
  public DomainEvent Original {get { return original; }}
  public DomainEvent Replacement {get { return replacement; }}

  public ReplacementEvent(DomainEvent oldEvent, DomainEvent replacement)
    : base(oldEvent.Occurred) {
    this.original = oldEvent;
    this.replacement = replacement;
  }
  internal override void Process() {
    throw new Exception("Replacements should not be processed directly");
  }
  internal override void Reverse() {
    throw new Exception("Cannot reverse replacements");
  }

你會注意到，在這些情況下，我已封鎖基本流程和反轉方法。在這些範例中，我大多數時候都偏好讓事件處理自己的處理邏輯。替換不同：它們不包含任何網域邏輯，而且其處理涉及事件佇列的知識以及對該事件佇列的處理。與事件佇列的密切互動是我希望僅留給事件處理器處理的事情，因此事實上，我將所有替換行為都建置到事件佇列中。

類別 EventProcessor...

  public void Process(DomainEvent ev) {
    try {
      if (ev is ReplacementEvent)
        ProcessReplacement((ReplacementEvent) ev);
      else if (OutOfOrder(ev))
        ProcessOutOfOrder(ev);
      else BasicProcessEvent(ev);
    } catch (Exception ex) {
      ev.ProcessingError = ex;
      if (ShouldRethrowExceptions) throw ex;
    }
    InsertToLog(ev);
  }

您會在此處注意到，我正在執行物件導向程式設計的重大錯誤：根據方法引數的類型，明確條件式行為。我並不常這樣做，但在此處執行是因為我希望事件處理器將佇列的知識保留給自己。（我也可以使用雙重分派，但此處的情況似乎並不複雜到需要使用它。）

您可能也會注意到，我沒有針對拒絕的事件建立案例。正如我們將看到的，我將這些事件視為取代事件為 null 的取代事件來處理。

我將從最簡單的案例開始。基本的 Process（和反向）方法只是一個簡單的包裝器，允許我在需要時加入一些追蹤行為。

類別 EventProcessor...

  private void BasicProcessEvent(DomainEvent e) {
    traceProcess(e);
    e.Process();
  }
  private void BasicReverseEvent(DomainEvent e) {
    traceReverse(e);
    e.Reverse();
  }

無序事件是最容易首先描述的。這些只是我們以非順序接收的常規事件，我們不會使用取代事件來建模它們。基本上，這些會在串流中插入一個新事件。處理器透過將無序事件與最後一個事件進行比較來測試它。

類別 EventProcessor...

  private bool OutOfOrder(DomainEvent e) {
    if (LogIsEmpty()) return false;
    return (LastEvent().after(e));
  }
  private DomainEvent LastEvent() {
    if (LogIsEmpty()) return null;
    return (DomainEvent) log[log.Count - 1];
  }
  private bool LogIsEmpty() {
    return 0 == log.Count;
  }

為了讓這個範例好笑地簡單，我只按發生日期對事件進行排序。要實際執行此操作，可能只需轉換為更精細的時間點解析度就足夠了。有時其他因素可能會影響排序。

處理無序案例的步驟很簡單。

類別 EventProcessor...

  private void ProcessOutOfOrder(DomainEvent e) {
    RewindTo(e);
    BasicProcessEvent(e);
    ReplayAfter(e);
  }

要倒轉，我只會選取日誌中晚於無序事件的所有事件。同樣，為了簡化，我將日誌（或至少快取）保存在記憶體中。

類別 EventProcessor...

  private void RewindTo(DomainEvent priorEvent) {
    IList consequences = Consequences(priorEvent);
    for (int i = consequences.Count - 1; i >= 0; i--)
      BasicReverseEvent(((DomainEvent) consequences[i]));
  }
  private IList Consequences(DomainEvent baseEvent) {
    IList result = new ArrayList();
    foreach (DomainEvent candidate in log)
      if (candidate.IsConsequenceOf(baseEvent)) result.Add(candidate);
    return result;
  }

class DomainEvent...

  public bool IsConsequenceOf(DomainEvent other) {
    return (!ShouldIgnoreOnReplay && this.after(other));
  }

如您所見，並非所有事件都選取在倒轉期間重新處理，儘管我沒有使用任何選擇性重播。基本上，我不會重新處理錯誤事件或已被拒絕的事件。

再次向前重播事件很簡單

類別 EventProcessor...

  private void ReplayAfter(DomainEvent ev) {
    foreach (DomainEvent e in Consequences(ev)) BasicProcessEvent(e);
  }

現在讓我們繼續進行取代。

類別 EventProcessor...

  private void ProcessReplacement(ReplacementEvent e) {
    if (e.Original.ShouldIgnoreOnReplay)
      throw new ProcessingException("Cannot replace event twice");
    else if (null == e.Replacement) 
      ProcessRejection(e);
    else if (e.HasPriorReplacement) 
      ProcessPriorReplacement(e);
    else 
      ProcessPriorOriginal(e);
    
  }

類別 ReplacementEvent...

  public bool HasPriorReplacement {
    get {
      if (null == replacement) return false;
      else return original.after(replacement);
    }
  }

此處有幾個案例要處理。如果原始事件已標記為忽略，表示有處理錯誤，因為我們不應該拒絕已拒絕的事件。完成此操作後，我們有三個主要案例：取代為拒絕、取代事件早於原始事件，以及原始事件早於取代事件。

讓我們從拒絕案例開始，它是由一個空值替換事件指示的。我們在此倒帶到被拒絕的事件，拒絕它，反轉它，並向前重播。

類別 EventProcessor...

  private void ProcessRejection(ReplacementEvent e) {
    RewindTo(e.Original);
    BasicReverseEvent(e.Original);
    e.Original.Reject();
    ReplayAfter(e.Original);
  }

class DomainEvent...

  public bool after (DomainEvent other) {
    return this.CompareTo(other) > 0;
  }
  public void Reject() {
    _isRejected = true;
  }
  private bool _isRejected;
  public virtual bool ShouldIgnoreOnReplay {
    get {
      if (WasProcessingError) return true;
      return _isRejected;
    }
  }

拒絕會標記事件，因此它不會再次被處理或倒帶。

如果我們有先前的替換，我們會倒帶到替換，拒絕原始事件，處理替換並向前重播。

類別 EventProcessor...

  private void ProcessPriorReplacement(ReplacementEvent e) {
    RewindTo(e.Replacement);
    e.Original.Reject();
    BasicProcessEvent(e.Replacement);
    ReplayAfter(e.Replacement);
  }

對於先前的原始事件，我們會倒帶到原始事件，反轉並拒絕它，重播到替換，處理它，並繼續向前重播。

類別 EventProcessor...

  private void ProcessPriorOriginal(ReplacementEvent e) {
    RewindTo(e.Original);
    BasicReverseEvent(e.Original);
    e.Original.Reject();
    ReplayBetween(e.Original, e.Replacement);
    BasicProcessEvent(e.Replacement);
    ReplayAfter(e.Replacement);
  }
  private void ReplayBetween(DomainEvent first, DomainEvent last) {
    IList eventsToReplay = new ArrayList();
    foreach (DomainEvent e in log) {
      if (e.IsConsequenceOf(first) && last.after(e))
        eventsToReplay.Add(e);
    }
    foreach (DomainEvent e in eventsToReplay)
      BasicProcessEvent(e);
  }

我遵循此處的慣例，即在成功處理事件之前不將其新增到記錄檔。如果記錄檔與應用程式狀態不在同一個交易中，這會很有用。如果它們全部都在一個交易中，我可以在到達分支點並向前重播（這將選取新的事件）後，將事件新增到記錄檔中。

範例：更新外部系統（C#）

對於簡單的事件溯源，在重播期間關閉外部通知就足夠了。然而，對於追溯事件，我們需要進一步進行偵測和修正。為在倒帶/重播期間執行此操作，我們將儲存所有產生的事件，並進行比較以確定需要修正的內容。然後，我們將假設一個簡單的自動修正案例，即傳送取消訊息和變更的新訊息。

我將繼續使用運送範例，說明美國的海關當局需要在任何通過加拿大進入美國港口的貨物時收到通知。

當貨物物件處理抵達事件時，會進行外部通知。

class Cargo...

  public void HandleArrival(ArrivalEvent ev) {
    ev.priorCargoInCanada[this] = _hasBeenInCanada;
    if ("CA" == ev.Port.Country) 
      _hasBeenInCanada = true;
    if (HasBeenInCanada && "US" == ev.Port.Country) {
      Registry.CustomsGateway.Notify(ev.Occurred, ev.Ship, ev.Port);
      ev.WasNotificationSent = true;
    }
  }
  private bool _hasBeenInCanada = false;
  public bool HasBeenInCanada {get { return _hasBeenInCanada;}}

為了反轉此事件，我們會在正向播放中記錄是否已傳送通知，並在傳送時再次傳送。

class Cargo...

  public void ReverseArrival(ArrivalEvent ev) {
    _hasBeenInCanada = (bool) ev.priorCargoInCanada[this];
    if (ev.WasNotificationSent) 
      Registry.CustomsGateway.Notify(ev.Occurred, ev.Ship, ev.Port);
  }

我遵循的原則是，網域模型應該不了解事件處理中的重播邏輯。它知道如何為每個事件反轉其自己的狀態，但不在乎與外部系統對話的複雜性。

與外部系統對話的邏輯由閘道處理，在這種情況下，它是一群物件。

圖 1：一群物件組成閘道來處理追溯活動。

在此範例中，CustomsGatewayFront 是將面向網域的介面（由 ICustomsGateway 定義）轉換為實際訊息基礎架構的正常閘道。我們可以安全地忽略它執行的動作，我們假設如果我們呼叫它上面的方法，它會確保海關收到訊息。

有趣的事情發生在之前，由 CustomsGatewayBuffer 驅動，它包裝實際的 CustomsGatewayFront，實作相同的介面，但新增處理追溯重繞的能力。

因為許多追溯重繞是通用的，我可以將通用行為放入不同的類別 ReplayBuffer 中，讓 Customs Gateway Buffer 僅處理海關案例中特定的事項。重播緩衝區需要與事件處理器進行一些通訊，以處理重繞/重播程序中的各種階段。它也需要一些方式來調整最後不匹配的事件，這項工作在此案例中我指派給海關閘道緩衝區。在這兩種情況下，重繞緩衝區透過協作介面進行通訊，以保持其通用純度。

因此，這就是戲劇人物，現在讓我們進入動作。當我們執行時，我將從正常案例開始。海關閘道緩衝區透過建立更新事件並將其傳送至重播緩衝區來實作通知操作。

類別 CustomsGatewayBuffer...

  public void Notify (DateTime arrivalDate, Ship ship, Port port) {
    CustomsNotificationEvent ev =
        new CustomsNotificationEvent(gateway, arrivalDate, ship, port);
    buffer.Send(ev);
  }
  ICustomsGateway gateway;
  ReplayBuffer buffer;

類別 CustomsNotificationEvent...

  public CustomsNotificationEvent(
    ICustomsGateway gateway, DateTime arrivalDate, 
    Ship ship, Port port) 
  {
    this.gateway = gateway;
    this.arrivalDate = arrivalDate;
    this.ship = ship;
    this.port = port;
  }

如果重播緩衝區處於活動狀態，則它會處理事件，這會導致呼叫實際閘道前端。

類別 ReplayBuffer...

  internal void Send(IGatewayEvent ev) {
    current.Add(ev);
    if (isActive) 
      ev.Process();
  }

類別 CustomsNotificationEvent...

  public virtual void Process() {
    gateway.Notify(arrivalDate, ship, port) ;
  }

（我將在稍後說明重播緩衝區如何變得活躍，以及目前的清單是什麼。）

因此，如您所見，對於網域事件的活動處理，所發生的一切就是閘道緩衝區建立閘道事件，將其傳送至處理它的重播緩衝區，這會導致將原始呼叫套用至實際閘道。所有這些都是簡單委派的一種極為複雜的形式。

處理追溯事件時，這種怪異的回報會出現。重播緩衝區透過可重繞介面連接到事件處理器。

類別 ReplayBuffer...

  IRewindable eventProcessor;

internal interface IRewindable  {
  event EventProcessor.EventHandler RewindStarted, RewindFinished, ReplayFinished;    
}

此介面定義三個重要的事件。事件處理器在處理各種案例時會發出這些事件。以下是無序案例

類別 EventProcessor...

  private void ProcessOutOfOrder(DomainEvent e) {
    RewindStarted();
    RewindTo(e);
    RewindFinished();
    BasicProcessEvent(e);
    ReplayAfter(e);
    ReplayFinished();
  }

使用替換時，整體開始和結束是共通的，但每個案例需要特定放置來指出重繞何時結束。

類別 EventProcessor...

  private void ProcessReplacement(ReplacementEvent e) {
    RewindStarted();
    if (e.Original.ShouldIgnoreOnReplay)
      throw new ProcessingException("Cannot replace event twice");
    else if (null == e.Replacement) 
      ProcessRejection(e);
    else if (e.HasPriorReplacement) 
      ProcessPriorReplacement(e);
    else 
      ProcessPriorOriginal(e);
    ReplayFinished();
  }

private void ProcessRejection(ReplacementEvent e) {
  RewindTo(e.Original);
  BasicReverseEvent(e.Original);
  RewindFinished();
  e.Original.Reject();
  ReplayAfter(e.Original);
}

private void ProcessPriorOriginal(ReplacementEvent e) {
  RewindTo(e.Original);
  BasicReverseEvent(e.Original);
  RewindFinished();
  e.Original.Reject();
  ReplayBetween(e.Original, e.Replacement);
  BasicProcessEvent(e.Replacement);
  ReplayAfter(e.Replacement);
}

private void ProcessPriorReplacement(ReplacementEvent e) {
  RewindTo(e.Replacement);
  RewindFinished();
  e.Original.Reject();
  BasicProcessEvent(e.Replacement);
  ReplayAfter(e.Replacement);
}

這些事件提供重播緩衝區正確的計時資訊，以正確處理追溯處理。我在這裡使用事件，因為任何事件處理器都可以有任意數量的重播緩衝區，無需知道這些物件，只要知道它們可以瞭解這三個事件即可。

現在讓我們轉到重播緩衝區，看看它如何處理這些。重播緩衝區是使用一堆清單建立的。

類別 ReplayBuffer...

  internal ReplayBuffer (IRewindable processor, IAdjustable adjuster) {
    this.eventProcessor = processor;
    this.adjuster = adjuster;
    SubscribeToProcessorEvents();
    sent = new ArrayList();
    isActive = true;
    current = sent;
  }
  IList sent,rewound, replayed, current;
  IAdjustable adjuster;
  private bool isActive;
  private void SubscribeToProcessorEvents() {
    eventProcessor.RewindStarted += 
      new EventProcessor.EventHandler(processor_RewindStarted);
    eventProcessor.RewindFinished += 
      new EventProcessor.EventHandler(processor_RewindFinished);
    eventProcessor.ReplayFinished += 
      new EventProcessor.EventHandler(processor_ReplayFinished);
  }

建立緩衝區時，緩衝區是處於活動狀態，表示它會處理傳送給它的任何事件。目前的清單變數會設定為三個核心清單之一，視我們在倒帶的哪個位置而定 - 當處於活動狀態時，它會指向傳送的清單。這讓我們記錄所有實際傳送的事件。它也會訂閱事件處理器上的所有相關事件。（現在不用擔心調整器，我們會在修正中處理它。）

在倒帶的開始，緩衝區會讓自己處於非活動狀態，以避免將事件傳遞到真實閘道，它也會交換目前的清單變數，指向一個新的清單，以便它可以擷取所有倒帶事件。

類別 ReplayBuffer...

  private void processor_RewindStarted() {
    isActive = false;
    rewound = new ArrayList();
    current = rewound;
  }

有了這個範例，你可能可以找出倒帶停止時會發生什麼事，但以防萬一...

類別 ReplayBuffer...

  private void processor_RewindFinished() {
    replayed = new ArrayList();
    current = replayed;
  }

所以到目前為止，很明顯最後一個事件會是最有趣的，因為我們需要在這裡計算不匹配清單。

類別 ReplayBuffer...

  private void processor_ReplayFinished() {
    current = sent;
    DetermineChange();
    isActive = true;
    rewound = null;
    replayed = null;
  }
  private void DetermineChange() {
    IList matchingEvents = new ArrayList();
    foreach (IGatewayEvent ev in replayed) {
      if (rewound.Contains(ev)) matchingEvents.Add(ev);
    }
    foreach (IGatewayEvent ev in matchingEvents) {
      replayed.Remove(ev);
      rewound.Remove(ev);
    }
    adjuster.Adjust(rewound, replayed);
  }

演算法其實很簡單，我們只要找出哪些事件相符，然後將它們從兩個清單中移除。完成後，我們會有兩個不匹配事件的清單，我們會將它們傳遞給調整器處理。

類別 CustomsGatewayBuffer…

  public void Adjust(IList oldEvents, IList newEvents) {
    foreach (CustomsNotificationEvent e in oldEvents)
      new CustomsCancellationEvent(e).Process();
    foreach (CustomsNotificationEvent e in newEvents)
      e.Process();
  }
class CustomsCancellationEvent {
  private CustomsNotificationEvent original;
  public CustomsCancellationEvent(CustomsNotificationEvent original) {
    this.original = original;
  }
  public void Process() {
    original.Gateway.Cancel(original.ArrivalDate, original.Ship, original.Port);
  }
}

在這個案例中，我讓海關閘道緩衝區透過傳送取消通知給所有結束後倒帶的事件，以及傳送新通知給新的事件，來處理調整。在我的幻想世界中，政府機關真的非常配合。在實際上，每個調整都需要個別考量，它們很容易變得非常複雜，需要人工介入。不過，事件清單應該能大幅協助釐清外部系統需要執行的動作。