Zenn
🛠️

【ドメイン駆動設計】ドメインイベントの実装方法 - 1.概要編

2022/08/13に公開
Python
#
DDD
#
ドメイン駆動設計
#
ドメインイベント
tech

なにこれ

この記事では、「実践ドメイン駆動設計」(IDDD本)の「ドメインイベント」の具体的な実装方法をまとめています。

すでにIDDD本でドメインイベントの解説がされていますが、実務に活用できるだけの詳細な実装方法を示していなかったり、または理解できなかったり、そもそも例題で扱っているアジャイルプロジェクトがよくわからなかったりしました。そのため、本記事では、IDDD本を引用しながら分かりやすい例題でドメインイベントの実装方法を記載しました。

質問や感想、要望、ツッコミどころなどありましたら、コメントしてもらえると嬉しいです！

対象読者

  • ドメインイベントの実装方法を知りたい方
  • IDDD本を読んでみたけど、具体的にどう実装したらよいかわからない方
  • 未来の自分

ドメインイベントとは

ドメインイベントの概要説明やメリットは以下のサイトで分かりやすく解説されています。

実践DDD本 第8章「ドメインイベント」～出来事を記録して活用～ (1/3)：CodeZine（コードジン）

ドメインイベントを使うときの例



ドメインイベントの処理の流れ

1.サブスクライバの登録

アプリケーションサービスからイベントの発行のきっかけとなる集約のメソッドを実行する前に、アプリケーションサービスでサブスクライバをパブリッシャーに登録しておきます。

アプリケーションサービス
# ドメインイベントサブスクライバ
class ArticleLikedSubscriber(DomainEventSubscriber[ArticleLiked]):
	# ドメインイベントサブスクライバの実装クラス
	# ...

# アプリケーションサービス
class UserLikeArticleApplicationService:
    def __init__(self):
        subscriber = ArticleLikedSubscriber()
        DomainEventPublisher  # ドメインイベントパブリッシャー
	    .instance()  # スレッドローカルなインスタンスを返す
	    .subscribe(subscriber)  # ドメインイベントサブスクライバを登録

サブスクライバをパブリッシャーに登録する場所は？

  • アプリケーションサービス
  • (ごく稀)ドメインサービス

上記のサンプルコードだと、initメソッド内でサブスクライバを登録していますが、IDDD本ではアプリケーションサービスの実行メソッド内でサブスクライバを登録する例も示しています。(「ヴォーン・ヴァーノン. 実践ドメイン駆動設計 (Japanese Edition) (p.289). Kindle版」を参照)

明記箇所

ヘキサゴナルアーキテクチャを使う場合は、アプリケーションサービスがドメインモデルの直接のクライアントとなる。イベントの発行のきっかけとなる振る舞いを集約上で実行する前にサブスクライバをパブリッシャーに登録する場所としては、アプリケーションサービスが最適だろう。

ヴォーン・ヴァーノン. 実践ドメイン駆動設計 (Japanese Edition) (pp.288-289). Kindle 版.

ドメインサービスが、サブスクライバを登録できるようにする必要がある場合もある。その動機はアプリケーションサービスに登録させる場合と似ているが、イベントを待ち受けるための、そのドメインに特化した理由もある。

ヴォーン・ヴァーノン. 実践ドメイン駆動設計 (Japanese Edition) (p.290). Kindle 版.

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

ドメインイベントクラスが生成されるタイミングは、以下の通りです。

  • アプリケーションサービスから集約のメソッドを実行したとき
    • そのメソッドは、"ドメインエキスパートが気にかける、何かの出来事"であること
    • Zennの運営者) 「ユーザーが記事をLikeしたら...」 user.like(articleId)
  • アプリケーションサービスのメソッドを実行したとき
  • 他の境界付けられたコンテキストに対して通知する必要がある処理が実行されたとき

このうち、多くのドメインイベントは「アプリケーションサービスから集約のメソッドを実行したとき」に生成されます。


アプリケーションサービス
# ユーザーが記事をLikeするアプリケーションサービス
class UserLikeArticleApplicationService:
    # ...

    def like(self, a_user_id: str, an_article_id: str) -> NoReturn:
        # ...
	
	# 「ドメインエキスパートが気にかける、何か重要な出来事」=「ユーザーが記事をLikeしたら...」
	user.like(article_id)
	
	# ...
集約
class User:
    def like(self, article_id: ArticleId) -> NoReturn:
        # ...

	domain_event = ArticleLiked(article_id, user_id) # ドメインイベントを生成

4.ドメインイベントをドメインイベントパブリッシャーに発行する

生成されたドメインイベントは、集約のメソッド内でドメインイベントパブリッシャーに発行されます。

集約
class User:
    def like(self, article_id: ArticleId) -> NoReturn:
        # ...

	domain_event = ArticleLiked(article_id, user_id) # ドメインイベントを生成

	DomainEventPublisher\      # ドメインイベントパブリッシャー
	    .instance()\            # スレッドローカルなオブジェクトを返す
	    .publish(domain_event)  # 発行する

5.発行されたドメインイベントをドメインイベントサブスクライバで受信する

ドメインイベントパブリッシャーに発行されたら、発行されたそのドメインイベントを受信するサブスクライバに処理を委譲します。

ドメインイベントパブリッシャー
class DomainEventPublisher:
    # ...

    def publish(self, domain_event: DomainEvent) -> NoReturn:
        event_type = type(domain_event)  # 発行されたドメインイベントの型
        for subscriber in self.__subscribers:  # あらかじめパブリッシャーに登録されていたサブスクライバをループ
            # 対象のドメインイベントかどうかをチェック
	    if subscriber.subscribed_to_event_type() == event_type:
	        # ドメインイベントサブスクライバで受信する
                subscriber.handle_event(domain_event)

    # ...

ドメインイベントサブスクライバの処理方式

上節まででドメインイベントを発行し、サブスクライバまで受信する流れを説明しました。サブスクライバがドメインイベントを受信した後は、次の処理方式のいずれかで処理を行います。

1.同期処理方式

2.イベント格納処理方式

3.分散処理方式

サブスクライバがやってはいけないこと

別の集約のインスタンスを取得し、状態を変更させること。もし変更したい場合は、イベント格納処理方式を採用し、別途処理で変更するようにします。

明記箇所

サブスクライバがやってはいけないことがある。別の集約のインスタンスを取得し、そのコマンドを実行して状態を変更することだ。これは、単一のトランザクション内では単一の集約のインスタンスだけを変更するという経験則に反する。この経験則については第10章「集約」で説明する。[Evans]が言うように、トランザクションの中で使う集約ひとつを除いて、他のすべての集約の整合性は非同期的手段で保証する必要がある。

ヴォーン・ヴァーノン. 実践ドメイン駆動設計 (Japanese Edition) (p.290). Kindle 版.

設計・実装

モデリング

ドメインイベント


インターフェースのドメインイベントを実装し、実際のドメインイベント(上記の例でいうとArticleLiked)はそのインターフェースのドメインイベントを継承して実装する。

Pythonの実装
domain.model.domain_event.py
import abc
import datetime
from dataclasses import dataclass


@dataclass(init=False, unsafe_hash=True, frozen=True)
class DomainEvent(abc.ABC):
    """ドメインイベントは、不変クラスとして設計される
    
    クラス名の命名
     - 「イベント発生元のコンテキスト」のユビキタス言語に沿ったものとする
     - その出来事が過去に発生したことがわかるような名前にする
    
    プロパティについて
     - イベントバージョン
     - イベントの発生時刻
     - その他、サブスクライバの処理に必要なプロパティ

    メソッドについて
     - 各プロパティの読み取り用メソッド
     - その他、サブスクライバの処理に必要だと思われるメソッド(オブジェクトの不変性を保証するメソッドであること)
    """
    event_version: int
    occurred_on: datetime.datetime

    def __init__(self, event_version: int, occurred_on: datetime.datetime):
        """
	event_version: イベントバージョン
	occurred_on: イベントの発生時刻
	"""
        super().__setattr__("event_version", event_version)
        super().__setattr__("occurred_on", occurred_on)
domain.model.user.article_liked.py
from domain.model.domain_event import DomainEvent
from domain.model.article.id.article_id import ArticleId
from domain.model.user.id.user_id import UserId


class ArticleLiked(DomainEvent):
    """
    「記事がいいねされた」ことを示す(過去形になっている)
    """
    like_user_id: UserId  # 記事をLikeしたユーザーのID(だれがいいねしたか)
    liked_article_id: ArticleId  # Likeされた記事のID(何がいいねされたのか)
    
    def __init__(self, like_user_id: UserId, liked_article_id: ArticleId):
        super().__init__(1, datetime.datetime.now())
	super().__setattr__("like_user_id", like_user_id)
	super().__setattr__("liked_article_id", liked_article_id)

ドメインイベントパブリッシャー

Pythonでの実装
domain.model.domain_event_publisher.py
from __future__ import annotations

import threading
from typing import NoReturn, Set, Optional

from domain.model import DomainEvent, DomainEventSubscriber


class DomainEventPublisher(threading.local):
    __shared: Optional[DomainEventPublisher] = None

    def __init__(self):
        self.__subscribers: Set[DomainEventSubscriber] = set()

    @classmethod
    def instance(cls) -> DomainEventPublisher:
        if cls.__shared:
            return cls.__shared

        cls.__shared = DomainEventPublisher()
        return cls.__shared

    def reset(self) -> DomainEventPublisher:
        self.__subscribers = set()
        return self

    def publish(self, domain_event: DomainEvent) -> NoReturn:
        a_type = type(domain_event)
        for subscriber in self.__subscribers:
            if subscriber.subscribed_to_event_type() == a_type or isinstance(domain_event, DomainEvent):
                subscriber.handle_event(domain_event)

    def subscribe(self, subscriber: DomainEventSubscriber) -> NoReturn:
        self.__subscribers.add(subscriber)

ドメインイベントサブスクライバ

サブスクライバの実装は、「サブスクライバの処理方式」に応じて異なります。

  • 「1.同期処理方式」→実装されるドメインイベントごとに個別のサブスクライバを実装
  • 「2.イベント格納処理方式」「3.分散処理方式」→単一のサブスクライバを実装
明記箇所

後者の二つの例（イベントストアへの格納と、メッセージング基盤による転送）の場合は通常、この例のようなユースケースごとのアプリケーションサービスによるイベント処理は行わない。単一のサブスクライバコンポーネントを作って対応するだろう。

ヴォーン・ヴァーノン. 実践ドメイン駆動設計 (Japanese Edition) (p.289). Kindle 版.

Pythonでの実装
domain.model.domain_event_subscriber.py
# from ... import ...

T = TypeVar('T')


class DomainEventSubscriber(abc.ABC, Generic[T]):
    @abc.abstractmethod
    def subscribed_to_event_type(self) -> type:
        pass

    @abc.abstractmethod
    def handle_event(self, a_domain_event: T) -> NoReturn:
        pass
application.xxx.yyy.py(実装されるドメインイベントごとに個別のサブスクライバを実装する例)
# from ... import ...

class ArticleLikedSubscriber(DomainEventSubscriber[ArticleLiked]):
    def subscribed_to_event_type(self) -> type:
        return ArticleLiked

    def handle_event(self, a_domain_event: ArticleLiked) -> NoReturn:
        # ドメインイベント受信時の処理...
application.xxx.yyy.py(単一のサブスクライバを実装)
# from ... import ...

class DomainEventSubscriberImpl(DomainEventSubscriber[DomainEvent]):
    def subscribed_to_event_type(self) -> type:
        return DomainEvent

    def handle_event(self, a_domain_event: ArticleLiked) -> NoReturn:
        # ドメインイベント受信時の処理...

シーケンス図

クラス図

次回

ドメインイベントサブスクライバのそれぞれの処理方式の実装を紹介していきます。

※力尽きたので、1と3は要望があれば公開しますmm

参考文献

Discussion