Spring XD - Streams

소개

Spring XD에서 가장 기본적인 stream은 source로부터 이벤트 드리븐 데이타를 획득하여 다수의 sink로 전달하는 것이다. Stream 프로세스는 XD Containter상에서 수행되며, Stream 정의는 XD Admin server를 통해 Container로 전달되면 배포가 완료된다. 

Source, Sink 그리고 Processor는 모듈의 기정의된 설정이다. 모듈 정의는 xd-root/xd/modules 하위 디렉토리에서 확인 할 수 있다. 모듈 정의는 일반적인 스프링 정의 파일이다. 정의에서는 EIP 패턴을 지원하기 위해 Spring integration의 Input/Output adapter와 Transformer와 같은 기존의 스프링 클래스들을 사용한다.

Stream을 정의하기 위해서는 고수준의 DSL을 사용한다. 

http | file

DSL은 UNIX의 파이프/필터 문법이랑 유사하다. 포트와 파일명을 위한 기본값이 오버라이드 되어 다음과 같이 사용된다.

http --port=8091 | file --dir=/tmp/httpdata/

이러한 스트림 정의를 위해서는 XD Admin Server로 HTTP POST 요청을 전달해야 한다.

간단한 스트림 만들기

XD Admin server는 stream 정의는 생명주기를 관리하기 위한 모든 RESTful API를 제공한다. 그러나 가장 쉬운 방법은 XD Shell을 사용하는 방식이다. Shell을 시작하는 간단한 방법은 Getting started를 참조하자.

새로운 스트림은 스트림 정의를 포스팅함으로서 생성된다. 정의는 단순한 DSL로 부터 생성된다. 예를 들어 다음과 같은 shell 명령을 수행해보자.

xd:> stream create --definition "time | log" --name ticktock

ticktock으로 명명된 DSL 표현식 time | log 에 기반한 정의를 생성한다. 파이프 심볼은 source와 sink를 연결하기 위해 사용한다.

그리고 스트림을 배포하기 위해서는 다음과 같은 커맨드를 사용한다.

xd:> stream deploy --name ticktock

스트림 서버는 module 디렉토리에서 time | log 정의를 발견하고 stream을 설정한다. 단순한 예제이지만 time 소스는 매초당 현재 시간을 log 싱크로 전송하고 싱크는 로깅 프레임워크를 사용하여 출력한다.

만약 스트림의 모듈의 다수 인스턴스를 생성하고 싶다면 프로퍼티를 포함하여 디플로이 커맨드를 할 수 있다.

xd:> stream deploy --name ticktock --properties "module.time.count=3"

또한 SpEL 표현식을 포함하여 모듈에 조건을 프로퍼티에 포함 시킬 수도 있다. 모듈의 인스턴스들은 표현식이 참으로 판단된 Containter에만 배포될 것이다.

xd:> stream deploy --name ticktock --properties "module.time.count=3,module.log.criteria=groups.contains('x')"

스트림 삭제하기

스트림 destory 커맨드를 통해서 모듈을 삭제할 수 있다.

xd:> stream destroy --name ticktock

스트림 디플로이/언디플로이

종종 스트림을 중지하기를 원하지만 미래에 사용하기 위해 정의를 유지하고 싶을때가 있다. 이러한 경우 undeploy 명령을 통해 스트림을 중지시키고 deploy 명령을 수행하여 다음에 재시작하면 된다.

xd:> stream undeploy --name ticktock

xd:> stream deploy --name ticktock

예제

Time 소스보다 좀더 복잡한 예제를 살펴보자. Http는 지원되는 또다른 소스 타입이다. HTTP POSTS를 통해서 데이터를 획득한다. Http 소스는 어드민 서버(기본 8080)랑 다른 포트(기본 9000)로 데이터를 취득한다는 것을 주의하자.

다음과 같은 커맨드를 수행해보자.

xd:> stream create --definition "http | log" --name myhttpstream --deploy

그리고 커맨드를 이용해 post 데이터를 전송해보자.

xd:> http post --target http://localhost:9000 --data "hello"

xd:> http post --target http://localhost:9000 --data "goodbye"

log 싱크에서는 결과를 출력으로 보낼 것이다.

스트림 처리

다음은 단순한 스트림 처리의 예이다. HTTP 포스트 데이터를 대문자로 변경하는 것이다.

xd:> stream create --definition "http | transform --expression=payload.toUpperCase() | log" --namemyprocstrem --deploy

DSL 문법

위의 예에서 소스와 싱크를 연결하기 위해서는 파이프 심볼 |을 사용하였다. 또한 소스와 싱크 설정에 파라미터를 전달할 수 있다. 만약 http 소스에서 포트를 8000으로 바꾸고 싶다면 다음과 같이 정의할 수 있다.

xd:> stream create --definition "http --port=8000 | log" --name myhttpstream

스프링 설정 파일에 대한 조금만 알고 있어도 모듈 정의에 해당 프로퍼티들이 노출되어 있다는 것을 알 수 있을 것이다.

복잡한 기능들

위에 제공된 예제에서는 각 스트림은 단순한 모듈 정의로 생성되었다. 하지만 모듈들은 메시징 미들웨어를 통한 데이터 전송 및 중복을 피하기위해 그룹핑 될 수 있다. Composing Modules 섹션에서 더 많은 특징에 대해 알 수 있다.

단순한 선형 처리를 대신해 방향성 있는 그래프 처리가 요구 될 수도 있다. 두 가지 기능이 여기에 연관된다. 첫번째는 채널로부터 복수의 플로우를 결합하기 위해서 named channel을 사용하는 것이다. 이러한 채널의 특징은 queue-based 또는 topic-based 채널이며, 접두어를 사용한다.(queue:myqueue, topic:mytopic) Named Channel 섹션에서 더 확인해보자. 두번째는 실행시 어떤 정보에 의해서 출력 채널을 결정해야할 경우가 있다. 이러한 경우에 대한 사항은 Dynamic Router 섹션에서 확인해보길 바란다.