🦫

DDDを実践するための手引き（ドメインイベント編）

2024/07/17に公開

DDD

はじめに

ドメインイベントはドメイン駆動設計で用いられる設計パターンの一つです。
ドメインイベント自体はシンプルな概念ですが、応用が利く概念なだけに様々な文脈で使われたり語られたりしているため、なかなか理解が難しいところがあります。私自身も最近ドメインイベントに関連する設計をすることがあり、いろいろと調査したのでその整理も兼ねてこの記事を書きました。

「集約」や「境界づけれれたコンテキスト」等の言葉にピンとこない方は先にこちらの記事を読んでみるか、その下のざっくり説明を開いて見てください。

時間がない人向けのざっくり説明

※今回の記事では↓のように読み替えてもらって支障ないと思います

集約:
- 雑な説明: エンティティのこと
- ちゃんとした説明: こちら
境界づけられたコンテキスト:
- 雑な説明: （ビジネスドメインによって分割された）1サービスもしくは1モジュール
- ちゃんとした説明: こちら

ドメインイベントとは

イベントとは「過去に発生した出来事」であり、ドメインイベントは「ビジネスドメイン上で発生した重要な出来事を表すメッセージ」です。
（例: チケットが割り当てられた、注文がキャンセルされた）

ドメインイベントはシステム内の状態の変化（=集約の状態の変化）を表現するものであり、通常は集約がドメインイベントの発生源となります。

用途

ドメインイベントは主に次のような目的で使用されます。

1. イベントの発生を起点に、別の処理をトリガーする

ドメインイベントは、システムの異なる部分間を連携させるために使用されます。

ドメイン上の要件として「...したら...する」のようなフレーズが出てきた場合、ドメインイベントの使い所である可能性があります。

複数集約間の整合性を取る（例: 注文がキャンセルされたら在庫を戻す）
外部システムとの連携（例: 新規予約が入ったらメール送信、外部APIの呼び出し）
ログの記録
etc.

利点としては次のようなものがあります:

イベントの発生源となる処理と、それに連動する処理を分離できる（関心の分離）
既存のコードを変更せずに、イベントに反応する新しい処理を追加できる（拡張性）

本記事ではこれを中心に実装方法を解説します。

2. ドメインイベントを記録し活用する

ドメインイベントを永続化し蓄積することで、現在の状態に至った過程を追跡することができます。

監査証跡
調査やデバッグ
ユーザーの活動の分析

3. イベントを情報源とした状態管理（イベントソーシング）

イベントソーシングは、システムの状態をイベントの連続として保存する設計パターンです。
ドメインオブジェクトの状態を直接保存（＝ステートソーシング）する代わりに、発生したすべてのイベントを逐次記録し、そのイベントを再生することで現在の状態を再構築します。
イベントソーシングは多くの場合 CQRS と併用されます。

4. コマンドとクエリの責務分離（CQRS）

CQRS (Command Query Responsibility Segregation) は、コマンド（書き込み操作）とクエリ（読み取り操作）を分離するアーキテクチャパターンです。
集約（コマンドモデル）に対する変更をイベントとして通知し、それを受け取ったイベントハンドラーが読み取り専用のビュー（クエリモデル）を構築することで、書き込みと読み取りそれぞれに最適化されたモデルやアーキテクチャ特性を持つことができます。

5. イベントを中心としたビジネスドメインの分析・モデリング（イベントストーミング）

システムの実装に直接関係するものではありませんが、ドメインイベントを中心にビジネスドメインを分析する手法としてイベントストーミングがあります。
これはビジネスドメインの理解を深め、共有し、モデリングや設計に活かすために使用されます。

ドメインイベントはビジネスドメインで起きる重要な結果であり、ここから遡って全体を明らかにしていきます。
（イベント → イベントを引き起こしたコマンド → コマンドの実行主体（アクター、ポリシー） → アクターが意思決定のために参照したリードモデル）

ドメインイベントを扱うための実装

ドメインイベントを生成し、境界づけれれたコンテキスト内で処理するために必要な実装を紹介します。

ドメインイベントの発生〜消費は同期でも非同期でも構いません
通常ドメインイベントはそれが発生した境界づけられたコンテキスト内で消費されます
- コンテキスト外に通知したい場合は、公開用に変換・整形した上で通知するのが一般的です

