[스프링 인 액션] 9.스프링 통합하기

• 이 장에서 배우는 내용
  실시간으로 데이터 처리하기
  통합 플로우 정의하기
  스프링 통합의 자바 DSL 정의 사용하기
  이메일과 파일 시스템 및 다른 외부 시스템과 통합하기

스프링 통합은 <Enterprise Integration Patterns 2003>에서 보여준 대부분의 통합 패턴을 사용할 수 있게 구현한 것이다.
각 통합 패턴은 하나의 컴포넌트로 구현되며, 이것을 통해서 파이프라인으로 메시지가 데이터를 운반한다.
스프링 구성을 사용하면 데이터가 이동하는 파이프라인으로 이런 컴포넌트들을 조립할 수 있다.

우선, 스프링 통합 사용 시의 기능과 특성을 보여주는 간단한 통합 플로우를 정의하는 것부터 시작해보자.

9.1 간단한 통합 플로우 선언하기

애플리케이션은 통합 플로우를 통해서 외부 리소스나 애플리케이션 자체에 데이터를 수신 또는 전송할 수 있으며,
스프링 통합은 이런 통합 플로우를 생성할 수 있게 해준다.

애플리케이션이 통합할 수 있는 그런 리소스 중 하나가 파일 시스템이다.
이에 따라 스프링 통합의 많은 컴포넌트 중에 파일을 읽거나 쓰는 채널 어댑터가 있다.

스프링 통합을 시작하기 위해 여기서는 파일 시스템에 데이터를 쓰는 통합 플로우를 생성할 것이다.

<dependency>
	<groupId>org.springframework.boot</groupId>
    	<artifactId>spring-boot-starter-integration</artifactId>
</dependency>

<dependency> 
	<groupId>org.springframework.integration</groupId> 
    	<artifactId>spring-integration-file</artifactId>
</dependency>

첫 번째 의존성은 스프링 통합의 스프링 부트 스타터이다.
통합하려는 플로우와 무관하게 이 의존성은 스프링 통합 플로우의 개발 시에 반드시 추가해야 한다.

두 번째 의존성은 스프링 통합의 파일 엔드포인트(endpoint) 모듈이다.
이 모듈은 외부 시스템 통합에 사용되는 24개 이상의 엔드포인트 모듈 중 하나다.

그 다음은 파일에 데이터를 쓸 수 있도록 애플리케이션에서 통합 플로우로 데이터를 전송하는
게이트웨이(gateway)를 생성해야 한다.

import org.springframework.integration.file.FileHeaders;
import org.springframework.messaging.handler.annotation.Header;

@MessagingGateway(defaultRequestChannel="textInChannel")
public interface FileWriterGateway {
	void writeToFile(
		@Header(FileHeaders.FILENAME) String filename, 
		String data
	);
}

• @MessagingGateway : FileWriterGateway 인터페이스의 구현체(클래스)를 런타임 시에 생성하라고 스프링 통합에 알려준다.
• defaultRequestChannel : 해당 인터페이스의 메서드 호출로 생성된 메시지가 이 속성에 지정된 메시지 채널로 전송된다는 것을 나타낸다.
  writeToFile()의 호출로 생긴 메시지가 textInChannel이라는 이름의 채널로 전송된다.
• @Header : filename에 전달되는 값이 메시지 페이로드가 아닌 메시지 헤더에 있다는 것을 나타낸다.

이제는 메시지 게이트웨이가 생성되었으므로 통합 플로우를 구성해야 한다.

통합 플로우는 다음 세 가지 구성 방법으로 정의할 수 있다.

1. XML 구성
2. 자바 구성
3. DSL을 사용한 자바 구성

파일-쓰기 통합 플로우

9.1.1 XML을 사용해서 통합 플로우 정의하기

XML 구성의 사용은 가급적 피하는게 좋다.

9.1.2 JAVA로 통합 플로우 구성하기

@Configuration
public class FileWriterIntegrationConfig {

  @Bean
  @Transformer(inputChannel="textInChannel", // Declares a transformer
      outputChannel="fileWriterChannel")
  public GenericTransformer<String, String> upperCaseTransformer() {
    return text -> text.toUpperCase(); }

  @Bean
  @ServiceActivator(inputChannel="fileWriterChannel")
  public FileWritingMessageHandler fileWriter() { // Declares a file writer
    FileWritingMessageHandler handler =
        new FileWritingMessageHandler(new File("/tmp/sia5/files"));
    
		handler.setExpectReply(false); 
		handler.setFileExistsMode(FileExistsMode.APPEND); 
		handler.setAppendNewLine(true);
    
		return handler;
  }
}

자바 구성에서는 두 개의 빈을 정의한다.

1. 변환기(GenericTransformer)

@Transformer를 지정하여 GenericTransformer가 textInChannel의 메시지를 받아서 fileWriterChannel로 쓰는 통합 플로우 변환기라는 것을 지정한다.

