멀티 벤더 IoT 연동 설계하기 (2편) - Event-Driven Architecture로 확장하기

public void turnOnLight(Long deviceId, int brightness) {
    // 기기 제어
    adapter.executeCommand(...);

    // 알림 발송
    pushNotificationService.send(...);

    // 로그 저장
    deviceLogRepository.save(...);

    // 통계 업데이트
    statisticsService.update(...);

    // 슬랙 알림 (실패 시)
    if (failed) {
        slackService.send(...);
    }
}

@Service
@RequiredArgsConstructor
public class SmartHomeService {

    private final IoTAdapterFactory adapterFactory;
    private final DeviceRepository deviceRepository;
    private final PushNotificationService pushService;
    private final DeviceLogRepository logRepository;
    private final StatisticsService statisticsService;
    private final SlackService slackService;
    private final DataCollectionService dataCollectionService;
    // ... 의존성이 계속 늘어남

    @Transactional
    public void turnOnLight(Long deviceId, int brightness) {
        try {
            // 1. 기기 정보 조회
            Device device = deviceRepository.findById(deviceId)
                .orElseThrow();

            // 2. 기기 제어 (핵심 로직)
            IoTVendorAdapter adapter = adapterFactory.getAdapter(
                device.getVendorId()
            );
            adapter.executeCommand(device.getExternalId(), command);

            // 3. 상태 업데이트
            device.updateStatus(true, brightness);
            deviceRepository.save(device);

            // 4. 푸시 알림 발송
            pushService.send(
                device.getUserId(),
                "조명이 켜졌습니다."
            );

            // 5. 로그 저장
            DeviceLog log = DeviceLog.builder()
                .deviceId(deviceId)
                .action("TURN_ON")
                .brightness(brightness)
                .timestamp(LocalDateTime.now())
                .build();
            logRepository.save(log);

            // 6. 통계 업데이트
            statisticsService.incrementUsageCount(deviceId);
            statisticsService.updateBrightnessAverage(deviceId, brightness);

            // 7. 데이터 수집 (분석용)
            dataCollectionService.collect(
                device.getVendorId(),
                "TURN_ON",
                brightness
            );

        } catch (Exception e) {
            // 8. 실패 시 슬랙 알림
            slackService.sendError(
                "기기 제어 실패: " + deviceId,
                e.getMessage()
            );
            throw e;
        }
    }
}

graph TB
    subgraph Before["기존 (동기, 강결합)"]
        A1[기기 제어] --> B1[알림 발송]
        B1 --> C1[로그 저장]
        C1 --> D1[통계 업데이트]
        D1 --> E1[데이터 수집]
    end

    subgraph After["이벤트 기반 (비동기, 느슨한 결합)"]
        A2[기기 제어] --> Event[이벤트 발행]
        Event -.->|비동기| B2[알림 리스너]
        Event -.->|비동기| C2[로그 리스너]
        Event -.->|비동기| D2[통계 리스너]
        Event -.->|비동기| E2[데이터 리스너]
    end

    style Event fill:#4CAF50,color:#fff
    style A1 fill:#f44336,color:#fff
    style A2 fill:#2196F3,color:#fff

// 기기 제어 성공 이벤트
@Getter
@AllArgsConstructor
public class DeviceControlSuccessEvent {
    private final Long deviceId;
    private final String vendorId;
    private final CommandType commandType;
    private final Map<String, Object> parameters;
    private final LocalDateTime timestamp;

    public static DeviceControlSuccessEvent of(
        Device device,
        DeviceCommand command
    ) {
        return new DeviceControlSuccessEvent(
            device.getId(),
            device.getVendorId(),
            command.getType(),
            command.getParameters(),
            LocalDateTime.now()
        );
    }
}

// 기기 제어 실패 이벤트
@Getter
@AllArgsConstructor
public class DeviceControlFailureEvent {
    private final Long deviceId;
    private final String vendorId;
    private final CommandType commandType;
    private final String errorMessage;
    private final Exception exception;
    private final LocalDateTime timestamp;
}

@Service
@RequiredArgsConstructor
public class SmartHomeService {

