简单的文件系统整合流

<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>

创建一个消息网关

package cc.conyli.restlearn.integration;

import org.springframework.integration.annotation.MessagingGateway;
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);
}

1. @MessagingGateway是用于在运行时生成这个接口的实现类的一个Bean，就像Spring Data在运行时自动生成神奇接口对应的实现类一样。在需要的地方进行依赖注入这个接口就可以了。
2. (defaultRequestChannel = "textInChannel")是说对于writeToFile的调用，都往名称叫做textInChannel的管道里去放。
3. writeToFile()这个方法，接受一个@Header注解的文件名，这里表示文件名的信息，实际上会被放到消息的头部中叫做file_name的键中，而不是放在报文（payload）中，后边的data表示需要写入到文件中的内容，这部分是放在消息的报文中的。

1. XML配置
2. Java配置
3. 使用DSL的Java配置

XML配置

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:int="http://www.springframework.org/schema/integration"
       xmlns:int-file="http://www.springframework.org/schema/integration/file"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
        http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/integration
        http://www.springframework.org/schema/integration/spring-integration.xsd
        http://www.springframework.org/schema/integration/file
        http://www.springframework.org/schema/integration/file/spring-integration-file.xsd">

    <int:channel id="textInChannel"/>

    <int:transformer id="upperCase"
                     input-channel="textInChannel"
                     output-channel="fileWriterChannel"
                     expression="payload.toUpperCase()"/>

    <int:channel id="fileWriterChannel"/>

    <int-file:outbound-channel-adapter id="writer"
                                       channel="fileWriterChannel"
                                       directory="/tmp/sia5/files"
                                       mode="APPEND"
                                       append-new-line="true"/>
</beans>

1. <int:channel id="textInChannel"/>定义了管道的名称，这个名称和消息网关里注解里的管道名称要相符，如果多个管道，名称不能相同。
2. <int:transformer id="upperCase" ...的部分实际上是一个消息处理器，接受textInChannel传进来的消息，进行大写处理，然后输出给fileWriterChannel
3. <int:channel id="fileWriterChannel"/>又定义了一个叫做fileWriterChannel的管道，就是上一个配置里接受输出的管道。
4. <int-file:outbound-channel-adapter id="writer" ...是对刚才的fileWriterChannel管道的详细配置，注意这里的命名空间换成了int-file，这是Spring Integration规定的写文件的管道命名空间。其中定义了管道名称，然后设置了directory，会在这个目录下，按照消息头部的文件名，将消息报文中的数据写入这个文件里。mode和append-new-line的设置表示如果文件存在，则开始新行并向文件的尾部追加。

package cc.conyli.restlearn.config;

import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.ImportResource;

@Configuration
@ImportResource(locations = "classpath:/filewriterconfig.xml")
public class FileWriterXMLConfig {
}

Java配置

package cc.conyli.restlearn.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.annotation.Transformer;
import org.springframework.integration.file.FileWritingMessageHandler;
import org.springframework.integration.file.support.FileExistsMode;
import org.springframework.integration.transformer.GenericTransformer;

import java.io.File;

@Configuration
public class FileWriterJavaConfig {

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

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

@Bean
public MessageChannel textInChannel() {
    return new DirectChannel();
}
...
@Bean
public MessageChannel fileWriterChannel() {
    return new DirectChannel();
}

Spring Integration DSL

package cc.conyli.restlearn.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.MessageChannels;
import org.springframework.integration.file.dsl.Files;
import org.springframework.integration.file.support.FileExistsMode;

import java.io.File;

@Configuration
public class FileWriterDSLConfig {

    @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();
    }
}

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

Spring Integration 管道组件