1. ドメインイベントの設計

ドメインイベントは集約に対する操作（コマンド）の結果として発生します。
例えば、「注文」という集約に対する「キャンセル」操作は、集約の状態をキャンセル済みに変更し、「注文がキャンセルされた」というドメインイベントを発生させます。

まずはドメインイベントを設計・実装します。

/**「注文がキャンセルされた」イベント */
// OrderCancelledEvent, OrderCancelledDomainEvent のように命名する方針もありえる
data class OrderCancelled(
  // 出来事を説明するための情報をプロパティとして持つ
  val orderId: OrderId,
  val reason: String,
  val occurredAt: Instant,
  ...
) : DomainEvent

ポイント:

過去に発生した出来事であるため過去分詞形で命名する（例: OrderCancelled、ItemShipped）
ビジネスドメインの言葉で何が起こったかを名前に正確に反映する
過去に発生した出来事であるためイミュータブル
ドメインイベントはドメインモデルの一部

ドメインイベントに持たせる情報

ドメインイベントは、ドメインで何が起こったかを記述する完全な情報を持ちます。

イベントを説明するために必要なすべての情報を持つ（何が、なぜ、いつ起きたか）
イベントが暗黙的に依存している外部情報も含める
- 例えば、税込価格を計算し記録するイベントであれば、その計算に用いた消費税率のスナップショットも含める（将来税率が変わった場合でも計算結果が再現できるように）
その他、サブスクライバが必要とする情報も含めることもできる（※流派による）
上記を満たした上で必要最小限な情報に留める

すべてのドメインイベントに共通する情報

すべてのドメインイベントで共通して持つべき情報があります。

集約のID (必須) ... イベントは集約から発生するものなので、集約の識別子を持つ
イベントの発生時刻 (必須) ... 過去の出来事であるため、発生時刻を持つ
イベントの識別子 ... 非同期にハンドリングするときにイベントの重複を除外するために便利（後述）
集約のバージョン ... イベントを永続化する場合に楽観的排他制御を行ったり、イベントの順序を保証するために使う

これらの項目は共通の interface に定義しておくといいかもしれません。

2. ドメインイベントを生成する

集約がドメインイベントを生成し発行する実装方法としては以下の3つのパターンがあります。

パターン1. 集約自身が Event Publisher を使ってドメインイベントを発行する

集約自身がドメインイベントを発行します。
実践ドメイン駆動設計などで紹介されているパターンです。

class Order {
  fun cancel(reason: String) {
    check(canCancel())
    // 集約の状態を変化させる
    this.status = OrderStatus.CANCELLED
    this.cancelReason = reason

    // イベントを生成し、発行する
    val event = OrderCancelled(id, reason, ...)
    DomainEventPublisher.publish(event)
  }
}

（ここでは DomainEventPublisher がイベントを受取りディスパッチする部品としています。実装については後述）

このパターンは実装はシンプルですが、以下の問題があるためあまり用いるべきではありません。

集約がイベント発行処理に依存するため、安定性やテスタビリティが下がる
即時にディスパッチされるため、集約を永続化する前にハンドラーがイベントを処理する可能性がある

パターン2. 集約がドメインイベントを保持し、後で取り出して発行する

これは最もよく紹介されるパターンです。（Jimmy Bogard の A better domain events pattern等）

class Order {
  // 集約内で発生した未発行のイベントを保持するフィールド
  // 共通の基底クラスに持たせても良い
  private val events = mutableListOf<DomainEvent>()

  fun cancel(reason: String) {
    check(canCancel())
    this.status = OrderStatus.CANCELLED
    this.cancelReason = reason

    // 生成したイベントをeventsに追加
    // この時点ではイベントはディスパッチされない
    val event = OrderCancelled(id, reason, ...)
    events.add(event)
  }

  fun events(): List<DomainEvent> { // 外から取り出せるようにしておく
    return events.toList()
  }
}

// Use case
fun cancelOrder(orderId: OrderId, reason: String) {
  val order = orderRepository.findById(orderId)
  order.cancel(reason)
  orderRepository.save(order)

  // 生成されたイベントを発行
  // リポジトリの実装内でこれをやる流派もある（リポジトリの責務範囲外ではあるが...）
  domainEventPublisher.publish(order.events())
}

パターン3. 集約のコマンドがイベントを返す

