Service Activator와 Message Handler - Service Activator를 사용한 비즈니스 로직 구현

2. Spring Integration 주요 컴포넌트 활용
2.1 Message Filter와 Router
2.2 Service Activator와 Message Handler

Spring Integration for Backend Developers

Service Activator를 사용한 비즈니스 로직 구현

1.1 Service Activator란?

Service Activator는 Spring Integration에서 메시지를 처리하기 위한 비즈니스 로직을 실행하는 엔드포인트(Endpoint)입니다.

• 목적: 메시지를 소비하고, 메시지의 Payload 또는 Header를 기반으로 비즈니스 로직을 실행.
• 사용 방식:
  • POJO 기반의 메서드에 메시지를 전달.
  • 비즈니스 로직 실행 후 결과를 반환하거나 메시지 흐름을 종료.

1.2 Service Activator의 주요 특징

1. 메시지 처리:
   • 메시지의 Payload와 Header를 읽어 비즈니스 로직을 수행.
   • 처리 결과를 반환하여 다음 단계로 전달하거나 흐름을 종료.
2. 유연성:
   • POJO의 메서드를 호출하므로 기존 비즈니스 로직과 쉽게 통합 가능.
3. 주요 어노테이션:
   • @ServiceActivator: 메서드를 메시지 엔드포인트로 정의.
   • XML 기반 설정에서도 <int:service-activator> 사용 가능.

1.3 Service Activator 기본 구성

1. 입력 채널: 메시지가 전달되는 채널.
2. Service Activator: 메시지를 소비하고 비즈니스 로직을 실행하는 메서드.
3. 출력 채널(선택적): 처리 결과를 전달하는 채널.

1.4 예제: 주문 처리 시스템

요구사항

1. 주문 메시지를 처리하여 상태를 업데이트.
2. 처리 결과를 다른 채널로 전달.

1. Service Activator 설정

package kr.co.thekeytech.spring.eai.activator;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.messaging.MessageChannel;

@Configuration
public class OrderService {

    @Bean
    public MessageChannel orderInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageChannel orderOutputChannel() {
        return new DirectChannel();
    }

    private static final Logger logger = LoggerFactory.getLogger(OrderService.class);

    @ServiceActivator(inputChannel = "orderInputChannel", outputChannel = "orderOutputChannel")
    public String processOrder(String orderPayload) {
        logger.info("OrderService Processed Order: {}", orderPayload);
        return "OrderService UpdatedOrder, " + orderPayload;
    }
}

 

설명:

• orderInputChannel: 주문 메시지를 전달받는 채널.
• orderOutputChannel: 처리 결과를 전달하는 채널.
• @ServiceActivator: processOrder 메서드를 Service Activator로 설정.

2. Gateway 설정

package kr.co.thekeytech.spring.eai.activator;

import org.springframework.integration.annotation.Gateway;
import org.springframework.integration.annotation.MessagingGateway;

@MessagingGateway
public interface OrderGateway {
    @Gateway(requestChannel = "orderInputChannel")// inputChannel을 명시적으로 지정
    void sendMessage(String message);
}

3. 메시지 전송

Service Activator 테스트 코드

package kr.co.thekeytech.spring.eai.activator;

import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;

import static org.assertj.core.api.Assertions.assertThat;

@SpringBootTest
class OrderFlowConfigTest {

    @Autowired
    private OrderGateway orderGateway;

    @Autowired
    private MessageChannel orderOutputChannel;

    @Test
    void testServiceActivator() {
        QueueChannel outputQueue = new QueueChannel();
        ((DirectChannel) orderOutputChannel).subscribe(outputQueue::send);

        // 주문 메시지 생성 및 전송
        orderGateway.sendMessage("Order ID: 12345");

        // 결과 출력
        Message<?> outputMessage = outputQueue.receive(1000);
        assertThat(outputMessage).isNotNull();
        System.out.println("Final Service Activator Output: " + outputMessage.getPayload());
    }
}

728x90

4. Integration Flow로 구현

orderFlow가 없어도 @ServiceActivator만으로 충분히 동작합니다. 그러나 다음과 같은 경우 orderFlow를 사용하는 것이 더 적합합니다:

• 복잡한 메시지 처리 로직(조건부 흐름, 데이터 변환, 다중 처리 등)이 필요한 경우.
• POJO 메서드 호출 외의 핸들러를 사용해야 하는 경우.
• 흐름을 명시적으로 정의하고자 하는 경우.

package kr.co.thekeytech.spring.eai.activator;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.dsl.IntegrationFlow;

@Configuration
public class OrderFlowConfig {
    private static final Logger logger = LoggerFactory.getLogger(OrderFlowConfig.class);

    @Bean
    public IntegrationFlow orderFlow(OrderService orderService) {
        return IntegrationFlow.from("activator.order.input.channel")
                .handle((payload, headers) -> {
                    logger.info("###########=> Before Handle OrderFlow: {}", payload);
                    return payload;
                })
                .transform(String.class, String::toUpperCase)
                .handle(orderService, "processOrder") // Service Activator 호출
                .handle((payload, headers) -> {
                    logger.info("###########=> After Handle OrderFlow: {}", payload);
                    return payload;
                })
                .channel("activator.order.output.channel")
                .get();
    }
}

 

설명:

• Service Activator에서 사용하는 InputChannel, OutputChannel 대신 별도의 채널 "activator.order.input.channel", "activator.order.output.channel"을 선언하여 사용합니다.
• 초기 로깅 Handle을 지정합니다.
• 메시지 변환(transform)으로  payload내용을 대문자로 치환합니다.
• Service  Activator 메서드를 호출합니다.
• 후처리 로깅 Handle을 지정합니다.
• 출력 채널을 "activator.order.output.channel"을 지정합니다.

5. Gateway 설정

IntegrationFlow용 "sendMessageWithFlow" gateway을 설정합니다.

package kr.co.thekeytech.spring.eai.activator;

import org.springframework.integration.annotation.Gateway;
import org.springframework.integration.annotation.MessagingGateway;

@MessagingGateway
public interface OrderGateway {
    @Gateway(requestChannel = "orderInputChannel")// inputChannel을 명시적으로 지정
    void sendMessage(String message);

    @Gateway(requestChannel = "activator.order.input.channel")// inputChannel을 명시적으로 지정
    void sendMessageWithFlow(String message);
}

6. 메시지 전송

IntegrationFlow  테스트 코드

package kr.co.thekeytech.spring.eai.activator;

import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;

import static org.assertj.core.api.Assertions.assertThat;

@SpringBootTest
class OrderFlowConfigTest {

    @Autowired
    private OrderGateway orderGateway;

    @Autowired
    @Qualifier("activator.order.output.channel")
    private MessageChannel orderOutputFlowChannel;

    @Test
    void testFlow() {
        QueueChannel outputQueue = new QueueChannel();
        ((DirectChannel) orderOutputFlowChannel).subscribe(outputQueue::send);

        // 주문 메시지 생성 및 전송
        orderGateway.sendMessageWithFlow("Order ID: 67890");

        // 결과 출력
        Message<?> outputMessage = outputQueue.receive(1000);
        assertThat(outputMessage).isNotNull();
        System.out.println("Final Flow Output: " + outputMessage.getPayload());
    }
}