[SpringBoot] AbstractAggregateRoot로 DomainEvent 발행하기

DDD에서 AggregateRoot와 Domain Event란?

도메인 주도 개발(Domain-Driven Design)에는 애그리거트(Aggregate)라는 단위로 도메인을 관리한다.

애그리거트는 복잡한 도메인을 관리하기 쉬운 단위로 분리한 것인데 이커머스를 예로 들면 상품이 하나의 애그리거트, 리뷰가 하나의 애그리거트가 될 수 있다.

애그리거트는 다른 애그리거트에 관여하지 않고, 생성되고 변경되는 이유가 각자 다르다.

AggregateRoot

애그리거트는 여러 개의 모델로 이루어져 있을 수 있다. 이 때, 여러개의 모델을 한 번에 관리하는 주체를 AggregateRoot라고 한다.

애그리거트의 내부를 변경하기 위해서는 오직 애그리거트 루트만 가능하며, 한 트랜잭션에서는 한 개의 애그리거트만 수정해야 한다.

ProductAggregate의 AggregateRoot는 Product이다.

DomainEvent

한 트랜잭션에서는 한 개의 애그리거트만 수정해야하지만 업무 규칙으로 인해 하나의 애그리트가 변경되면 그와 연쇄된 다른 로직이 수행되어야 하면 어떻게 해야할까?

서비스 계층에서 이를 구현할 수 있지만 사실 DDD관점에서 비즈니스 로직이 서비스 계층이 있는건 원하는 것이 아니다.

그래서 연관이 있는 애그리거트 간의 통신은 도메인 이벤트(Domain Event)를 이용한다.

DomainEvent 발행하기

예를 들어, 상품의 가격이 변경되면 리뷰를 썼던 유저들에게 알림이 가는 로직을 구현해보자.

상품과 리뷰의 결합을 최대한 약하게 하기 위해 도메인 이벤트를 이용할것이다.

1. 상품의 가격을 변경한다.
2. 성공적으로 변경하면 상품 가격 변경 이벤트를 발행한다.
3. 핸들러가 이벤트를 수신하고 리뷰한 유저에게 알림을 보낸다.

또한, 트랜잭션이 성공한 경우에만 이벤트를 처리하도록 @TransactionalEventListener를 사용할것이다.

@Component
class ProductPriceChangedEventListener {
    @Async
    @TransactionalEventListener
    fun handlerProductPriceChanged(productPriceChanged: ProductPriceChanged) {
        println("product price changed: $productPriceChanged")
    }
}

도메인 내부에서 이벤트 발행하기

@Entity
class Product(
    @Id @GeneratedValue(strategy = GenerationType.IDENTITY)
    val id: Long = 0,
    param: ProductParam

) {

    val name: String = param.name

    var price: Long = param.price

    @OneToMany(mappedBy = "product", fetch = FetchType.LAZY)
    @BatchSize(size = 10)
    val images: List<ProductImage> =
        param.imageUrls.mapIndexed { seq, url -> ProductImage(product = this, url = url, seq = seq) }

    fun updatePrice(newPrice: Long, eventPublisher: ApplicationEventPublisher) {
        val oldPrice = this.price
        this.price = newPrice
        eventPublisher.publishEvent(ProductPriceChanged(id, oldPrice, this.price))
    }
}

도메인 내부에서 이벤트를 발행하니 뭔가 조금 더 DDD 스러워졌다.(사실 도메인 계층에서 Spring Data JPA를 의존하고 있지만...)

하지만 이 방법은 문제가 있다.

1. 트랜잭션 내부가 아닌 상황에서 updatePrice를 호출하면 이벤트가 제대로 수신되지 않는다.(이벤트 리스너의 옵션에 따라 실제로 데이터가 변경되지 않았는데 불구하고 이벤트를 수신하는 경우가 있을 수도 있다.)
2. 도메인 계층에서 ApplicationEventPublisher를 알고 있어야 한다.

서비스 계층에서 이벤트 발행하기

@Transactional
@Service
class ProductService(
    private val repository: ProductRepository,
    private val eventPublisher: ApplicationEventPublisher
) {
    fun create(param: ProductParam): Product {
        val product = Product(param = param)
        return repository.save(product)
    }

    fun updatePrice(productId: Long, newPrice: Long): Product {
        val product = repository.findById(productId).orElseThrow { RuntimeException("Product not found") }
        val oldPrice = product.price
        product.updatePrice(newPrice)
        eventPublisher.publishEvent(ProductPriceChanged(id=product.id, oldPrice=oldPrice, newPrice=newPrice))
        return repository.save(product)
    }
}