集約に対する操作がイベントを返すようにすることで、イベントの発行を明示的に行うことができます。

class Order {
  fun cancel(reason: String): List<DomainEvent> {
    check(canCancel())
    this.status = OrderStatus.CANCELLED
    this.cancelReason = reason

    // イベントを生成し、返す
    val event = OrderCancelled(orderId, reason, ...)
    return listOf(event)
  }
}

// Use case
fun cancelOrder(orderId: OrderId, reason: String) {
  val order = orderRepository.findById(orderId)
  val events = order.cancel(reason)
  orderRepository.save(order)

  domainEventPublisher.publish(events) // イベントを発行
}

このパターンは集約の設計をイミュータブルにする場合にも使えます。

// Use case
fun cancelOrder(orderId: OrderId, reason: String) {
  val order = orderRepository.findById(orderId)
  val (cancelledOrder, events) = order.cancel(reason) // 変更後の集約とイベントを返す
  orderRepository.save(cancelledOrder)

  domainEventPublisher.publish(events)
}

パターン2,3においては通常 DomainEventPublisher の呼び出しはトランザクションのコミット直前に行います。

3. イベントを処理する

DomainEventPublisher から発行されたイベントを購読し処理する実装例を示します。

まずドメインイベントを処理するための interface として、DomainEventHandler を定義します。

interface DomainEventHandler<T : DomainEvent> {
  fun eventType(): KClass<T> // このハンドラが興味のあるイベントの型を返す
  fun handle(event: T) // イベントを処理する
}

次にこれを実装したクラスを作成します。

class SendEmailWhenOrderCancelledHandler : DomainEventHandler<OrderCancelled> {
  override fun eventType() = OrderCancelled::class
  override fun handle(event: OrderCancelled) {
    // イベントに反応して行うアクションを記述
    // イベントハンドラーはアプリケーション層の住人であり、リポジトリやサービスを利用することもできる
  }
}

命名はドメインイベント: 設計と実装 - .NET | Microsoft Learn の例に倣っています。（${何をするか}When${何が起こったか}Handler）

1つのイベントに対して実行したい処理が複数ある場合、通常はハンドラーを複数作成します。（単一責任の原則、開放閉鎖の原則）

4-1. イベントの同期的なディスパッチ

最後に発生したイベントをハンドラーにディスパッチする DomainEventPublisher を実装します。
イベントのディスパッチを同期的に行う場合と非同期的に行う場合があります。

まずは同期的なディスパッチの実装例を示します。

class DomainEventPublisher {
  // イベントタイプごとのハンドラーを保持する
  private val handlers = mutableMapOf<KClass<out DomainEvent>, List<DomainEventHandler<out DomainEvent>>>()

  // イベントハンドラーを登録する
  fun subscribe(handler: DomainEventHandler<out DomainEvent>) {
    val eventType = handler.eventType()
    handlers[eventType] = handlers.getOrDefault(eventType, emptyList()) + handler
  }

  // 発生したイベントのタイプに応じたハンドラーにディスパッチする
  // イベントに対応するハンドラーを順次、同期的に呼び出す
  fun publish(events: List<DomainEvent>) {
    events.forEach { event ->
      handlers[event::class]?.forEach { handler ->
        (handler as DomainEventHandler<DomainEvent>).handle(event)
      }
    }
  }
}

イベントハンドラーはアプリケーション起動時に登録しておきます。
DIコンテナを使っているのであれば、DIコンテナに登録されている全ハンドラーを自動的に登録すると便利です。

fun registerDomainEventHandlers() { // アプリケーション起動時に呼び出す
  val handlers = applicationContext.getBeansOfType(DomainEventHandler::class.java) // Spring Framework の例
  handlers.values.forEach { handler -> domainEventPublisher.subscribe(handler) }
}

呼び出し元でトランザクションをコミットする前に DomainEventPublisher.publish() を呼び出すことで、集約の更新とイベントの処理を同一トランザクションで行うことができます。

4-2. イベントの非同期的なディスパッチ

同一コンテキスト（同一システム）内であっても、イベントを非同期的に処理したいことがあります。

イベントを消費する側の処理を待機しないため、応答性が向上する
イベントを消費する側がダウンしていても処理を続行できるため、可用性が向上する
トランザクションが小さくなり、データベースの排他制御の競合が減る