    private final IoTAdapterFactory adapterFactory;
    private final DeviceRepository deviceRepository;
    private final ApplicationEventPublisher eventPublisher; // Spring 제공

    @Transactional
    public void turnOnLight(Long deviceId, int brightness) {
        try {
            // 1. 기기 정보 조회
            Device device = deviceRepository.findById(deviceId)
                .orElseThrow();

            // 2. 기기 제어 (핵심 로직)
            IoTVendorAdapter adapter = adapterFactory.getAdapter(
                device.getVendorId()
            );

            DeviceCommand command = DeviceCommand.builder()
                .type(CommandType.TURN_ON)
                .parameter("brightness", brightness)
                .build();

            adapter.executeCommand(device.getExternalId(), command);

            // 3. 상태 업데이트
            device.updateStatus(true, brightness);
            deviceRepository.save(device);

            // 4. 성공 이벤트 발행
            eventPublisher.publishEvent(
                DeviceControlSuccessEvent.of(device, command)
            );

        } catch (Exception e) {
            // 실패 이벤트 발행
            eventPublisher.publishEvent(
                new DeviceControlFailureEvent(
                    deviceId,
                    device.getVendorId(),
                    CommandType.TURN_ON,
                    e.getMessage(),
                    e,
                    LocalDateTime.now()
                )
            );
            throw e;
        }
    }
}

@Component
@RequiredArgsConstructor
@Slf4j
public class DeviceNotificationListener {

    private final PushNotificationService pushService;
    private final DeviceRepository deviceRepository;

    @Async
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void handleDeviceControlSuccess(DeviceControlSuccessEvent event) {
        try {
            Device device = deviceRepository.findById(event.getDeviceId())
                .orElseThrow();

            String message = createNotificationMessage(
                event.getCommandType(),
                event.getParameters()
            );

            pushService.send(device.getUserId(), message);

            log.info("Push notification sent: deviceId={}", event.getDeviceId());

        } catch (Exception e) {
            log.error("Failed to send push notification", e);
            // 알림 실패는 전체 프로세스에 영향 없음
        }
    }

    private String createNotificationMessage(
        CommandType commandType,
        Map<String, Object> parameters
    ) {
        return switch (commandType) {
            case TURN_ON -> "조명이 켜졌습니다.";
            case TURN_OFF -> "조명이 꺼졌습니다.";
            case SET_VALUE -> {
                int brightness = (int) parameters.get("brightness");
                yield "밝기가 " + brightness + "%로 설정되었습니다.";
            }
        };
    }
}

@Component
@RequiredArgsConstructor
@Slf4j
public class DeviceLogListener {

    private final DeviceLogRepository logRepository;

    @Async
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void handleDeviceControlSuccess(DeviceControlSuccessEvent event) {
        try {
            DeviceLog deviceLog = DeviceLog.builder()
                .deviceId(event.getDeviceId())
                .vendorId(event.getVendorId())
                .commandType(event.getCommandType())
                .parameters(event.getParameters())
                .success(true)
                .timestamp(event.getTimestamp())
                .build();

            logRepository.save(deviceLog);

            log.info("Device log saved: deviceId={}", event.getDeviceId());

        } catch (Exception e) {
            log.error("Failed to save device log", e);
        }
    }

    @Async
    @EventListener
    public void handleDeviceControlFailure(DeviceControlFailureEvent event) {
        try {
            DeviceLog deviceLog = DeviceLog.builder()
                .deviceId(event.getDeviceId())
                .vendorId(event.getVendorId())
                .commandType(event.getCommandType())
                .success(false)
                .errorMessage(event.getErrorMessage())
                .timestamp(event.getTimestamp())
                .build();

            logRepository.save(deviceLog);

        } catch (Exception e) {
            log.error("Failed to save failure log", e);
        }
    }
}

@Component
@RequiredArgsConstructor
@Slf4j
public class DeviceStatisticsListener {

    private final StatisticsService statisticsService;

