[스프링인액션] 스프링 통합 플로우 사용

1implementation 'org.springframework.boot:spring-boot-starter-integration'
2implementation 'org.springframework.integration:spring-integration-file'

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

 1@Configuration
 2public class FileWriterIntegrationConfig {
 3
 4  @Bean
 5  @Transformer(inputChannel="textInChannel", outputChannel="fileWriterChannel")
 6  public GenericTransformer<String, String> upperCaseTransformer() {
 7      return text -> text.toUpperCase();
 8  }
 9
10  @Bean
11  @ServiceActivator(inputChannel="fileWriterChannel")
12  public FileWritingMessageHandler fileWriter() {
13      FileWritingMessageHandler handler = new FileWritingMessageHandler(new File("/tmp/sia5/files"));
14      handler.setExpectReply(false);
15      handler.setFileExistsMode(FileExistsMode.APPEND);
16      handler.setAppendNewLine(true);
17      return handler;
18  }
19}

 1@Configuration
 2public class FileWriterIntegrationConfig {
 3
 4  @Bean
 5  public IntegrationFlow fileWriterFlow() {
 6    return IntegrationFlows
 7        .from(MessageChannels.direct("textInChannel"))        // 입력 채널
 8        .<String, String>transform(t -> t.toUpperCase())      // 변환기 선언
 9        .channel(MessageChannels.direct("fileWriterChannel")) // 파일-쓰기 채널
10        .handle(Files                                         // 아웃바운드 채널 어뎁터
11            .outboundAdapter(new File("/tmp/sia5/files"))
12            .fileExistsMode(FileExistsMode.APPEND)
13            .appendNewLine(true))
14        .get(); 
15  }
16
17}

1@Filter(inputChannel="numberChannel", outputChannel="evenNumberChannel")
2public boolean evenNumberFilter(Integer number) {
3    return number % 2 == 0;
4}

1@Bean 
2public IntegrationFlow evenNumberFlow(AtomicInteger integerSource) {
3  return IntegrationFlows
4      .<Integer>filter((p) -> p % 2 == 0)
5      .get();
6}

1@Bean
2@Transformer(inputChannel="numberChannel", outputChannel="romanNumberChannel")
3public GenericTransformer<Integer, String> romanNumTransformer() {
4    return RomanNumbers::toRoman;
5}

1@Bean 
2public IntegrationFlow transformerFlow() {
3    return IntegrationFlows
4        .transform(RomanNumbers::toRoman)
5        .get();
6}

 1@Bean
 2@Router(inputChannel="numberChannel")
 3public AbstractMessageRouter evenOddRouter() {
 4  return new AbstractMessageRouter() {
 5    @Override
 6    protected Collection<MessageChannel> determineTargetChannels(Message<?> message) {
 7      Integer number = (Integer) message.getPayload();
 8      if (number % 2 == 0) {
 9        return Collections.singleton(evenChannel());
10      }
11      return Collections.singleton(oddChannel());
12    }  
13  };
14}

 1@Bean 
 2public IntegrationFlow numberRoutingFlow(AtomicInteger source) {
 3  return IntegrationFlows
 4      .<Integer, String>route(n -> n%2==0 ? "EVEN":"ODD", mapping -> mapping
 5        .subFlowMapping("EVEN", 
 6            sf -> sf.<Integer, Integer>transform(n -> n * 10)
 7                      .handle((i, h) -> { ... })
 8            )
 9        .subFlowMapping("ODD",
10            sf -> sf.transform(RomanNumbers::toRoman)
11                      .handle((i, h) -> { ... })
12            )
13        )
14      .get();
15}

 1@Bean
 2@Splitter(inputChannel="poChannel" outputChannel="splitOrderChannel")
 3public OrderSplitter orderSplitter() {
 4    return new OrderSplitter();
 5}
 6
 7@Bean
 8@Router(inputChannel="splitOrderChannel")
 9public MessageRouter splitOrderRouter() {
10    PayloadTypeRouter router = new PayloadTypeRouter();
11    router.setChannelMapping(BillingInfo.class.getName(), "billingInfoChannel");
12    router.setChannelMapping(List.class.getName(), "lineItemsChannel");
13    return router;
14}

 1@Bean 
 2public IntegrationFlow numberRoutingFlow(AtomicInteger source) {
 3  return IntegrationFlows
 4      .split(orderSplitter())
 5      .<Object, String>route(
 6        p -> {
 7          if (p.getClass().isAssignableFrom(BillingInfo.class)) {
 8            return "BILLING_INFO";
 9          } else {
10            return "LINE_ITEMS";
11          }
12        }, mapping -> mapping
13        .subFlowMapping("BILLING_INFO", 
14            sf -> sf.<BillingInfo> handle((billingInfo, h) -> { ... }))
15        .subFlowMapping("LINE_ITEMS",
16            sf -> sf.split().<LineItem> handle((lineItem, h) -> { ... }))
17        )
18      .get();
19}

1@Bean
2@ServiceActivator(inputChannel="someChannel")
3public MessageHandler sysoutHandler() {
4  return message -> {
5    System.out.println("Message Payload: " + message.getPayload()); 
6  };
7}

1@Bean
2@ServiceActivator(inputChannel="orderChannel", outputChannel="completeChannel")
3public GenericHandler<Order> orderHandler(OrderRepository orderRepo) {
4  return (payload, headers) -> {
5    return orderRepo.save(payload);
6  };
7}

1@Bean 
2public IntegrationFlow someFlow() {
3  return IntegrationFlows
4      .handle(msg -> {
5        System.out.println("Message Payload: " + msg.getPayload());
6      })
7      .get();
8}

1@Bean 
2public IntegrationFlow orderFlow(OrderRepository orderRepo) {
3  return IntegrationFlows
4      .<Order>handle((payload, headers) -> {
5        return orderRepo.save(payload);
6      })
7      .get();
8}

1@Component
2@MessageGateway(defaultRequestChannel="inChannel", 
3                defaultReplyChannel="outChannel")
4public interface UpperCaseGateway {
5  String uppercase(String in);
6}

1@Bean 
2public IntegrationFlow uppercaseFlow() {
3  return IntegrationFlows
4      .from("inChannel")
5      .<String, String>transform(s -> s.toUpperCase())
6      .channel("outChannel")
7      .get();
8}

1@Bean
2@InboundChannelAdapter(poller=@Poller(fixedRate="1000"), channel="numberChannel")
3public MessageSource<Integer> numberSource(AtomicInteger source) {
4  return () -> {
5    return new GenericMessage<>(source.getAndIncrement());
6  };
7}

1@Bean 
2public IntegrationFlow someFlow(AtomicInteger integerSource) {
3  return IntegrationFlows
4      .from(integerSource, "getAndIncrement", 
5          c -> c.poller(Pollers.fixedRate(1000)))
6      .get();
7}