一方で、イベントが確実に処理されるようにするために複雑な実装が必要になります。
（イベントの重複、順序保証、エラー処理、補償アクションなど）

結果整合性について

イベントを非同期に処理すると、イベントが発生してからイベントハンドラー側の処理が完了するまでの間、システムの状態に一時的に整合性が取れない状態が発生します。
（例: 商品購入処理とポイント付与処理が非同期で行われる場合、購入したのにポイントが未付与の状態が一時的に発生する）

このように一時的な不整合を許容し、時間の経過とともに最終的に整合性を担保する設計を結果整合性と呼びます。
反対語はトランザクション整合性で、常に整合性を保ちます。

Outbox パターン + メッセージバスによるイベント配信

Producer - Consumer 間を結果整合性にするとしても、「イベントの発生源（＝集約）の状態の永続化」と「イベントが公開されること」は確実に整合する必要があります。（片方だけが成功するのはダメ）

このような場合 Outbox パターンを利用するのが一般的です。
（別の方法として、ドメインイベントのみを永続化し、集約はイベントから再構築する方法もあります）

Outbox パターンでは、集約の保存と同じトランザクションで通知したいイベントを一旦永続化し、別のプロセスがそのイベントを取得して通知します。

通知したいイベントを永続化するためのテーブル（Outbox テーブル）を用意する
同一トランザクション内で、集約の永続化とイベントの永続化を行い、トランザクションをコミットする
別プロセス（メッセージリレー）が Outbox テーブルを監視し、新しいイベントを取得する
メッセージリレーは取得したイベントをメッセージバス（Kafka, RabbitMQ, etc.）に投入する
成功すればメッセージリレーは Outbox テーブルからイベントを削除するか、公開済みとしてマークする

5で失敗する場合、3〜4が再度実行されるので、イベントは2回以上送信される可能性があります（At-least-once）

メッセージリレーの実装

Outbox テーブルに書き込まれた新規イベントを取得する方法は次の2つがあります。

ポーリング ... 定期的に Outbox テーブルにアクセスし、新規イベントを取得する
CDC（Change Data Capture） ... 何かしらの仕組みでデータの変更通知を受け取る（DB によってはできない場合もある）
- セルフホステッドなRDBであれば Debezium などのツールを使う
- AWS の場合、
  - DynamoDB であれば DynamoDB Streams で実現できる
  - PostgresSQL であれば論理レプリケーションを使うことで実現できる

イベント受信側の実装

メッセージバスからイベントを取得しハンドラーに渡します。
ハンドラーの実装は基本的には同期的の場合と同じです。

ただし、イベントの通知が非同期になっていることにより、次のような問題が発生する可能性があるため、それに対処する必要があります。

イベントが重複して到着する
イベントが順不同で到着する
ハンドラーがイベントの処理に失敗する

イベントの重複排除

通常、イベントは少なくとも一度（At-least-once）通知されます（メッセージバスや Outbox が配送の過程で潜在的な再試行を行うため）。
そのため、受け取ったイベントの重複を排除する仕組みが必要です。

イベントハンドラーの処理を冪等に設計し、同じイベントが複数回処理されても結果が変わらないようにする
各イベントに一意の識別子を付与し、処理済みのイベントIDを記録。重複しているイベントが到着した場合は無視する

イベントの順序保証

イベントを非同期に通知する場合、イベントがコンシューマーに到達する順序が発生順と一致しない場合があります。
基本的には同一の集約に関するイベントは発生順に処理されるようにします。

メッセージバスの順序保証機能を使う
- 例: Kafka のパーティション（パーティションキーを集約のIDにする）
受信側でイベントの順序を並び替える

エラー処理

イベント受信側で予期せずエラーが発生することがあります。（ネットワークエラーなど）
一般的には次のようなアプローチが取られます。実装方法はメッセージバスによって異なります。

自動的にリトライする（リトライ間隔を指数バックオフで調整）
エラーが発生したイベントをデッドレターキューに移動し、後で再処理する

補償アクション

イベントハンドラーが何らかの理由で処理を完遂できない場合があります。
例: 注文イベントに反応して決済を行うハンドラーで、指定された決済方法での支払いが不可能だった

