Spring Integration Mail: отправка электронной почты после всех вставок базы данных

Question

Здравствуйте, у меня есть поток интеграции, который разбивает файл на строку, преобразует каждую строку в POJO, а затем вставляет этот POJO в базу данных через исходящий шлюз JDB C.

Я хочу иметь возможность отправить одно электронное письмо после завершения процесса файла. В настоящее время я отправляю на канал smtpFlow после моего jdbcOutboundGateway, однако это отправляет электронное письмо после каждой вставки БД.

Вот мой текущий поток DSL

IntegrationFlow ftpFlow() {
    return IntegrationFlows.from(
            ftpSource(), spec -> spec.poller(Pollers.fixedDelay(5, TimeUnit.SECONDS)))
            .split(splitFile())
            .transform(this::transformToIndividualScore)
            .handle(jdbcOutboundGateway(null))
            .channel("smtpFlow")
            .get();

Как получить этот поток для отправки только одного электронного письма после обработки всех файлов в jdbcOutboundGateway?

Вот мой splitFile() метод

@Bean
FileSplitter splitFile() {
    FileSplitter fs = new FileSplitter(true, false);
    fs.setFirstLineAsHeader("IndividualScore");
    return fs;

Вот мой transformToIndividualScore метод

@Transformer
private IndividualScore transformToIndividualScore(String payload) {
    String[] values = payload.split(",");
    IndividualScore is = new IndividualScore();
    is.setScorecardDate(values[0]);
    is.setVnSpId(values[1]);
    is.setPrimaryCat(values[2]);
    is.setSecondaryCat(values[3]);
    is.setScore(Integer.parseInt(values[4]));
    is.setActual(values[5]);
    return is;
}

Answer 1

Так что решение моей проблемы (вроде).

Маркировка итератора для false на моем FileSplitter теперь позволяет использовать заголовки последовательности.

обновленный splitFile() ниже

@Bean
FileSplitter splitFile() {
    FileSplitter fs = new FileSplitter(false, false);
    fs.setFirstLineAsHeader("IndividualScore");
    fs.setApplySequence(true);
    return fs;
}

Моя интуиция говорит мне что стратегия выпуска по умолчанию .aggregate() должна заключаться в том, что заголовок сообщения sequenceSize == объединяет список сообщений.

При создании FileSplitter с iterator, установленным на true, sequenceSize устанавливается на 0, что никогда не будет соответствовать стратегии выпуска по умолчанию .aggregate()

Однако это заставляет FileSplitter использовать List для хранения всех строк файла в памяти. Агрегатор также хранит в памяти еще ArrayList строк.

Есть ли лучшее решение для создания собственного агрегатора, который обрабатывает END FileMarker, что позволяет использовать итератор для разбиения файл

Answer 2

с помощью @ ArtemBilan

Я смог использовать методы publishSubscribeChannel() и chain 2 subscribe() в следующей последовательности: новый IntegrationFlow

 @Bean
IntegrationFlow ftpFlow() {
    return IntegrationFlows.from(
            ftpSource(), spec -> spec.poller(Pollers.fixedDelay(5, TimeUnit.SECONDS)))
            .publishSubscribeChannel(channel -> channel
                    .subscribe(
                        a -> a
                                .split(splitFile())
                                .transform(this::transformToIndividualScore)
                                .handle(jdbcMessageHandler(null)))
                    .subscribe(
                        b -> b
                                .transform(this::transformToSuccessEmail)
                                .handle(emailHandler()))
            )
            .get();

Answer 3

Добавьте .aggregate() после дескриптора, чтобы собрать результаты разбиений обратно в одно сообщение.