From 43425863618b76bee70cf3a10f1f96dd33af3e5b Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 16 Mar 2021 15:06:36 -0400 Subject: [PATCH] INT-4116: Introduce FileAggregator (#3511) * INT-4116: Introduce FileAggregator JIRA: https://jira.spring.io/browse/INT-4116 * Implement a `FileSplitter.FileMarker`-based aggregation strategies and utilize them in a general `FileAggregator` component * Make `HeaderAttributeCorrelationStrategy.attributeName` as `final`; add `Assert.notEmpty()` * Fix `AggregatorFactoryBean` and `AggregatorSpec` to parse the provided processor for possible `CorrelationStrategy` and/or `ReleaseStrategy` * Introduce short-cut methods into Java & Kotlin DSL for an `aggregate()` configuration * Introduce a `FileHeaders.LINE_COUNT` for header to be populated in the `FileSplitter`. We need this info in the `FileAggregator` to avoid possible overhead with JSON deserialization of the `FileSplitter.FileMarker` messages * Test and document the feature * Improve `FileSplitter` doc for code block switch (tabs) * * Rework FileAggregator do not use Java Streams --- .../HeaderAttributeCorrelationStrategy.java | 12 +- .../config/AggregatorFactoryBean.java | 30 ++- .../integration/dsl/AggregatorSpec.java | 7 +- .../dsl/BaseIntegrationFlowDefinition.java | 11 + .../dsl/KotlinIntegrationFlowDefinition.kt | 10 +- .../integration/file/FileHeaders.java | 5 + .../FileAggregatingMessageGroupProcessor.java | 60 +++++ .../file/aggregator/FileAggregator.java | 78 +++++++ .../aggregator/FileMarkerReleaseStrategy.java | 63 ++++++ .../file/aggregator/package-info.java | 4 + .../file/splitter/FileSplitter.java | 10 +- .../file/aggregator/FileAggregatorTests.java | 146 ++++++++++++ .../file/aggregator/FileAggregatorTests.xml | 21 ++ .../aggregator/KotlinFileAggregatorTests.kt | 117 ++++++++++ src/reference/asciidoc/aggregator.adoc | 2 + src/reference/asciidoc/file.adoc | 208 +++++++++++++----- src/reference/asciidoc/splitter.adoc | 4 +- src/reference/asciidoc/whats-new.adoc | 6 + 18 files changed, 719 insertions(+), 75 deletions(-) create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/aggregator/FileAggregatingMessageGroupProcessor.java create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/aggregator/FileAggregator.java create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/aggregator/FileMarkerReleaseStrategy.java create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/aggregator/package-info.java create mode 100644 spring-integration-file/src/test/java/org/springframework/integration/file/aggregator/FileAggregatorTests.java create mode 100644 spring-integration-file/src/test/java/org/springframework/integration/file/aggregator/FileAggregatorTests.xml create mode 100644 spring-integration-file/src/test/kotlin/org/springframework/integration/kotlin/file/aggregator/KotlinFileAggregatorTests.kt diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/HeaderAttributeCorrelationStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/HeaderAttributeCorrelationStrategy.java index c647f4bf1a..2f7bf88aff 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/HeaderAttributeCorrelationStrategy.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/HeaderAttributeCorrelationStrategy.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -17,23 +17,25 @@ package org.springframework.integration.aggregator; import org.springframework.messaging.Message; +import org.springframework.util.Assert; /** - * Default implementation of {@link CorrelationStrategy}. Uses a header - * attribute to determine the correlation key value. + * Default implementation of {@link CorrelationStrategy}. + * Uses a provided header attribute to determine the correlation key value. * * @author Marius Bogoevici + * @author Artem Bilan */ public class HeaderAttributeCorrelationStrategy implements CorrelationStrategy { - private String attributeName; + private final String attributeName; public HeaderAttributeCorrelationStrategy(String attributeName) { + Assert.hasText(attributeName, "the 'attributeName' must not be empty"); this.attributeName = attributeName; } - public Object getCorrelationKey(Message message) { return message.getHeaders().get(this.attributeName); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java index c0e355f978..44fb0e934c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2020 the original author or authors. + * Copyright 2015-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -22,6 +22,7 @@ import java.util.Map; import java.util.function.Function; import org.aopalliance.aop.Advice; +import org.jetbrains.annotations.Nullable; import org.springframework.expression.Expression; import org.springframework.integration.aggregator.AbstractAggregatingMessageGroupProcessor; @@ -211,14 +212,15 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe } AggregatingMessageHandler aggregator = new AggregatingMessageHandler(outputProcessor); + JavaUtils.INSTANCE .acceptIfNotNull(this.expireGroupsUponCompletion, aggregator::setExpireGroupsUponCompletion) .acceptIfNotNull(this.sendTimeout, aggregator::setSendTimeout) .acceptIfNotNull(this.outputChannelName, aggregator::setOutputChannelName) .acceptIfNotNull(this.lockRegistry, aggregator::setLockRegistry) .acceptIfNotNull(this.messageStore, aggregator::setMessageStore) - .acceptIfNotNull(this.correlationStrategy, aggregator::setCorrelationStrategy) - .acceptIfNotNull(this.releaseStrategy, aggregator::setReleaseStrategy) + .acceptIfNotNull(obtainCorrelationStrategy(), aggregator::setCorrelationStrategy) + .acceptIfNotNull(obtainReleaseStrategy(), aggregator::setReleaseStrategy) .acceptIfNotNull(this.groupTimeoutExpression, aggregator::setGroupTimeoutExpression) .acceptIfNotNull(this.forceReleaseAdviceChain, aggregator::setForceReleaseAdviceChain) .acceptIfNotNull(this.taskScheduler, aggregator::setTaskScheduler) @@ -236,6 +238,28 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe return aggregator; } + @Nullable + private CorrelationStrategy obtainCorrelationStrategy() { + if (this.correlationStrategy == null && this.processorBean != null) { + CorrelationStrategyFactoryBean correlationStrategyFactoryBean = new CorrelationStrategyFactoryBean(); + correlationStrategyFactoryBean.setTarget(this.processorBean); + correlationStrategyFactoryBean.afterPropertiesSet(); + return correlationStrategyFactoryBean.getObject(); + } + return this.correlationStrategy; + } + + @Nullable + private ReleaseStrategy obtainReleaseStrategy() { + if (this.releaseStrategy == null && this.processorBean != null) { + ReleaseStrategyFactoryBean releaseStrategyFactoryBean = new ReleaseStrategyFactoryBean(); + releaseStrategyFactoryBean.setTarget(this.processorBean); + releaseStrategyFactoryBean.afterPropertiesSet(); + return releaseStrategyFactoryBean.getObject(); + } + return this.releaseStrategy; + } + @Override protected Class getPreCreationHandlerType() { return AggregatingMessageHandler.class; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/AggregatorSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/AggregatorSpec.java index fb8c3a831c..34a49460f0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/AggregatorSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/AggregatorSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2020 the original author or authors. + * Copyright 2016-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -69,7 +69,10 @@ public class AggregatorSpec extends CorrelationHandlerSpec aggregator.processor(aggregatorProcessor))} + * @param aggregatorProcessor the POJO representing aggregation strategies. + * @return the current {@link BaseIntegrationFlowDefinition}. + * @since 5.5 + * @see AggregatorSpec + */ + public B aggregate(Object aggregatorProcessor) { + return aggregate((aggregator) -> aggregator.processor(aggregatorProcessor)); + } + /** * Populate the {@link AggregatingMessageHandler} with provided options from {@link AggregatorSpec}. * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}. diff --git a/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinIntegrationFlowDefinition.kt b/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinIntegrationFlowDefinition.kt index f562e74be4..13614edacf 100644 --- a/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinIntegrationFlowDefinition.kt +++ b/spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinIntegrationFlowDefinition.kt @@ -1,5 +1,5 @@ /* - * Copyright 2020 the original author or authors. + * Copyright 2020-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -718,6 +718,14 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ this.delegate.resequence(resequencer) } + /** + * A short-cut for the `aggregate { processor(aggregatorProcessor) }` + * @since 5.5 + */ + fun aggregate(aggregator: Any) { + this.delegate.aggregate(aggregator) + } + /** * Populate the [AggregatingMessageHandler] with provided options from [AggregatorSpec]. * In addition accept options for the integration endpoint using [GenericEndpointSpec]. diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java index cb924aef61..afcdbd1842 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java @@ -47,6 +47,11 @@ public abstract class FileHeaders { */ public static final String MARKER = PREFIX + "marker"; + /** + * The line count for END marker message after splitting + */ + public static final String LINE_COUNT = PREFIX + "lineCount"; + /** * A remote file information representation */ diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/aggregator/FileAggregatingMessageGroupProcessor.java b/spring-integration-file/src/main/java/org/springframework/integration/file/aggregator/FileAggregatingMessageGroupProcessor.java new file mode 100644 index 0000000000..bc3bfab026 --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/aggregator/FileAggregatingMessageGroupProcessor.java @@ -0,0 +1,60 @@ +/* + * Copyright 2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.file.aggregator; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.Map; + +import org.springframework.integration.aggregator.AbstractAggregatingMessageGroupProcessor; +import org.springframework.integration.file.FileHeaders; +import org.springframework.integration.store.MessageGroup; +import org.springframework.messaging.Message; + +/** + * An {@link AbstractAggregatingMessageGroupProcessor} implementation for file content collecting + * previously splitted by the {@link org.springframework.integration.file.splitter.FileSplitter} + * with the {@code markers} option turned on. + *

+ * If no file markers present in the {@link MessageGroup}, then behavior of this processor is + * similar to the {@link org.springframework.integration.aggregator.DefaultAggregatingMessageGroupProcessor}. + *

+ * When no file content (only file markers are grouped), this processor emits an empty {@link ArrayList}. + * Note: with no file content and markers turned off, + * the {@link org.springframework.integration.file.splitter.FileSplitter} doesn't emit any messages + * for possible aggregation downstream. + * + * @author Artem Bilan + * + * @since 5.5 + */ +public class FileAggregatingMessageGroupProcessor extends AbstractAggregatingMessageGroupProcessor { + + @Override + protected Object aggregatePayloads(MessageGroup group, Map defaultHeaders) { + Collection> messages = group.getMessages(); + List payloads = new ArrayList<>(messages.size() - 2); + for (Message message : messages) { + if (!message.getHeaders().containsKey(FileHeaders.MARKER)) { + payloads.add(message.getPayload()); + } + } + return payloads; + } + +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/aggregator/FileAggregator.java b/spring-integration-file/src/main/java/org/springframework/integration/file/aggregator/FileAggregator.java new file mode 100644 index 0000000000..34ff8c8d2d --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/aggregator/FileAggregator.java @@ -0,0 +1,78 @@ +/* + * Copyright 2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.file.aggregator; + +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.BeanFactoryAware; +import org.springframework.integration.aggregator.CorrelationStrategy; +import org.springframework.integration.aggregator.HeaderAttributeCorrelationStrategy; +import org.springframework.integration.aggregator.MessageGroupProcessor; +import org.springframework.integration.aggregator.ReleaseStrategy; +import org.springframework.integration.file.FileHeaders; +import org.springframework.integration.file.splitter.FileSplitter; +import org.springframework.integration.store.MessageGroup; +import org.springframework.messaging.Message; + +/** + * A convenient component to utilize a {@link FileSplitter.FileMarker}-based aggregation logic. + * Implements all three {@link CorrelationStrategy}, {@link ReleaseStrategy} and {@link MessageGroupProcessor} + * for runtime optimization. + * Delegates to {@link HeaderAttributeCorrelationStrategy} with {@link FileHeaders#FILENAME} attribute, + * {@link FileMarkerReleaseStrategy} and {@link FileAggregatingMessageGroupProcessor}, respectively. + *

+ * The default {@link FileSplitter} behavior with markers enabled is do not provide a sequence details + * headers, therefore correlation in this aggregator implementation is done by the {@link FileHeaders#FILENAME} + * header which is still populated by the {@link FileSplitter} for each line emitted, including + * {@link FileSplitter.FileMarker} messages. + *

+ * If default behavior of this component does not satisfy the target logic, it is recommended to + * configure an aggregator with individual strategies. + * + * @author Artem Bilan + * + * @since 5.5 + */ +public class FileAggregator implements CorrelationStrategy, ReleaseStrategy, MessageGroupProcessor, BeanFactoryAware { + + private final CorrelationStrategy correlationStrategy = new HeaderAttributeCorrelationStrategy(FileHeaders.FILENAME); + + private final FileMarkerReleaseStrategy releaseStrategy = new FileMarkerReleaseStrategy(); + + private final FileAggregatingMessageGroupProcessor groupProcessor = new FileAggregatingMessageGroupProcessor(); + + @Override + public void setBeanFactory(BeanFactory beanFactory) throws BeansException { + this.groupProcessor.setBeanFactory(beanFactory); + } + + @Override + public Object getCorrelationKey(Message message) { + return this.correlationStrategy.getCorrelationKey(message); + } + + @Override + public boolean canRelease(MessageGroup group) { + return this.releaseStrategy.canRelease(group); + } + + @Override + public Object processMessageGroup(MessageGroup group) { + return this.groupProcessor.processMessageGroup(group); + } + +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/aggregator/FileMarkerReleaseStrategy.java b/spring-integration-file/src/main/java/org/springframework/integration/file/aggregator/FileMarkerReleaseStrategy.java new file mode 100644 index 0000000000..f08f77cdcb --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/aggregator/FileMarkerReleaseStrategy.java @@ -0,0 +1,63 @@ +/* + * Copyright 2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.file.aggregator; + +import java.util.Collection; + +import org.springframework.integration.aggregator.ReleaseStrategy; +import org.springframework.integration.file.FileHeaders; +import org.springframework.integration.file.splitter.FileSplitter; +import org.springframework.integration.store.MessageGroup; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; + +/** + * A {@link ReleaseStrategy} which makes a decision based on the presence of + * {@link org.springframework.integration.file.splitter.FileSplitter.FileMarker.Mark#END} + * message in the group and its {@link org.springframework.integration.file.FileHeaders#LINE_COUNT} header. + * + * @author Artem Bilan + * + * @since 5.5 + */ +public class FileMarkerReleaseStrategy implements ReleaseStrategy { + + @Override + public boolean canRelease(MessageGroup group) { + int size = group.size(); + if (size > 1) { // Need more than only a START marker + Collection> messages = group.getMessages(); + for (Message message : messages) { + if (checkForEndMarker(size, message.getHeaders())) { + return true; + } + } + } + return false; + } + + private boolean checkForEndMarker(int groupSize, MessageHeaders headers) { + if (FileSplitter.FileMarker.Mark.END.name().equals(headers.get(FileHeaders.MARKER))) { + Long lineCount = headers.get(FileHeaders.LINE_COUNT, Long.class); + return lineCount != null && lineCount == groupSize - 2; + } + else { + return false; + } + } + +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/aggregator/package-info.java b/spring-integration-file/src/main/java/org/springframework/integration/file/aggregator/package-info.java new file mode 100644 index 0000000000..d44eb08cb7 --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/aggregator/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides support classes for file-based aggregation logic. + */ +package org.springframework.integration.file.aggregator; diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.java b/spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.java index 6b79875aee..2734412695 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.java @@ -412,8 +412,14 @@ public class FileSplitter extends AbstractMessageSplitter { else { payload = fileMarker; } - return getMessageBuilderFactory().withPayload(payload) - .setHeader(FileHeaders.MARKER, fileMarker.mark.name()); + AbstractIntegrationMessageBuilder messageBuilder = + getMessageBuilderFactory() + .withPayload(payload) + .setHeader(FileHeaders.MARKER, fileMarker.mark.name()); + if (Mark.END.equals(fileMarker.mark)) { + messageBuilder.setHeader(FileHeaders.LINE_COUNT, fileMarker.lineCount); + } + return messageBuilder; } @Override diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/aggregator/FileAggregatorTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/aggregator/FileAggregatorTests.java new file mode 100644 index 0000000000..205b650987 --- /dev/null +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/aggregator/FileAggregatorTests.java @@ -0,0 +1,146 @@ +/* + * Copyright 2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.file.aggregator; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.File; +import java.io.FileOutputStream; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.List; + +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.ImportResource; +import org.springframework.core.task.TaskExecutor; +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.file.FileHeaders; +import org.springframework.integration.file.dsl.Files; +import org.springframework.integration.file.splitter.FileSplitter; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; +import org.springframework.util.FileCopyUtils; + +/** + * @author Artem Bilan + * + * @since 5.5 + */ +@SpringJUnitConfig +@DirtiesContext +public class FileAggregatorTests { + + @TempDir + static File tmpDir; + + static File file; + + @Autowired + @Qualifier("fileSplitterAggregatorFlow.input") + MessageChannel fileSplitterAggregatorFlow; + + @Autowired + PollableChannel resultChannel; + + @Autowired + MessageChannel input; + + @Autowired + PollableChannel output; + + @BeforeAll + static void setup() throws IOException { + file = new File(tmpDir, "foo.txt"); + String content = + "file header\n" + + "first line\n" + + "second line\n" + + "last line"; + FileCopyUtils.copy(content.getBytes(StandardCharsets.UTF_8), new FileOutputStream(file, false)); + } + + @Test + void testFileAggregator() { + this.fileSplitterAggregatorFlow.send(new GenericMessage<>(file)); + + Message receive = this.resultChannel.receive(10_000); + assertThat(receive).isNotNull(); + assertThat(receive.getHeaders()) + .containsEntry(FileHeaders.FILENAME, "foo.txt") + .containsEntry(FileHeaders.LINE_COUNT, 3L) + .containsEntry("firstLine", "file header") + .doesNotContainKey(IntegrationMessageHeaderAccessor.CORRELATION_ID); + + assertThat(receive.getPayload()) + .isInstanceOf(List.class) + .asList() + .contains("SECOND LINE", "LAST LINE", "FIRST LINE"); + } + + @Test + void testFileAggregatorXmlConfig() { + this.input.send(new GenericMessage<>(file)); + + Message receive = this.output.receive(10_000); + assertThat(receive).isNotNull(); + assertThat(receive.getHeaders()) + .containsEntry(FileHeaders.FILENAME, "foo.txt") + .containsEntry(FileHeaders.LINE_COUNT, 4L) + .doesNotContainKeys("firstLine", IntegrationMessageHeaderAccessor.CORRELATION_ID); + + assertThat(receive.getPayload()) + .isInstanceOf(List.class) + .asList() + .containsExactly("file header", "first line", "second line", "last line"); + } + + @Configuration + @EnableIntegration + @ImportResource("org/springframework/integration/file/aggregator/FileAggregatorTests.xml") + public static class Config { + + @Bean + public IntegrationFlow fileSplitterAggregatorFlow(TaskExecutor taskExecutor) { + return f -> f + .split(Files.splitter() + .markers() + .firstLineAsHeader("firstLine")) + .channel(c -> c.executor(taskExecutor)) + .filter(payload -> !(payload instanceof FileSplitter.FileMarker), + e -> e.discardChannel("aggregatorChannel")) + .transform(String::toUpperCase) + .channel("aggregatorChannel") + .aggregate(new FileAggregator()) + .channel(c -> c.queue("resultChannel")); + } + + } + +} diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/aggregator/FileAggregatorTests.xml b/spring-integration-file/src/test/java/org/springframework/integration/file/aggregator/FileAggregatorTests.xml new file mode 100644 index 0000000000..ea331d3747 --- /dev/null +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/aggregator/FileAggregatorTests.xml @@ -0,0 +1,21 @@ + + + + + + + + + + + + + + + diff --git a/spring-integration-file/src/test/kotlin/org/springframework/integration/kotlin/file/aggregator/KotlinFileAggregatorTests.kt b/spring-integration-file/src/test/kotlin/org/springframework/integration/kotlin/file/aggregator/KotlinFileAggregatorTests.kt new file mode 100644 index 0000000000..8878f6624e --- /dev/null +++ b/spring-integration-file/src/test/kotlin/org/springframework/integration/kotlin/file/aggregator/KotlinFileAggregatorTests.kt @@ -0,0 +1,117 @@ +/* + * Copyright 2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kotlin.file.aggregator + +import assertk.all +import assertk.assertThat +import assertk.assertions.* +import org.junit.jupiter.api.BeforeAll +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.io.TempDir +import org.springframework.beans.factory.annotation.Autowired +import org.springframework.beans.factory.annotation.Qualifier +import org.springframework.context.annotation.Bean +import org.springframework.context.annotation.Configuration +import org.springframework.core.task.TaskExecutor +import org.springframework.integration.IntegrationMessageHeaderAccessor +import org.springframework.integration.config.EnableIntegration +import org.springframework.integration.dsl.integrationFlow +import org.springframework.integration.file.FileHeaders +import org.springframework.integration.file.aggregator.FileAggregator +import org.springframework.integration.file.dsl.Files +import org.springframework.integration.file.splitter.FileSplitter.FileMarker +import org.springframework.messaging.MessageChannel +import org.springframework.messaging.PollableChannel +import org.springframework.messaging.support.GenericMessage +import org.springframework.test.annotation.DirtiesContext +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig +import org.springframework.util.FileCopyUtils +import java.io.File +import java.io.FileOutputStream + +/** + * @author Artem Bilan + * + * @since 5.5 + */ +@SpringJUnitConfig +@DirtiesContext +class KotlinFileAggregatorTests { + + companion object { + + lateinit var file: File + + @BeforeAll + @JvmStatic + fun setup(@TempDir tmpDir: File) { + file = File(tmpDir, "foo.txt") + val content = """ + file header + first line + second line + last line + """.trimIndent() + FileCopyUtils.copy(content.toByteArray(), FileOutputStream(file, false)) + } + + } + + @Autowired + @Qualifier("fileSplitterAggregatorFlow.input") + private lateinit var fileSplitterAggregatorFlow: MessageChannel + + @Autowired + private lateinit var resultChannel: PollableChannel + + @Test + fun testFileAggregator() { + this.fileSplitterAggregatorFlow.send(GenericMessage(file)) + val receive = this.resultChannel.receive(10000) + assertThat(receive).isNotNull() + assertThat(receive.headers) + .all { + contains(FileHeaders.FILENAME, "foo.txt") + contains(FileHeaders.LINE_COUNT, 3L) + contains("firstLine", "file header") + doesNotContain(IntegrationMessageHeaderAccessor.CORRELATION_ID, null) + } + + assertThat(receive.payload) + .isInstanceOf(MutableList::class.java) + .containsAll("SECOND LINE", "LAST LINE", "FIRST LINE") + } + + @Configuration + @EnableIntegration + class Config { + + @Bean + fun fileSplitterAggregatorFlow(taskExecutor: TaskExecutor?) = + integrationFlow { + split(Files.splitter().markers().firstLineAsHeader("firstLine")) + channel { executor(taskExecutor) } + filter({ it !is FileMarker }) { discardChannel("aggregatorChannel") } + transform(String::toUpperCase) + channel("aggregatorChannel") + aggregate(FileAggregator()) + channel { queue("resultChannel") } + } + + } + +} \ No newline at end of file diff --git a/src/reference/asciidoc/aggregator.adoc b/src/reference/asciidoc/aggregator.adoc index 776bc33a51..8a64a236f3 100644 --- a/src/reference/asciidoc/aggregator.adoc +++ b/src/reference/asciidoc/aggregator.adoc @@ -969,3 +969,5 @@ Flux> window = .convertSendAndReceive(new Integer[] { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 }, Flux.class); ---- ==== + +See also <<./file.adoc#file-aggregator, File Aggregator>>. \ No newline at end of file diff --git a/src/reference/asciidoc/file.adoc b/src/reference/asciidoc/file.adoc index d59484b2b1..e3ec04e32f 100644 --- a/src/reference/asciidoc/file.adoc +++ b/src/reference/asciidoc/file.adoc @@ -363,7 +363,7 @@ Otherwise, the files from those events are treated the same way. The `ResettableFileListFilter` implementations pick up the `ENTRY_DELETE` events. Consequently, their files are provided for the `remove()` operation. -When this event is enabled, filters such as the `AcceptOnceFileListFilter` have the file removed +When this event is enabled, filters such as the `AcceptOnceFileListFilter` have the file removed. As a result, if a file with the same name appears, it passes the filter and is sent as a message. For this purpose, the `watch-events` property (`FileReadingMessageSource.setWatchEvents(WatchEventType... watchEvents)`) has been introduced. @@ -990,10 +990,77 @@ However, it is only practical for relatively short files. Inbound payloads can be `File`, `String` (a `File` path), `InputStream`, or `Reader`. Other payload types are emitted unchanged. -The following listing shows all the possible attributes for ``: +The following listing shows possible ways to configure a `FileSplitter`: ==== -[source, xml] +[source, java, role="primary"] +.Java DSL +---- +@SpringBootApplication +public class FileSplitterApplication { + + public static void main(String[] args) { + new SpringApplicationBuilder(FileSplitterApplication.class) + .web(false) + .run(args); + } + + @Bean + public IntegrationFlow fileSplitterFlow() { + return IntegrationFlows + .from(Files.inboundAdapter(tmpDir.getRoot()) + .filter(new ChainFileListFilter() + .addFilter(new AcceptOnceFileListFilter<>()) + .addFilter(new ExpressionFileListFilter<>( + new FunctionExpression(f -> "foo.tmp".equals(f.getName())))))) + .split(Files.splitter() + .markers() + .charset(StandardCharsets.US_ASCII) + .firstLineAsHeader("fileHeader") + .applySequence(true)) + .channel(c -> c.queue("fileSplittingResultChannel")) + .get(); + } + +} +---- +[source, kotlin, role="secondary"] +.Kotlin DSL +---- +@Bean +fun fileSplitterFlow() = + integrationFlow( + Files.inboundAdapter(tmpDir.getRoot()) + .filter( + ChainFileListFilter() + .addFilter(AcceptOnceFileListFilter()) + .addFilter(ExpressionFileListFilter(FunctionExpression { f: File? -> "foo.tmp" == f!!.name })) + ) + ) { + split( + Files.splitter() + .markers() + .charset(StandardCharsets.US_ASCII) + .firstLineAsHeader("fileHeader") + .applySequence(true) + ) + channel { queue("fileSplittingResultChannel") } + } +---- +[source, java, role="secondary"] +.Java +---- +@Splitter(inputChannel="toSplitter") +@Bean +public MessageHandler fileSplitter() { + FileSplitter splitter = new FileSplitter(true, true); + splitter.setApplySequence(true); + splitter.setOutputChannel(outputChannel); + return splitter; +} +---- +[source, xml, role="secondary"] +.XML ---- iterator="" <2> @@ -1062,68 +1129,14 @@ When `markersJson` is true, the markers are represented as a JSON string (using Version 5.0 introduced the `firstLineAsHeader` option to specify that the first line of content is a header (such as column names in a CSV file). The argument passed to this property is the header name under which the first line is carried as a header in the messages emitted for the remaining lines. -This line is not included in the sequence header (if `applySequence` is true) nor in the `lineCount` associated with `FileMarker.END` . +This line is not included in the sequence header (if `applySequence` is true) nor in the `lineCount` associated with `FileMarker.END`. +NOTE: Starting with version 5.5, the lineCount` is also included as a `FileHeaders.LINE_COUNT` into headers of the `FileMarker.END` message, since the `FileMarker` could be serialized into JSON. If a file contains only the header line, the file is treated as empty and, therefore, only `FileMarker` instances are emitted during splitting (if markers are enabled -- otherwise, no messages are emitted). By default (if no header name is set), the first line is considered to be data and becomes the payload of the first emitted message. If you need more complex logic about header extraction from the file content (not first line, not the whole content of the line, not one particular header, and so on), consider using <<./content-enrichment.adoc#header-enricher,header enricher>> ahead of the `FileSplitter`. Note that the lines that have been moved to the headers might be filtered downstream from the normal content process. -==== Configuring with Java Configuration - -The following Spring Boot application shows an example of how to configure a file splitter with Java configuration: - -==== -[source, java] ----- -@Splitter(inputChannel="toSplitter") -@Bean -public MessageHandler fileSplitter() { - FileSplitter splitter = new FileSplitter(true, true); - splitter.setApplySequence(true); - splitter.setOutputChannel(outputChannel); - return splitter; -} ----- -==== - -==== Configuring with the Java DSL - -The following Spring Boot application shows an example of how to configure a file splitter with the Java DSL: - -==== -[source, java] ----- -@SpringBootApplication -public class FileSplitterApplication { - - public static void main(String[] args) { - new SpringApplicationBuilder(FileSplitterApplication.class) - .web(false) - .run(args); - } - - @Bean - public IntegrationFlow fileSplitterFlow() { - return IntegrationFlows - .from(Files.inboundAdapter(tmpDir.getRoot()) - .filter(new ChainFileListFilter() - .addFilter(new AcceptOnceFileListFilter<>()) - .addFilter(new ExpressionFileListFilter<>( - new FunctionExpression(f -> "foo.tmp".equals(f.getName())))))) - .split(Files.splitter() - .markers() - .charset(StandardCharsets.US_ASCII) - .firstLineAsHeader("fileHeader") - .applySequence(true)) - .channel(c -> c.queue("fileSplittingResultChannel")) - .get(); - } - -} ----- -==== - [[idempotent-file-splitter]] ==== Idempotent Downstream Processing a Split File @@ -1168,6 +1181,83 @@ public IntegrationFlow flow() { ---- ==== +[[file-aggregator]] +=== File Aggregator + +Starting with version 5.5, a `FileAggregator` is introduced to cover other side of `FileSplitter` use-case when START/END markers are enabled. +For convenience the `FileAggregator` implements all three sequence details strategies: + +- The `HeaderAttributeCorrelationStrategy` with the `FileHeaders.FILENAME` attribute is used for correlation key calculation. +When markers are enabled on the `FileSplitter`, it does not populate sequence details headers, since START/END marker messages are also included into the sequence size. +The `FileHeaders.FILENAME` is still populated for each line emitted, including START/END marker messages. + +- The `FileMarkerReleaseStrategy` - checks for `FileSplitter.FileMarker.Mark.END` message in the group and then compare a `FileHeaders.LINE_COUNT` header value with the group size minus `2` - `FileSplitter.FileMarker` instances. + +- The `FileAggregatingMessageGroupProcessor` just removes `FileSplitter.FileMarker` messages from the group and collect the rest of messages into a list payload to produce. + +The following listing shows possible ways to configure a `FileAggregator`: + +==== +[source, java, role="primary"] +.Java DSL +---- +@Bean +public IntegrationFlow fileSplitterAggregatorFlow(TaskExecutor taskExecutor) { + return f -> f + .split(Files.splitter() + .markers() + .firstLineAsHeader("firstLine")) + .channel(c -> c.executor(taskExecutor)) + .filter(payload -> !(payload instanceof FileSplitter.FileMarker), + e -> e.discardChannel("aggregatorChannel")) + .transform(String::toUpperCase) + .channel("aggregatorChannel") + .aggregate(new FileAggregator()) + .channel(c -> c.queue("resultChannel")); +} +---- +[source, kotlin, role="secondary"] +.Kotlin DSL +---- +@Bean +fun fileSplitterAggregatorFlow(taskExecutor: TaskExecutor?) = + integrationFlow { + split(Files.splitter().markers().firstLineAsHeader("firstLine")) + channel { executor(taskExecutor) } + filter({ it !is FileMarker }) { discardChannel("aggregatorChannel") } + transform(String::toUpperCase) + channel("aggregatorChannel") + aggregate(FileAggregator()) + channel { queue("resultChannel") } + } +---- +[source, java, role="secondary"] +.Java +---- +@serviceActivator(inputChannel="toAggregateFile") +@Bean +public AggregatorFactoryBean fileAggregator() { + AggregatorFactoryBean aggregator = new AggregatorFactoryBean(); + aggregator.setProcessorBean(new FileAggregator()); + aggregator.setOutputChannel(outputChannel); + return aggregator; +} +---- +[source, xml, role="secondary"] +.XML +---- + + + + + + +---- +==== + +If default behavior of the `FileAggregator` does not satisfy the target logic, it is recommended to configure an aggregator endpoint with individual strategies. +See `FileAggregator` JavaDocs for more information. + [[remote-persistent-flf]] === Remote Persistent File List Filters diff --git a/src/reference/asciidoc/splitter.adoc b/src/reference/asciidoc/splitter.adoc index 7ab7aeddaf..3fdcd0ec9f 100644 --- a/src/reference/asciidoc/splitter.adoc +++ b/src/reference/asciidoc/splitter.adoc @@ -142,6 +142,4 @@ List extractItems(Order order) { ---- ==== -See also <<./handler-advice.adoc#advising-with-annotations,Advising Endpoints Using Annotations>>. - -See also <<./dsl.adoc#java-dsl-splitters,Splitters>> in the Java DSL chapter. +See also <<./handler-advice.adoc#advising-with-annotations,Advising Endpoints Using Annotations>>, <<./dsl.adoc#java-dsl-splitters,Splitters>> and <<./file.adoc#file-splitter, File Splitter>>. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 2209a78249..9870817db4 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -15,6 +15,12 @@ If you are interested in more details, see the Issue Tracker tickets that were r [[x5.5-new-components]] === New Components +[[x5.5-file-aggregator]] +==== File Aggregator + +A `FileSplitter.FileMaker`-based implementation of `CorrelationStrategy`, `ReleaseStrategy` and `MessageGroupProcessor` as a `FileAggregator` component was introduced. +See <<./file.adoc#file-aggregator, File Aggregator>> for more information. + [[x5.5-general]] === General Changes