서비스 계층에서 이벤트를 발행하도록 바꿔주었다. 이렇게 하면 확실하게 트랜잭션이 완료되면 이벤트를 처리할 것이다.

하지만 도메인 로직을 수행하고나서 이벤트를 발행해야하는 책임이 서비스 계층에 있어 서비스 계층은 상품 관리외에 이벤트 관리에 대한 하나의 책임을 더 져야한다.

AbstractAggregateRoot 사용하기

Spring Data JPA에선 위의 문제점을 해소한 이벤트를 처리하는 과정을 추상화한 AbstractAggregateRoot를 제공해준다.

@Entity
class Product(
    @Id @GeneratedValue(strategy = GenerationType.IDENTITY)
    val id: Long = 0,
    param: ProductParam
): AbstractAggregateRoot<Product>() {

    val name: String = param.name

    var price: Long = param.price

    @OneToMany(mappedBy = "product", fetch = FetchType.LAZY)
    @BatchSize(size = 10)
    val images: List<ProductImage> =
        param.imageUrls.mapIndexed { seq, url -> ProductImage(product = this, url = url, seq = seq) }

    fun updatePrice(newPrice: Long) {
        val oldPrice = this.price
        this.price = newPrice
        registerEvent(ProductPriceChanged(id, oldPrice, newPrice))
    }
}

AbstractAggregateRoot를 상속받으면 repository.save() 시점에 updatePrice에서 등록해놓은 ProductPriceChanged 이벤트를 발행하게 된다.

public class AbstractAggregateRoot<A extends AbstractAggregateRoot<A>> {
    @Transient
    private final transient List<Object> domainEvents = new ArrayList();

    public AbstractAggregateRoot() {
    }

    protected <T> T registerEvent(T event) {
        Assert.notNull(event, "Domain event must not be null");
        this.domainEvents.add(event);
        return event;
    }

    @AfterDomainEventPublication
    protected void clearDomainEvents() {
        this.domainEvents.clear();
    }

    @DomainEvents
    protected Collection<Object> domainEvents() {
        return Collections.unmodifiableList(this.domainEvents);
    }

    protected final A andEventsFrom(A aggregate) {
        Assert.notNull(aggregate, "Aggregate must not be null");
        this.domainEvents.addAll(aggregate.domainEvents());
        return this;
    }

    protected final A andEvent(Object event) {
        this.registerEvent(event);
        return this;
    }
}

실제 이벤트 발행이 어떻게 되는지 보려면 EventPublishingRepositoryProxyPostProcessor 를 보면 되는데 repository를 프록시로 감싸고, 조건에 맞는 repository 함수를 호출하면 프록시 객체에서 repository 기능을 수행한 다음 이벤트를 발행한다. 설명에 필요하다고 생각되는 중요한 로직만 가져와보았다.

public class EventPublishingRepositoryProxyPostProcessor implements RepositoryProxyPostProcessor {
    private final ApplicationEventPublisher publisher;

    public EventPublishingRepositoryProxyPostProcessor(ApplicationEventPublisher publisher) {
        Assert.notNull(publisher, "Object must not be null");
        this.publisher = publisher;
    }

    public void postProcess(ProxyFactory factory, RepositoryInformation repositoryInformation) {
        EventPublishingMethod method = EventPublishingRepositoryProxyPostProcessor.EventPublishingMethod.of(repositoryInformation.getDomainType());
        if (method != null) {
            factory.addAdvice(new EventPublishingMethodInterceptor(method, this.publisher));
        }
    }

    ...

    static class EventPublishingMethod {
        ...

        public void publishEventsFrom(@Nullable Iterable<?> aggregates, ApplicationEventPublisher publisher) {
            ...
                        publisher.publishEvent(event);

        }
       ...
    }

    static class EventPublishingMethodInterceptor implements MethodInterceptor {
        private final EventPublishingMethod eventMethod;
        private final ApplicationEventPublisher publisher;

        private EventPublishingMethodInterceptor(EventPublishingMethod eventMethod, ApplicationEventPublisher publisher) {
            this.eventMethod = eventMethod;
            this.publisher = publisher;
        }

