[Camel in Action] 12-2. Camel Wire Tap과 Event Sourcing 패턴

Table of Contents

Wire Tap이란?

Wire Tap은 메시지 처리 파이프라인에서 메시지의 복사본을 다른 곳으로 보내는 EIP입니다. 원본 메시지 흐름에 영향을 주지 않고 감사 로그, 모니터링, 분석을 위한 복사본을 만들 수 있습니다.

Wire Tap 기본 사용

from("activemq:queue:orders")
  .wireTap("activemq:topic:order-audit") // 감사 로그로 복사
  .to("bean:orderService");              // 원본은 계속 처리

Wire Tap은 기본적으로 비동기로 동작합니다. 원본 라우트는 Wire Tap 처리 완료를 기다리지 않고 계속 진행합니다.

Wire Tap으로 메시지 변환 후 전송

감사 목적으로 메시지를 변환해서 보낼 때는 onPrepare를 사용합니다.

from("direct:processPayment")
  .wireTap("direct:auditLog")
    .onPrepareRef("auditPrepare") // 복사본 변환 빈
  .to("bean:paymentService");

// 감사 메시지 준비 빈
@Component("auditPrepare")
public class AuditPrepare implements Processor {
  @Override
  public void process(Exchange exchange) {
    AuditLog log = new AuditLog();
    log.setTimestamp(LocalDateTime.now());
    log.setPayload(exchange.getIn().getBody(String.class));
    log.setUser(exchange.getIn().getHeader("userId", String.class));
    exchange.getIn().setBody(log);
  }
}

Event Sourcing 패턴

Event Sourcing은 시스템 상태를 현재값이 아닌 이벤트 이력으로 저장하는 패턴입니다. 모든 변경이 이벤트로 기록되어 언제든 특정 시점의 상태를 재구성할 수 있습니다.

// 모든 상태 변경을 이벤트로 저장
from("direct:updateOrder")
  .process(exchange -> {
    Order order = exchange.getIn().getBody(Order.class);
    OrderEvent event = new OrderEvent("ORDER_UPDATED", order, LocalDateTime.now());
    exchange.getIn().setHeader("domainEvent", event);
  })
  .wireTap("direct:saveEvent")  // 이벤트 저장 (비동기)
  .to("bean:orderService?method=update");

from("direct:saveEvent")
  .setBody(header("domainEvent"))
  .marshal().json()
  .to("kafka:order-events"); // Kafka에 이벤트 영구 저장

이벤트 스토어에서 상태 재구성

Event Sourcing의 핵심 이점은 이벤트 재생(event replay)을 통해 어느 시점의 상태든 재구성할 수 있다는 것입니다.

from("direct:replayOrderHistory")
  .process(exchange -> {
    String orderId = exchange.getIn().getHeader("orderId", String.class);
  })
  .to("kafka:order-events?groupId=replay-&{header.orderId}")
  .aggregate(header("orderId"), new OrderStateAggregator())
    .completionTimeout(5000)
  .to("bean:orderStateService?method=updateProjection");