1. Channels，把消息从一个元素发到另一个元素
2. Filters，有条件的允许消息通过
3. Transformers，根据设置好的条件修改消息
4. Routers，通常根据消息的头部信息，将消息发到不同的channel里
5. Splitters，将消息拆成两个或者更多，发送到不同的channel里
6. Aggregators，与Splitters相反，将来自不同channel的消息组装成一个消息
7. Service activators，将消息交给一个Java方法进行处理，然后把结果输出到channel中
8. Channel adapters，将管道连接到外部系统，可以接受或者发送数据
9. Gateways，将消息发送到一个整合流中

Message Channels

1. PublishSubscribeChannel，发布订阅模式管道，将消息发给一个或者多个组件
2. QueueChannel，队列模式，发布的消息会存在一个队列里，直到有人来拿走，如果有多个消费者，只有一个消费者可以拿到某一个数据
3. PriorityChannel，优先级管道，和队列模式相似，区别是依据消息头部的优先级来决定哪些消息先被送出去
4. RendezvousChannel，与队列模式相似，但是发送者会阻塞发送直到消息被拿走，实际是一个同步的方式，有一个消息就必须被消费掉才能发下一个
5. DirectChannel，像发布订阅管道，但是会发送一个单独的消息给在同一个线程里的消费者，可以让数据传输跨channel（没看懂）
6. ExecutorChannel，像DirectChannel，但是消息由一个TaskExecutor分发。
7. FluxMessageChannel，一个基于Project Reactor's Flux的响应式管道，将在第十章来讲Spring 5新增的Reactive系列玩意。

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

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

Filters

@Filter(inputChannel = "firstChannel", outputChannel = "textInChannel")
public boolean isContentH(String string) {
    return string.contains("H");
}

.<Integer>filter((p) -> p % 2 == 0)

Transformers

Routers

@Bean
@Router(inputChannel = "numberChannel")
public AbstractMessageRouter evenOddRouter() {
    return new AbstractMessageRouter() {
        @Override
        protected Collection<MessageChannel> determineTargetChannels(Message<?> message) {
            Integer number = (Integer) message.getPayload();
            if (number % 2 == 0) {
                return Collections.singleton(evenChannel());
            }
            return Collections.singleton(oddChannel());
        }
    };
}

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

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

Splitters

1. 一个信息体中是一个相同类型数据的集合，将其分拆成多个单独的数据，或者只想处理其中的一个。
2. 一个信息体中包含多个相关的信息，但是可以拆分成不同的不同类型的对象。

@Splitter(inputChannel="lineItemsChannel", outputChannel="lineItemChannel")
public List<LineItem> lineItemSplitter(List<LineItem> lineItems) {
    return lineItems;
}

Service activators

@Bean
@ServiceActivator(inputChannel="someChannel")
public MessageHandler sysoutHandler() {
    return message -> {
        System.out.println("Message payload: " + message.getPayload());
    };
}

@Bean
@ServiceActivator(inputChannel="orderChannel", outputChannel="completeOrder")
public GenericHandler<Order> orderHandler(OrderRepository orderRepo) {
    return (payload, headers) -> {
        return orderRepo.save(payload);
    };
}

Gateway

@Component
@MessagingGateway(defaultRequestChannel="inChannel", defaultReplyChannel="outChannel")
public interface UpperCaseGateway {
    String uppercase(String in);
}

Channel adapters

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

Endpoint modules

整合邮件流

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

package cc.conyli.restlearn.entity;

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;

@Data
@ConfigurationProperties(prefix = "mymock.email")
@Component
public class EmailProperties {
    private String username;
    private String password;
    private String host;
    private String mailbox;
    private long pollRate = 30000;

    public String getImapUrl() {
        return String.format("imaps://%s:%s@%s/%s", this.username, this.password, this.host, this.mailbox);
    }
}

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-configuration-processor</artifactId>
    <optional>true</optional>
</dependency>

mymock.email.host=imap.vip.sina.com
mymock.email.mailbox=""
mymock.email.username=lee0709@vip.sina.com
mymock.email.password=********
mymock.email.poll-rate=30000

1. 将流创建在当前应用的内部
2. 将流创建为一个独立的应用