2. 파일-쓰기 메시지 핸들러(FileWritingMessageHandler)

@ServiceActivator을 지정하여 fileWriterChannel로부터 메시지를 받아서 FileWritingMessageHandler의 인스턴스로 정의된 서비스에 넘겨준다.

handler.setExpectReply(false)는 서비스에서 응답 채널(플로우의 업스트림 컴포넌트로 값이 반환될 수 있는 채널)을 사용하지 않음을 나타낸다.

해당 메서드를 호출하지 않았다면 통합 플로우가 정상적으로 작동하더라도 응답 채널이 구성되지 않았다는 로그 메시지들이 나타난다.

9.1.3 스프링 통합의 DSL 구성 사용하기

통합 플로우의 각 컴포넌트를 별도의 빈으로 선언하지 않고 전체 플로우를 하나의 빈으로 선언한다.

@Configuration
public class FileWriterIntegrationConfig {

    @Bean
    public IntegrationFlow fileWriterFlow() {
        return IntegrationFlows.from(MessageChannels.direct("textInChannel"))
            .<String, String>transform(t -> t.toUpperCase())
            .handle(Files
                .outboundAdapter(new File("/tmp/sia5/files"))
                .fileExistsMode(FileExistsMode.APPEND)
                .appendNewLine(true))
            .get();
    }
}

IntegrationFlows 클래스는 플로우를 선언할 수 있는 빌더 API를 시작시킨다.
channel() 메서드를 호출하여 해당 채널을 이름(textInChannel)으로 참조할 수 있다.
코드는 많이 줄었지만 코드의 가독성을 높이기 위해 들여쓰기를 잘 해야 한다.

9.2 스프링 통합의 컴포넌트 살펴보기

스프링 통합은 다수의 통합 시나리오를 갖는 많은 영역을 포함한다.
따라서 그 모든 것을 하나의 챕터에 포함시키려고 하는 것은 마치 코끼리를 봉투에 맞춰 넣으려고 하는 것과 같다.
통합 플로우는 하나 이상의 컴포넌트로 구성되며 다음과 같다.

1. 채널(Channel)
   : 한 요소로부터 다른 요소로 메시지를 전달한다.
2. 필터(Filter)
   : 조건에 맞는 메시지가 플로우를 통과하게 해준다.
3. 변환기(Transaformer)
   : 메시지 값을 변경하거나 메시지 페이로드의 타입을 다른 타입으로 변환한다.
4. 라우터(Router)
   : 여러 채널 중 하나로 메시지를 전달하며, 대개 메시지 헤더를 기반으로 한다.
5. 분배기(Splitter)
   : 들어오는 메시지를 두 개 이상의 메시지로 분할하며, 분할된 각 메시지는 다른 채널로 전송된다.
6. 집적기(Aggregator)
   : 분배기와 상반된 것으로 별개의 채널로부터 전달되는 다수의 메시지를 하나의 메시지로 결합한다.
7. 서비스 액티베이터(Service activator)
   : 메시지를 처리하도록 자바 메서드에 메시지를 넘겨준 후 메서드의 반환값을 출력 채널로 전송한다.
8. 채널 어댑터(Channel adapter)
   : 외부 시스템에 채널을 연결한다. 외부 시스템으로부터 입력을 받거나 쓸 수 있다.
9. 게이트웨이(Gateway)
   : 인터페이스를 통해 통합 플로우로 데이터를 전달한다.

9.2.1 메시지 채널 (= 통로)

메시지 채널은 통합 파리프라인을 통해서 메시지가 이동하는 수단이다.
즉, 채널은 스프링 통합의 다른 부분을 연결하는 통로다.

스프링 통합은 다음을 포함해서 여러 채널 구현체(클래스)를 제공한다.

• PublishSubscribeChannel
  : 전송되는 메시지는 하나 이상의 컨슈머로 전달된다. 컨슈머가 여럿일 때는 모든 컨슈머가 해당 메시지를 수신한다.
• QueueChannel
  : 전송되는 메시지는 FIFO 방식으로 컨슈머가 가져갈 때까지 큐에 저장된다. 컨슈머가 여럿일 때는 그중 하나의 컨슈머만 해당 메시지를 수신한다.
• PriorityChannel
  : QueueChannel과 유사하지만, FIFO 방식 대신 메시지의 priority 헤더를 기반으로 컨슈머가 메시지를 가져간다.
• RendezvousChannel
  : QueueChannel과 유사하지만, 전송자와 동일한 스레드로 실행되는 컨슈머를 호출하여 단일 컨슈머에게 메시지를 전송한다. 이 채널은 트랜잭션을 지원한다.
• ExecutorChannel
  : DirectChannel과 유사하지만, TaskExecutor를 통해서 메시지가 전송된다. (전송자와 다른 스레드에서 처리된다) 이 채널 타입은 트랜잭션을 지원하지 않는다.