イベントは既に発生したことを記述するものであり、なかったことにはできません。
イベントを覆す唯一の方法は、イベントの効果を打ち消すための追加のアクションを実行することです。
先ほどの例の場合、「注文をキャンセルし在庫を戻す」などの操作を行う補償アクションが考えられます。

複数集約の更新と整合性

同一システム（境界づけれれたコンテキスト）内の複数の集約の整合性を担保したいことがあります（＝片方を更新したらもう片方もあわせて更新したい）。実装としては、片方の集約で発生したドメインイベントをトリガーにして他の集約を更新することになります。

ここで、DDD では一般的には「1つのトランザクションにつき1つの集約を更新する」という原則があり、結果整合性を採用すべきとされています。そもそも「集約」は一貫性を保つ単位として設計されているはずであり、複数の集約間で即時の整合性を保つ必要はないはずです。
これに従うなら、このような状況でのドメインイベントの処理は非同期である必要があります。

一方、単一のトランザクションで複数の集約を更新しても構わないという主張もあります。DDD の原則を無視してトランザクション整合性を選ぶ理由は以下の2つが考えられます。

ドメイン上の理由により厳密な整合性が必要な場合
結果整合性を実現するためのコストを払うことが難しい場合（イベントの確実な配送、重複の排除、再試行、補償アクションなど）

ドメインイベント: 設計と実装 - .NET | Microsoft Learn では、次のように述べられています。

たとえば、Bogard は「A better domain events pattern」(よりよいドメインイベントパターン) で次のように書いています。

通常はドメインイベントの副作用は同じ論理トランザクション内で発生する必要がありますが、必ずしも同じドメインイベント発生スコープ内である必要はありません [...]。トランザクションをコミットする直前に、対応するハンドラーにイベントをディスパッチします。

実際には、どちらの方法 (単一のアトミックトランザクションと最終的な整合性) も適切な場合があります。どちらがよいかは、ドメインまたはビジネスの要件と、ドメイン専門家の意見に依存します。また、サービスに必要なスケーラビリティのレベルにも依存します (トランザクションの粒度が細かいほど、データベースロックに関する影響は小さくなります)。また、最終的な整合性では、集約間で可能性のある不整合を検出するためにより複雑なコードが必要であり、補正アクションを実装する必要があるため、コードにかけられる費用によっても左右されます。元の集約に変更をコミットした後、イベントがディスパッチされるとき、問題があってイベントハンドラーが副作用をコミットできない場合は、集約の間に不整合が発生することを考慮してください。

ドメインイベントをコンテキストの外部に通知する（統合イベント）

複数コンテキストの連携のために、ドメインイベントを境界づけれれたコンテキストの外部に通知したいことがあります。
イベントをコンテキスト内に通知する場合とはいくつか異なる点があります。

ドメインイベントをそのまま外部に公開することの是非

まずそもそもドメインイベントを外部にそのまま公開していいのかという問題があります。

ドメインイベントはドメインモデルの一部であり、境界づけれれたコンテキストの内部表現である
- 外部に公開することでカプセル化が崩れる
- 内部表現と外部システムが結合することにより、イベントの構造の変更が困難になる
外部との連携のために必要な情報とドメインイベントの情報が一致しないことがある

これらの理由から一般的にはドメインイベントをそのまま外部に公開することは避けるべきと考えられます。が、書籍や記事によってはそのまま公開する実装例を紹介しているものもあります。

変換して公開する派

ドメインイベントを外部のコンテキストに送出する前に、公開用のイベントに変換します。（以後統合イベント（Integration Event）と呼びます。）
変換にはステートレスな変換とステートフルな変換があります。

ステートレスな変換（＝ドメインイベントのみを情報源とする）
- 不要なメッセージやプロパティを省く
  - コンテキスト間の偶発的な結合を避けるため
  - セキュリティのため
- 連携に適した形式や名称に変換
ステートフルな変換（＝過去のドメインイベントや他のデータストアから取得した情報を用いる）
- 複数のイベントを集計・統合
- リポジトリから取得した情報を追加

変換して公開する派の書籍や記事

ドメインイベント: 設計と実装 - .NET | Microsoft Learn では、ドメインイベントを統合イベント（Integration Event）に変換して外部に公開している
- この変換は DomainEventHandler で行い、統合イベント送信用のサービスに渡している
- プロパティはドメインイベントと同じではなく、必要に応じて追加、削除されている
Learning Domain-Driven Designでは、ドメインイベントを公開言語に変換して外部に公開している
- ドメインイベントを送出する前にプロキシを通して変換する
  - ステートレスモデル変換
  - ステートフルモデル変換