    @Async
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void handleDeviceControlSuccess(DeviceControlSuccessEvent event) {
        try {
            statisticsService.incrementUsageCount(
                event.getDeviceId(),
                event.getCommandType()
            );

            if (event.getParameters().containsKey("brightness")) {
                int brightness = (int) event.getParameters().get("brightness");
                statisticsService.updateBrightnessAverage(
                    event.getDeviceId(),
                    brightness
                );
            }

            log.info("Statistics updated: deviceId={}", event.getDeviceId());

        } catch (Exception e) {
            log.error("Failed to update statistics", e);
        }
    }
}

@Component
@RequiredArgsConstructor
@Slf4j
public class DeviceErrorAlertListener {

    private final SlackService slackService;

    @Async
    @EventListener
    public void handleDeviceControlFailure(DeviceControlFailureEvent event) {
        try {
            String message = String.format(
                "기기 제어 실패\n" +
                "- 기기 ID: %d\n" +
                "- 제조사: %s\n" +
                "- 명령: %s\n" +
                "- 에러: %s",
                event.getDeviceId(),
                event.getVendorId(),
                event.getCommandType(),
                event.getErrorMessage()
            );

            slackService.sendToChannel("iot-alerts", message);

            log.info("Slack alert sent: deviceId={}", event.getDeviceId());

        } catch (Exception e) {
            log.error("Failed to send Slack alert", e);
        }
    }
}

@Async
@EventListener
public void handleEvent(DeviceControlSuccessEvent event) {
    // 별도 스레드에서 실행
}

@Configuration
@EnableAsync
public class AsyncConfig {

    @Bean
    public Executor asyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("event-async-");
        executor.initialize();
        return executor;
    }
}

@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
public void handleEvent(DeviceControlSuccessEvent event) {
    // 트랜잭션 커밋 후 실행
}

sequenceDiagram
    participant S as Service
    participant DB as Database
    participant E as EventPublisher
    participant L as Listener

    S->>DB: 1. 기기 제어
    S->>DB: 2. DB 저장 (device.save)
    S->>E: 3. 이벤트 발행
    Note over E: 이벤트 대기열에 등록
    S->>DB: 4. 트랜잭션 커밋
    DB-->>S: 커밋 완료
    E->>L: 5. 리스너 실행
    Note over L: @Async면 별도 스레드

@EventListener(condition = "#event.commandType.name() == 'TURN_ON'")
public void handleTurnOnOnly(DeviceControlSuccessEvent event) {
    // TURN_ON 명령만 처리
}

@EventListener(condition = "#event.vendorId == 'VENDOR_A'")
public void handleVendorAOnly(DeviceControlSuccessEvent event) {
    // A사 기기만 처리
}

@Order(1)
@EventListener
public void firstListener(DeviceControlSuccessEvent event) {
    // 먼저 실행
}

@Order(2)
@EventListener
public void secondListener(DeviceControlSuccessEvent event) {
    // 나중에 실행
}

기기 제어 → 이벤트 발행 → ??? → 알림이 안 감

@Aspect
@Component
@Slf4j
public class EventLoggingAspect {

    @AfterReturning("@annotation(org.springframework.context.event.EventListener)")
    public void logEventListenerExecution(JoinPoint joinPoint) {
        Object[] args = joinPoint.getArgs();
        if (args.length > 0 && args[0] != null) {
            log.info("Event listener executed: {} for event: {}",
                joinPoint.getSignature().getName(),
                args[0].getClass().getSimpleName()
            );
        }
    }

    @AfterThrowing(
        pointcut = "@annotation(org.springframework.context.event.EventListener)",
        throwing = "ex"
    )
    public void logEventListenerError(JoinPoint joinPoint, Exception ex) {
        log.error("Event listener failed: {} - {}",
            joinPoint.getSignature().getName(),
            ex.getMessage(),
            ex
        );
    }
}

이벤트 발행 순서: A → B → C
리스너 실행 순서: C → A → B (?)

@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
public void handleEvent(DeviceControlSuccessEvent event) {
    // 여기는 별도 트랜잭션? 아니면 원래 트랜잭션?
}

@Transactional(propagation = Propagation.REQUIRES_NEW)
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
public void handleEvent(DeviceControlSuccessEvent event) {
    // 새 트랜잭션에서 실행
}