遡及イベント

例：巻き戻しによる遡及貨物（C＃）

私は「イベントソーシング」に関するほとんどの例を、船舶による港間での貨物の移動に基づいた例で説明してきました。今回は、「遡及イベント」の使用方法について、その例を続けます。

まず、「遡及イベント」の実施方法に関するいくつかの戦略的決定を行う必要があります。この例では、ライブモデルを巻き戻すことで分岐点に到達するというアプローチを使用します。この例は単純であるため、選択的な再生は使用せず、イベントリスト全体を巻き戻します。

巻き戻しを使用しているため、すべてのイベントが可逆的であることが不可欠です。イベント処理または逆転の詳細には触れません。「イベントソーシング」の例を参照して、それがどのように機能するかを確認してください。

「遡及イベント」の動作は、特別な種類のイベントによってトリガーされます。

class 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");
  }

これらの場合、基本的なプロセスと逆転メソッドをブロックしていることに気付くでしょう。これらの例では、ほとんどの場合、イベントが独自の処理ロジックを処理することを好んでいました。置換は異なります。ドメインロジックを含まず、その処理にはイベントキューに関する知識と、そのイベントキューの操作が含まれます。イベントキューとの親密なやり取りは、イベントプロセッサだけに任せておく方が良いと思うので、実際にはすべての置換動作をイベントキューに組み込みました。

class 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（およびreverse）メソッドは、必要な場合にトレース動作を追加できるようにする単純なラッパーです。

class EventProcessor...

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

シーケンス外のイベントは、最初に説明するのが最も簡単です。これらは、シーケンス外で受信した通常のイベントであり、置換イベントを使用してモデル化しません。本質的にこれらはストリームに新しいイベントを挿入します。プロセッサは、それを最後の一つと比較するだけで、シーケンス外のイベントをテストします。

class 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;
  }

この例を非常に単純にするために、発生日によってイベントの順序付けを行います。これを実際に行うには、時間ポイントの解像度を向上させるだけで十分な場合があります。場合によっては、他の要因が順序付けに影響を与える可能性があります。

シーケンス外ケースの処理の概要は簡単です。

class EventProcessor...

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

巻き戻すには、シーケンス外のイベントより後のログからすべてのイベントを選択します。簡略化のために、ログ（または少なくともそのキャッシュ）をメモリに保持しています。

class 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));
  }

ご覧のとおり、選択的再生を使用していないにもかかわらず、すべてのイベントが巻き戻し中に再処理のために選択されるわけではありません。本質的に、エラーイベントまたは拒否されたイベントのいずれも再処理しません。

イベントを再び前方へ再生するのは簡単です。

class EventProcessor...

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

では、置換に進みましょう。

class 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);
    
  }

class ReplacementEvent...

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

ここで処理するケースはいくつかあります。元のイベントがすでに無視するようにマークされている場合、拒否済みのイベントを拒否すべきではないため、処理エラーが発生しています。それを終えたら、3つの主なケースがあります。置換は拒否であり、置換イベントは元よりも早く、元は置換よりも早いです。

拒否のケースから始めましょう。これは、置換イベントがnullであることで示されます。ここでは、拒否されたイベントに巻き戻し、拒否し、反転し、前方へ再生します。

class 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;
    }
  }

拒否すると、イベントがマークされ、再度処理または巻き戻されなくなります。

以前の置換がある場合、置換に巻き戻し、元を拒否し、置換を処理し、前方へ再生します。

class EventProcessor...

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

元のイベントが以前にある場合、元に戻し、反転して拒否し、置換まで再生し、処理し、前方への再生を続けます。

class 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);
  }

ここでは、イベントが正常に処理されるまでログにイベントを追加しないという規則に従っています。これは、ログがアプリケーション状態と同じトランザクションにない場合に役立ちます。すべてが1つのトランザクションにある場合、分岐点に到達して前方へ再生した後に（新しいイベントを取得する）イベントをログに追加するだけです。

例：外部システムの更新（C＃）

単純な「イベントソーシング」の場合、再生中に外部通知をオフにするだけで十分です。ただし、「遡及イベント」の場合、さらに進んで検出と修正を行う必要があります。巻き戻し/再生中にこれを行うには、生成されたすべてのイベントを保存して比較し、修正する必要があるものを決定します。次に、キャンセルメッセージと変更の新しいメッセージを送信するという単純な自動修正ケースを想定します。

ここでは、カナダを通過した貨物が米国の港に入港した場合、米国の税関当局に通知する必要があると述べることで、出荷の例を続けます。

外部通知は、到着イベントを処理しているときにCargoオブジェクトによって行われます。

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によって定義）を実際のメッセージングインフラストラクチャに変換する通常のゲートウェイです。それが何をしているかは安全に無視できます。メソッドを呼び出すと、税関にメッセージが届くことを確認すると仮定します。

興味深いことは、実際のCustomsGatewayFrontをラップするCustomsGatewayBufferによって駆動される前に発生します。同じインターフェースを実装しますが、遡及的な巻き戻しに対処する機能を追加します。

遡及的な巻き戻しの多くは汎用的であるため、汎用的な動作を異なるクラスReplayBufferに配置できます。これにより、Customs Gateway Bufferは税関のケースに固有のものだけを処理します。再生バッファは、巻き戻し/再生プロセスのさまざまな段階に対処するために、イベントプロセッサとの通信を必要とします。また、最終的に不一致イベントを調整する方法も必要であり、この場合は税関ゲートウェイバッファに割り当てます。これら両方のケースで、巻き戻しバッファは、その汎用性の純粋性を維持するために、コラボレーションインターフェースを介して通信します。

それが登場人物なので、アクションに入りましょう。ライブ実行時の通常のケースから始めます。税関ゲートウェイバッファは、更新イベントを作成して再生バッファに送信することで、通知操作を実装します。

class 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;

class 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();
  }

class CustomsNotificationEvent...

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

(Replayバッファがどのようにアクティブになり、現在のリストがどうなるかについては、後で説明します。)

ご覧のように、ドメインイベントの積極的な処理では、ゲートウェイバッファがゲートウェイイベントを作成し、それをReplayバッファに送信して処理することで、元の呼び出しが実際のゲートウェイに適用されます。これは非常に複雑な単純な委譲です。

この複雑さの利点は、遡及的なイベントの処理にあります。Replayバッファは、巻き戻し可能なインターフェースを介してイベントプロセッサに接続されています。

ReplayBufferクラス…

  IRewindable eventProcessor;

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

このインターフェースは、重要な3つのイベントを定義します。イベントプロセッサは、様々なケースの処理中にこれらのイベントをシグナルします。これが順序外れのケースです。

class EventProcessor...

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

置換の場合、全体的な開始と終了は共通ですが、各ケースでは巻き戻しが終了する時点を示すために、具体的な配置が必要です。

class 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);
}

これらのイベントにより、Replayバッファは遡及的な処理を正しく処理するための正しいタイミング情報を取得します。ここではイベントを使用していますが、どのイベントプロセッサにも任意の数のReplayバッファが存在し、これらのオブジェクトについて、これらの3つのイベントを理解できること以外何も知る必要はありません。

それでは、Replayバッファに移り、それがどのようにこれらのイベントを処理するかを見てみましょう。Replayバッファは、複数のリストで作成されます。

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);
  }

作成時にバッファはアクティブになり、送信されたイベントを処理します。`current list`変数は、巻き戻しの位置に応じて3つのコアリストのいずれかに設定されます。アクティブな場合、`sent list`を指します。これにより、実際に送信されたすべてのイベントの記録が得られます。また、イベントプロセッサの関連するすべてのイベントを購読します。(調整については今は気にしないでください。修正時に説明します。)

巻き戻しの開始時に、バッファは自身を非アクティブにして、イベントを実際のゲートウェイに渡さないようにします。また、`current list`変数を新しいリストを指すように切り替えて、巻き戻されたすべてのイベントをキャプチャできるようにします。

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);
  }

アルゴリズムは非常に単純で、一致するイベントを見つけ、両方のリストから削除するだけです。完了すると、不一致イベントの2つのリストが得られ、調整器に渡して処理します。

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);
  }
}

この場合、巻き戻されたすべてのイベントのキャンセル通知と新しいイベントの新しい通知を送信するだけで、CustomsGatewayBufferに調整を処理させます。私の理想の世界では、政府機関は非常に協力的です。実際には、各調整を個別に検討する必要があり、非常に複雑になり、人的介入が必要になる可能性があります。しかし、イベントのリストは、外部システムに対して何を行う必要があるかを整理するのに大いに役立つはずです。