- ドメインイベントとその他のイベントの使い分けを推奨している
  - Event notification ... 出来事の発生を通知するが、詳細情報は持たない。詳細情報が必要な場合はコンシューマーから明示的に問い合わせる
  - Event-carried state transfer ... コンテキスト間でデータの複製を同期するために、上流コンテキストのデータの変化を別のコンテキストに転送する
  - Domain event ... ドメイン内で起きた出来事を説明するメッセージ

そのまま公開する派

イベントを変換するための手数を嫌う場合はこちらを選択することも考えられます。
ただし上述のような問題があるため注意が必要です。

そのまま公開する派の書籍や記事

実践ドメイン駆動設計では、JSONシリアライズしたドメインイベントを外部に通知する例が紹介されている
1. システム内で発生したすべてのドメインイベントを受け取るハンドラーを作成
2. イベントハンドラーは受け取ったイベントをJSONにシリアライズしイベントストア（RDBの1テーブル）に永続化
3. イベントストアに保存されたイベントは、次のいずれかの方法で外部に通知される
  - イベントを取得するためのRESTfulなAPIを公開し、外部サービスからポーリングさせる
  - メッセージングミドルウェアを使って送信する
マイクロサービスパターンでも、ドメインイベントを外部に通知している
- そもそもドメインイベントを外部との連携のためのものとして設計している
- 「イベントエンリッチメント」と称して外部サービスが要求する追加の情報をドメインイベントに含める設計を紹介している

統合イベントの設計

ドメインイベントは「ドメイン上の出来事を正確に説明する過不足ない情報」という観点から設計されます。一方で、統合イベントは境界づけられたコンテキストの公開I/Fの一部であり、「外部コンテキストとの連携のための契約」という性質があります。

ドメインイベント	統合イベント
ドメイン上の出来事を説明する情報	外部コンテキストとの連携のための契約
境界づけられたコンテキスト内で消費	境界づけられたコンテキストの外部に公開
イベントの処理は同期でも非同期でも構わない	常に非同期に処理される

契約としての統合イベント

関数型ドメインモデリングでは2つコンテキストが契約に同意するパターンとして次の3つを挙げています。

共有カーネル関係 ... コンテキスト間でドメイン設計を共同所有する。イベントの定義は他の共同所有者との協議に基づく
顧客/供給者の関係 ... 下流のコンテキストが上流のコンテキストに対して欲しい契約を定義する（コンシューマー駆動契約）。この契約を満たす限り、それぞれのコンテキストは独立して発展できる
順応者の関係 ... 下流のコンテキストは上流のコンテキストが提供する契約を受け入れる

コンシューマー駆動契約は、イベントに限らずマイクロサービス間のI/F設計に用いられる一般的なパターンです。
契約を壊していないことをコンシューマー、プロバイダーの双方がテストすることで、通信が壊れないことを保証できます。契約をテストするためのツールとして、Pact があります。

スキーマ進化

イベントのスキーマはビジネス要件の変化等により変更されます。
イベントの生産者と消費者が互いに独立して進化でき、かつすべての時点において途切れることなく連携できるようにするためには、スキーマの互換性を考慮する必要があります。

互換性タイプ:

前方互換性: 新しいスキーマのデータを古いスキーマとして読み取ることができる
- 生産者側で新しいスキーマのイベントを送信しても、消費者側はコードを変更する必要がない
- 消費者は新スキーマの情報を必要とする場合にのみコードを修正すればOK
後方互換性: 古いスキーマのデータを新しいスキーマとして読み取ることができる
- スキーマが先に定義されていれば、消費者は生産者よりも先にコードの更新をリリースできる
- 古いイベントを再処理する必要がある場合に対応できる
完全互換性: 前方互換性と後方互換性の両方を持つ
- 可能な限り目指すべき
- 互換性の要件をあとで緩めるのは簡単だが、厳しくすることは困難

破壊的変更が必要な場合は、古いイベントと新しいイベントの両方をサポートし、期限を設けて古いイベントを廃止することが一般的です。

統合イベントへの変換と送信