• FluxMessageChannel
  : 프로젝트 리액터(Product Reactor)의 플럭스(Flux)를 기반으로 하는 리액티브 스트림즈 퍼블리셔(Reactive Streams Publisher) 채널이다.

// option1. PublishSubscribeChannel
@Bean
public MessageChannel orderChannel() {
  return new PublishSubscribeChannel();
}

// option2. QueueChannel
@Bean
public MessageChannel orderChannel() {
  return new QueueChannel();
}
}

통합 플로우 정의 > 채널 이름 참조

@ServiceActivator(inputChannel="orderChannel")

자바 DSL 구성

@Bean
public IntegrationFlow orderFlow() {
	return IntegrationFlows 
		...
		.channel("orderChannel")
		...
		.get();
}

폴링 처리

@ServiceActivator(inputChannel="orderChannel",
		  poller=@Poller(fixedRate="1000"))

매 1초(1000ms) 당 1번씩 읽을 메시지가 있는 지 확인한다.

9.2.2 필터

필터는 통합 파이프라인의 중간에 위치할 수 있으며, 플로우의 전 단계로부터 다음 단계로의 메시지 전달을 허용 또는 불허한다.

// option1. 어노테이션 기반 필터 설정
@Filter(inputChannel="numberChannel",
        outputChannel="evenNumberChannel")
public boolean evenNumberFilter(Integer number) {
  return number % 2 == 0;
}

// option2. 자바 DSL 구성 필터 설정
@Bean
public IntegrationFlow evenNumberFlow(AtomicInteger integerSource) {
    return IntegrationFlows 
        ...
        .<Integer>filter((p) -> p % 2 == 0) 
        ...
        .get();
}

9.2.3 변환기

변환기는 메시지 값의 변경이나 타입을 변환하는 일을 수행한다.

9.2.4 라우터

라우터는 전달 조건을 기반으로 통합 플로우 내부를 분기(서로 다른 채널로 메시지를 전달)한다.

예를 들어, 정수값을 전달하는 numberChannel이라는 이름의 채널이 있다고 하자.

그리고 모든 짝수 메시지를 evenChannel이라는 이름의 채널로 전달하고,
홀수 메시지는 oddChannel이라는 이름의 채널로 전달하도록 분기처리를 라우터가 한다.

9.2.5 분배기

때로는 통합 플로우에서 하나의 메시지를 여러 개로 분할하여 독립적으로 처리하는 것이 유용할 수 있다.

분배기를 사용할 수 있는 중요한 두 가지 경우가 있다.

• 메시지 페이로드가 같은 타입의 컬렉션 항목들을 포함하며, 각 메시지 페이로드 별로 처리하고자 할 때다
• 연관된 정보를 함께 전달하는 하나의 메시지 페이로드는 두 개 이상의 서로 다른 타입 메시지로 분할될 수 있다.
  예를 들어, 주문 데이터를 전달하는 메시지는 대금 청구 정보와 주문 항목 리스트의 두 가지 메시지로 분할할 수 있다.

9.2.6 서비스 액티베이터

서비스 액티베이터는 입력 채널로부터 메시지를 수신하고 이 메시지를 MessageHandler 인터페이스를 구현한 클래스(빈)에 전달한다.

9.2.7 게이트 웨이

게이트웨이는 애플리케이션이 통합 플로우로 데이터를 제출(submit)하고 선택적으로 플로우의 처리 결과인 응답을 받을 수 있는 수단이다.

이전의 본 FileWriterGateway는 단방향 게이트웨이며, 파일에 쓰기 위해 문자열을 인자로 받고 void를 반환하는 메서드를 갖고 있다.
양방향 게이트웨이의 작성도 어렵지 않으며, 이때는 게이트웨이 인터페이스를 작성할 때 통합 플로우로 전송할 값을 메서드에서 반환해야 한다.

9.2.8 채널 어댑터

채널 어댑터는 통합 플로우의 입구와 출구를 나타낸다.
데이터는 인바운드(inbound) 채널 어댑터를 통해 통합 플로우로 들어오고, 아웃바운드(outbound) 채널 어댑터를 통해 통합 플로우에서 나간다.

요약

• 스프링 통합은 플로우를 정의할 수 있게 해준다. 데이터는 애플리케이션으로 들어오거나 나갈 때 플로우를 통해 처리할 수 있다.
• 통합 플로우는 XML, Java, Java DSL을 사용해서 정의할 수 있다.
• 메시지 게이트웨이와 채널 어댑처는 통합 플로우의 입구나 출구의 역할을 한다.
• 메시지는 플로우 내부에서 변환, 분할, 집적, 전달될 수 있으며, 서비스 액티베이터에의해 처리될 수 있다.
• 메시지 채널은 통합 플로우의 컴포넌트들을 연결한다.