        public static EventPublishingMethodInterceptor of(EventPublishingMethod eventMethod, ApplicationEventPublisher publisher) {
            return new EventPublishingMethodInterceptor(eventMethod, publisher);
        }

        @Nullable
        public Object invoke(MethodInvocation invocation) throws Throwable {
            Object result = invocation.proceed();
            if (!EventPublishingRepositoryProxyPostProcessor.isEventPublishingMethod(invocation.getMethod())) {
                return result;
            } else {
                Iterable<?> arguments = EventPublishingRepositoryProxyPostProcessor.asIterable(invocation.getArguments()[0], invocation.getMethod());
                this.eventMethod.publishEventsFrom(arguments, this.publisher);
                return result;
            }
        }
    }
}

스프링의 aop를 사용해 이벤트 처리 로직을 분리하고, 도메인 로직에만 집중할 수 있게 되었다.

다만, 이벤트가 발행되는 조건에 대해선 조금 조심해야한다.

이벤트 발행 조건

• Spring Data JPA repostiory를 사용해야 한다.
• save~, delete, deleteAll, deleteInBatch, deleteAllInBatch 가 명시적으로 호출되어야만 이벤트가 발행된다.
  • 명시적으로 save를 호출하지 않으면 트랜잭션 내부에서 변경이 감지되어 DB에 반영이 되었지만 이벤트가 발행되지 않을 수도 있다.

테스트

@SpringBootTest
@TestConstructor(autowireMode = TestConstructor.AutowireMode.ALL)
class ProductTest(
    private val repository: ProductRepository,
) {
    /***
     * 기존 애플리케이션 컨텍스트에 등록된 빈을 mock으로 대체
     */
    @MockitoBean
    private lateinit var handler: ProductPriceChangedEventListener

    @Test
    fun 상품_가격변경_이벤트_성공() {
        val product = Product(param = ProductParam(name = "Real MySQL", price = 10000, imageUrls = listOf()))
        repository.save(product)

        val newProduct = repository.findById(product.id).get()
        newProduct.updatePrice(50000)
        repository.save(newProduct)

        verify(handler, times(1)).handlerProductPriceChanged(any())
    }

    @Test
    fun 상품_가격변경_이벤트_발행_실패() {
        val product = Product(param = ProductParam(name = "Real MySQL", price = 10000, imageUrls = listOf()))
        repository.save(product)

        val newProduct = repository.findById(product.id).get()
        newProduct.updatePrice(50000)

        verifyNoInteractions(handler)
    }
}

@SpringBootTest
@TestConstructor(autowireMode = TestConstructor.AutowireMode.ALL)
class ProductServiceTest(
    private val service: ProductService,
    private val repository: ProductRepository
) {
    /***
     * 기존 애플리케이션 컨텍스트에 등록된 빈을 mock으로 대체
     */
    @MockitoBean
    private lateinit var handler: ProductPriceChangedEventListener

    @Test
    fun 상품_가격_변경_이벤트_save를_호출하지_않았을경우() {
        val product = service.create(ProductParam(name = "치즈감자전", price = 10000, imageUrls = listOf()))
        service.updatePrice(product.id, 30000)

        val updatedProduct = repository.findById(product.id).get()
        assertEquals(updatedProduct.price, 30000L)

        verifyNoInteractions(handler)
    }

}

인사이트

AbstractAggregateRoot를 실무에서 직접 사용해본적이 없지만, 적극적으로 Spring Data JPA를 사용하는 경우라면 도입하기 좋을 것 같다. 사실 AbstractAggregateRoot를 상속받지 않아도 @DomainEvent와 @AfterDomainEventPublication만으로도 충분히 구현 가능하다.(이 경우 이벤트 중복 발행 방지를 위해 꼭 clear를 해줘야한다.)

또한, EventListener 설정에 따라 트랜잭션 컨텍스트를 유지할 수도 있고, 새로운 트랜잭션에서 시작할 수도 있는데다 @Async를 활용해 비동기로 동작하게 할 수도 있으니 원하는 동작 방식에 따라 꼭 테스트를 해본 다음 도입해야겠다는 생각이 들었다.

참고

• https://www.baeldung.com/spring-data-ddd
• https://mangkyu.tistory.com/318
• https://jeong-pro.tistory.com/250