ドメインイベントの場合とは違い、統合イベントは常に非同期的に通知・処理されます。
統合イベントを扱う仕組みは、ドメインイベントの仕組みを土台として構築します。

パターン1. ドメインイベントハンドラーで統合イベントに明示的に変換して送信する

ドメインイベントを受け取ったハンドラーが統合イベントに変換し、送信します。

対象のドメインイベントを購読するハンドラーを作成
1.のハンドラー内でドメインイベントから統合イベントに変換
2.の統合イベントを Outbox パターン & メッセージバスを用いて非同期的に送信する

// OrderCancelled イベントを統合イベントに変換・送信するハンドラー
class PublishIntegrationEventWhenOrderCancelledHandler(
  private val integrationEventPublisher: IntegrationEventPublisher
) : DomainEventHandler<OrderCancelled> {
  override fun eventType() = OrderCancelled::class

  override fun handle(domainEvent: OrderCancelled) {
    // 統合イベントに変換
    // - 連携に適した情報に変換する
    // - 不要な情報を省く
    // - リポジトリを使用して追加の情報を取得
    // - 複数のイベントを集計・統合
    val integrationEvent = OrderCancelledIntegrationEvent(
      domainEvent.orderId,
      ...
    )
    // IntegrationEventPublisher は Outbox パターン & メッセージバスを用いて非同期的にイベントを送信する
    // （ドメインイベントを非同期にディスパッチする場合とだいたい同じ技術を使って実装する）
    integrationEventPublisher.publish(integrationEvent)
  }
}

上の例ではドメインイベントを個別にハンドリングしていますが、すべてのドメインイベントを中央集権的に処理するハンドラーを作成することもできます。単一責任の原則に反するようですが、見通しは良くなります。

class PublishIntegrationEventHandledHandler(
  private val integrationEventPublisher: IntegrationEventPublisher
) : DomainEventHandler<DomainEvent> {
  override fun eventType() = DomainEvent::class

  override fun handle(domainEvent: DomainEvent) {
    val integrationEvent = when (domainEvent) {
      is OrderCancelled -> OrderCancelledIntegrationEvent(
        domainEvent.orderId,
        ...
      )
      is OrderShipped -> OrderShippedIntegrationEvent(
        domainEvent.orderId,
        ...
      )
      ...
    }
    integrationEventPublisher.publish(integrationEvent)
  }
}

パターン2. イベント変換レイヤーを配置する

ドメインイベントを非同期に送信する仕組みがある場合、どこかしらでイベントの流れをインターセプトして統合イベントに変換することもできます。
イベント変換レイヤーを配置する場所としては、次のような候補があります。

Outbox テーブルからドメインイベントを取り出した直後に統合イベントに変換してメッセージバスに投入する
メッセージバスに投入されたドメインイベントを取り出して統合イベントに変換し、再度メッセージバスに投入する
メッセージバスによっては変換処理を行うためのフックを提供している場合がある
- AWS Kinesis Data Firehose、Kafka Connect など

パターン3. イベントを取得するAPIを提供し、購読側からポーリングする

全く違うアプローチとして、イベントを発行する側がイベント取得のエンドポイントを公開し、イベントを受け取る側からポーリングする方法もあります。
この方法は実践ドメイン駆動設計で「RESTfulなリソースによる通知の発行」として紹介されています。

イベント発行側:
- システム内で発生したすべてのイベントをイベントストアに永続化する
- 新しいイベントを取得するためのエンドポイントを用意する
  - RSSやAtomフィードのようなイメージ
  - 内部ではイベントストアからイベントを取得して返す
イベント受信側:
- イベント発行側のエンドポイントを定期的にポーリングする
- どこまで処理したかを自分で管理

統合イベントへの変換は、イベントストアへの書き込み時か、レスポンスを返す前に行うといいでしょう。（書籍では変換処理は省略されている）

おわりに

イベントソーシングやもう少し具体的な実装例なども書きたかったのですが、力尽きました。そのうち追記するかもしれません。

書籍やネット上の文献を読み漁りながらこの記事を書きましたが、内容にまだまだ至らない部分もあるかもしれません。間違って理解している箇所を見つけられた場合やさしく諭していただけるととても喜びます。筆者の理解が更新されたら、この記事も更新していきます。

よかったらこちらも: