diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IdempotentReceiverInterceptorParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IdempotentReceiverInterceptorParser.java index 51e4f9b161..8798d71ef9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IdempotentReceiverInterceptorParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IdempotentReceiverInterceptorParser.java @@ -59,6 +59,8 @@ public class IdempotentReceiverInterceptorParser extends AbstractBeanDefinitionP boolean hasValueStrategy = StringUtils.hasText(valueStrategy); String valueExpression = element.getAttribute("value-expression"); boolean hasValueExpression = StringUtils.hasText(valueExpression); + String compareValues = element.getAttribute("compare-values"); + boolean hasCompareValues = StringUtils.hasText(compareValues); String endpoints = element.getAttribute("endpoint"); @@ -68,10 +70,10 @@ public class IdempotentReceiverInterceptorParser extends AbstractBeanDefinitionP } if (hasSelector && (hasStore || hasKeyStrategy || hasKeyExpression || hasValueStrategy // NOSONAR complexity - || hasValueExpression)) { + || hasValueExpression || hasCompareValues)) { parserContext.getReaderContext().error("The 'selector' attribute is mutually exclusive with " + - "'metadata-store', 'key-strategy', 'key-expression', 'value-strategy' " + - "or 'value-expression'", source); + "'metadata-store', 'key-strategy', 'key-expression', 'value-strategy', " + + "'value-expression', and 'compare-values'", source); } if (hasKeyStrategy && hasKeyExpression) { @@ -133,6 +135,7 @@ public class IdempotentReceiverInterceptorParser extends AbstractBeanDefinitionP else { selectorBuilder.addConstructorArgValue(new RootBeanDefinition(SimpleMetadataStore.class)); } + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(selectorBuilder, element, "compare-values"); selectorBeanDefinition = selectorBuilder.getBeanDefinition(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/selector/MetadataStoreSelector.java b/spring-integration-core/src/main/java/org/springframework/integration/selector/MetadataStoreSelector.java index 1e2e57b0b8..eac404647d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/selector/MetadataStoreSelector.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/selector/MetadataStoreSelector.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2019 the original author or authors. + * Copyright 2014-2020 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. @@ -16,10 +16,13 @@ package org.springframework.integration.selector; +import java.util.function.BiPredicate; + import org.springframework.integration.core.MessageSelector; import org.springframework.integration.handler.MessageProcessor; import org.springframework.integration.metadata.ConcurrentMetadataStore; import org.springframework.integration.metadata.SimpleMetadataStore; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.util.Assert; @@ -58,6 +61,9 @@ public class MetadataStoreSelector implements MessageSelector { private final MessageProcessor valueStrategy; + @Nullable + private BiPredicate compareValues; + public MetadataStoreSelector(MessageProcessor keyStrategy) { this(keyStrategy, (MessageProcessor) null); } @@ -79,6 +85,26 @@ public class MetadataStoreSelector implements MessageSelector { this.valueStrategy = valueStrategy; } + /** + * Set a {@link BiPredicate} to compare old and new values in the metadata store for + * the key. The first parameter is the old value; return true if we should accept this + * message and replace the old value with the new value. + * @param compareValues the {@link BiPredicate}. + * @since 5.3 + */ + public void setCompareValues(@Nullable BiPredicate compareValues) { + this.compareValues = compareValues; + } + + /** + * Fluent version of {@link #setCompareValues(BiPredicate)}. + * @param compareValues the {@link BiPredicate}. + * @return this. + */ + public MetadataStoreSelector compareValues(@Nullable BiPredicate compareValues) { + setCompareValues(compareValues); + return this; + } @Override public boolean accept(Message message) { @@ -88,7 +114,21 @@ public class MetadataStoreSelector implements MessageSelector { ? this.valueStrategy.processMessage(message) : (timestamp == null ? "0" : Long.toString(timestamp)); - return this.metadataStore.putIfAbsent(key, value) == null; + if (this.compareValues == null) { + return this.metadataStore.putIfAbsent(key, value) == null; + } + else { + synchronized (this) { + String oldValue = this.metadataStore.get(key); + if (oldValue == null) { + return this.metadataStore.putIfAbsent(key, value) == null; + } + if (this.compareValues.test(oldValue, value)) { + return this.metadataStore.replace(key, oldValue, value); + } + return false; + } + } } } diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd index c5243b4dd0..807bf72e22 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd @@ -4853,6 +4853,19 @@ The list of component name patterns you want to track (e.g., tracked-components ]]> + + + + + + + + ' which is called if a value exists to determine whether the message + should be accepted and the old value replaced with the new value in the metadata store. + ]]> + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/IdempotentReceiverParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/IdempotentReceiverParserTests-context.xml index 26d2155685..380d051a6a 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/IdempotentReceiverParserTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/IdempotentReceiverParserTests-context.xml @@ -27,9 +27,13 @@ endpoint="foo" key-strategy="keyStrategy" value-strategy="valueStrategy" + compare-values="valueComparator" discard-channel="nullChannel" throw-exception-on-rejection="true"/> + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/IdempotentReceiverParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/IdempotentReceiverParserTests.java index 74ac3644cd..08eed3ada9 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/IdempotentReceiverParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/IdempotentReceiverParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2019 the original author or authors. + * Copyright 2014-2020 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. @@ -24,6 +24,7 @@ import java.io.ByteArrayInputStream; import java.util.List; import java.util.Map; import java.util.Properties; +import java.util.function.BiPredicate; import org.junit.Test; import org.junit.runner.RunWith; @@ -77,6 +78,9 @@ public class IdempotentReceiverParserTests { @Autowired private MessageProcessor valueStrategy; + @Autowired + private AlwaysAccept alwaysAccept; + @Autowired @Qualifier("nullChannel") private MessageChannel nullChannel; @@ -94,8 +98,8 @@ public class IdempotentReceiverParserTests { assertThat(getPropertyValue(this.selectorInterceptor, "throwExceptionOnRejection", Boolean.class)).isFalse(); @SuppressWarnings("unchecked") Map> idempotentEndpoints = - (Map>) getPropertyValue(this.idempotentReceiverAutoProxyCreator, - "idempotentEndpoints", Map.class); + getPropertyValue(this.idempotentReceiverAutoProxyCreator, + "idempotentEndpoints", Map.class); List endpoints = idempotentEndpoints.get("selectorInterceptor"); assertThat(endpoints).isNotNull(); assertThat(endpoints.isEmpty()).isFalse(); @@ -110,10 +114,11 @@ public class IdempotentReceiverParserTests { assertThat(messageSelector).isInstanceOf(MetadataStoreSelector.class); assertThat(getPropertyValue(messageSelector, "keyStrategy")).isSameAs(this.keyStrategy); assertThat(getPropertyValue(messageSelector, "valueStrategy")).isSameAs(this.valueStrategy); + assertThat(getPropertyValue(messageSelector, "compareValues")).isSameAs(this.alwaysAccept); @SuppressWarnings("unchecked") Map> idempotentEndpoints = - (Map>) getPropertyValue(this.idempotentReceiverAutoProxyCreator, - "idempotentEndpoints", Map.class); + getPropertyValue(this.idempotentReceiverAutoProxyCreator, + "idempotentEndpoints", Map.class); List endpoints = idempotentEndpoints.get("strategyInterceptor"); assertThat(endpoints).isNotNull(); assertThat(endpoints.isEmpty()).isFalse(); @@ -130,8 +135,8 @@ public class IdempotentReceiverParserTests { assertThat(keyStrategy.toString()).contains("headers.foo"); @SuppressWarnings("unchecked") Map> idempotentEndpoints = - (Map>) getPropertyValue(this.idempotentReceiverAutoProxyCreator, - "idempotentEndpoints", Map.class); + getPropertyValue(this.idempotentReceiverAutoProxyCreator, + "idempotentEndpoints", Map.class); List endpoints = idempotentEndpoints.get("expressionInterceptor"); assertThat(endpoints).isNotNull(); assertThat(endpoints.isEmpty()).isFalse(); @@ -172,7 +177,8 @@ public class IdempotentReceiverParserTests { catch (BeanDefinitionParsingException e) { assertThat(e.getMessage()) .contains("The 'selector' attribute is mutually exclusive with 'metadata-store', " + - "'key-strategy', 'key-expression', 'value-strategy' or 'value-expression'"); + "'key-strategy', 'key-expression', 'value-strategy', 'value-expression', and " + + "'compare-values'"); } } @@ -185,7 +191,8 @@ public class IdempotentReceiverParserTests { catch (BeanDefinitionParsingException e) { assertThat(e.getMessage()) .contains("The 'selector' attribute is mutually exclusive with 'metadata-store', " + - "'key-strategy', 'key-expression', 'value-strategy' or 'value-expression'"); + "'key-strategy', 'key-expression', 'value-strategy', 'value-expression', and " + + "'compare-values'"); } } @@ -198,7 +205,8 @@ public class IdempotentReceiverParserTests { catch (BeanDefinitionParsingException e) { assertThat(e.getMessage()) .contains("The 'selector' attribute is mutually exclusive with 'metadata-store', " + - "'key-strategy', 'key-expression', 'value-strategy' or 'value-expression'"); + "'key-strategy', 'key-expression', 'value-strategy', 'value-expression', and " + + "'compare-values'"); } } @@ -211,7 +219,8 @@ public class IdempotentReceiverParserTests { catch (BeanDefinitionParsingException e) { assertThat(e.getMessage()) .contains("The 'selector' attribute is mutually exclusive with 'metadata-store', " + - "'key-strategy', 'key-expression', 'value-strategy' or 'value-expression'"); + "'key-strategy', 'key-expression', 'value-strategy', 'value-expression', and " + + "'compare-values'"); } } @@ -224,7 +233,8 @@ public class IdempotentReceiverParserTests { catch (BeanDefinitionParsingException e) { assertThat(e.getMessage()) .contains("The 'selector' attribute is mutually exclusive with 'metadata-store', " + - "'key-strategy', 'key-expression', 'value-strategy' or 'value-expression'"); + "'key-strategy', 'key-expression', 'value-strategy', 'value-expression', and " + + "'compare-values'"); } } @@ -268,4 +278,13 @@ public class IdempotentReceiverParserTests { return ac; } + public static class AlwaysAccept implements BiPredicate { + + @Override + public boolean test(String t, String u) { + return true; + } + + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/selector/MetadataStoreSelectorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/selector/MetadataStoreSelectorTests.java new file mode 100644 index 0000000000..bc659cf6c7 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/selector/MetadataStoreSelectorTests.java @@ -0,0 +1,50 @@ +/* + * Copyright 2020 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.selector; + +import static org.assertj.core.api.Assertions.assertThat; + +import org.junit.jupiter.api.Test; + +import org.springframework.integration.metadata.SimpleMetadataStore; +import org.springframework.messaging.support.MessageBuilder; + +/** + * @author Gary Russell + * @since 5.3 + * + */ +public class MetadataStoreSelectorTests { + + @Test + void lineNumbers() { + SimpleMetadataStore store = new SimpleMetadataStore(); + store.put("file", "5"); + MetadataStoreSelector selector = new MetadataStoreSelector( + msg -> "file", msg -> msg.getHeaders().get("lineNum").toString(), store); + selector.setCompareValues((oldValue, newValue) -> Integer.parseInt(oldValue) < Integer.parseInt(newValue)); + for (int i = 0; i < 6; i++) { + assertThat(selector.accept(MessageBuilder.withPayload("foo").setHeader("lineNum", i).build())) + .isEqualTo(Boolean.FALSE); + assertThat(store.get("file")).isEqualTo("5"); + } + assertThat(selector.accept(MessageBuilder.withPayload("foo").setHeader("lineNum", 6).build())) + .isEqualTo(Boolean.TRUE); + assertThat(store.get("file")).isEqualTo("6"); + } + +} diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests.java index cdceabfa43..744609a1ba 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2019 the original author or authors. + * Copyright 2015-2020 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. @@ -49,6 +49,10 @@ import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.file.FileHeaders; import org.springframework.integration.file.splitter.FileSplitter.FileMarker; +import org.springframework.integration.handler.advice.IdempotentReceiverInterceptor; +import org.springframework.integration.metadata.ConcurrentMetadataStore; +import org.springframework.integration.metadata.SimpleMetadataStore; +import org.springframework.integration.selector.MetadataStoreSelector; import org.springframework.integration.support.json.JsonObjectMapper; import org.springframework.integration.support.json.JsonObjectMapperProvider; import org.springframework.messaging.Message; @@ -90,6 +94,12 @@ public class FileSplitterTests { @Autowired private PollableChannel output; + @Autowired + private MetadataStoreSelector selector; + + @Autowired + private ConcurrentMetadataStore store; + @BeforeAll static void setup(@TempDir File tempDir) throws IOException { file = new File(tempDir, "foo.txt"); @@ -104,12 +114,19 @@ public class FileSplitterTests { assertThat(receive).isNotNull(); //HelloWorld assertThat(receive.getPayload()).isEqualTo("HelloWorld"); assertThat(receive.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)).isEqualTo(2); + assertThat(receive.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER)).isEqualTo(1); + assertThat(this.selector.accept(receive)).isTrue(); + assertThat(this.store.get(this.file.getAbsolutePath())).isEqualTo("1"); receive = this.output.receive(10000); assertThat(receive).isNotNull(); //äöüß assertThat(receive.getPayload()).isEqualTo("äöüß"); assertThat(receive.getHeaders().get(FileHeaders.ORIGINAL_FILE)).isEqualTo(file); assertThat(receive.getHeaders().get(FileHeaders.FILENAME)).isEqualTo(file.getName()); + assertThat(receive.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER)).isEqualTo(2); assertThat(this.output.receive(1)).isNull(); + assertThat(this.selector.accept(receive)).isTrue(); + assertThat(this.store.get(this.file.getAbsolutePath())).isEqualTo("2"); + assertThat(this.selector.accept(receive)).isFalse(); this.input1.send(new GenericMessage<>(file.getAbsolutePath())); receive = this.output.receive(10000); @@ -396,6 +413,28 @@ public class FileSplitterTests { return fileSplitter; } + @Bean + public IdempotentReceiverInterceptor idempotentReceiverInterceptor() { + return new IdempotentReceiverInterceptor(selector()); + } + + @Bean + public MetadataStoreSelector selector() { + return new MetadataStoreSelector( + message -> message.getHeaders().get(FileHeaders.ORIGINAL_FILE, File.class) + .getAbsolutePath(), + message -> message.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER) + .toString(), + store()) + .compareValues( + (oldVal, newVal) -> Integer.parseInt(oldVal) < Integer.parseInt(newVal)); + } + + @Bean + public ConcurrentMetadataStore store() { + return new SimpleMetadataStore(); + } + } } diff --git a/src/reference/asciidoc/file.adoc b/src/reference/asciidoc/file.adoc index e0877b4e2f..e9ba68c9d3 100644 --- a/src/reference/asciidoc/file.adoc +++ b/src/reference/asciidoc/file.adoc @@ -1116,6 +1116,50 @@ public class FileSplitterApplication { ---- ==== +[[idempotent-file-splitter]] +==== Idempotent Downstream Processing a Split File + +When `apply-sequence` is true, the splitter adds the line number in the `SEQUENCE_NUMBER` header (when `markers` is true, the markers are counted as lines). +The line number can be used with an <<./handler-advice.adoc#idempotent-receiver,Idempotent Receiver>> to avoid reprocessing lines after a restart. + +For example: + +==== +[source, java] +---- +@Bean +public ConcurrentMetadataStore store() { + return new ZookeeperMetadataStore(); +} + +@Bean +public MetadataStoreSelector selector() { + return new MetadataStoreSelector( + message -> message.getHeaders().get(FileHeaders.ORIGINAL_FILE, File.class) + .getAbsolutePath(), + message -> message.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER) + .toString(), + store()) + .compareValues( + (oldVal, newVal) -> Integer.parseInt(oldVal) < Integer.parseInt(newVal)); +} + +@Bean +public IdempotentReceiverInterceptor idempotentReceiverInterceptor() { + return new IdempotentReceiverInterceptor(selector()); +} + +@Bean +public IntegrationFlow flow() { + ... + .split(new FileSplitter()) + ... + .handle("lineHandler", e -> e.advice(idempotentReceiverInterceptor())) + ... +} +---- +==== + [[remote-persistent-flf]] === Remote Persistent File List Filters diff --git a/src/reference/asciidoc/handler-advice.adoc b/src/reference/asciidoc/handler-advice.adoc index db87754512..04db1de2c4 100644 --- a/src/reference/asciidoc/handler-advice.adoc +++ b/src/reference/asciidoc/handler-advice.adoc @@ -872,6 +872,13 @@ See the https://docs.spring.io/spring-integration/api/org/springframework/integr You can also customize the `value` for `ConcurrentMetadataStore` by using an additional `MessageProcessor`. By default, `MetadataStoreSelector` uses the `timestamp` message header. +Normally, the selector selects a message for acceptance if there is no existing value for the key. +In some cases, it is useful to compare the current and new values for a key, to determine whether the message should be accepted. +Starting with version 5.3, the `compareValues` property is provided which references a `BiPredicate`; the first parameter is the old value; return `true` to accept the message and replace the old value with the new value in the `MetadataStore`. +This can be useful to reduce the number of keys; for example, when processing lines in a file, you can store the file name in the key and the current line number in the value. +Then, after a restart, you can skip lines that have already been processed. +See <<./file.adoc#idempotent-file-splitter,Idempotent Downstream Processing a Split File>> for an example. + For convenience, the `MetadataStoreSelector` options are configurable directly on the `` component. The following listing shows all the possible attributes: @@ -888,7 +895,8 @@ The following listing shows all the possible attributes: key-expression="" <7> value-strategy="" <8> value-expression="" <9> - throw-exception-on-rejection="" /> <10> + compare-values="" <10> + throw-exception-on-rejection="" /> <11> ---- <1> The ID of the `IdempotentReceiverInterceptor` bean. @@ -928,7 +936,8 @@ Used by the underlying `MetadataStoreSelector`. Evaluates a `value` for the `idempotentKey` by using the request message as the evaluation context root object. Mutually exclusive with `selector` and `value-strategy`. By default, the 'MetadataStoreSelector' uses the 'timestamp' message header as the metadata 'value'. -<10> Whether to throw an exception if the `IdempotentReceiverInterceptor` rejects the message. +<10> A reference to a `BiPredicate` bean which allows you to optionally select a message by comparing the old and new values for the key; `null` by default. +<11> Whether to throw an exception if the `IdempotentReceiverInterceptor` rejects the message. Defaults to `false`. It is applied regardless of whether or not a `discard-channel` is provided. ==== diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 0403e500fa..cfb5ee9921 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -81,6 +81,9 @@ See also <<./transactions.adoc#reactive-transactions,Reactive Transactions>>. A new `intercept()` operator to register `ChannelInterceptor` instances without creating explicit channels was added into Java DSL. See <<./dsl.adoc#java-dsl-intercept,Operator intercept()>> for more information. +The `MessageStoreSelector` has a new mechanism to compare an old and new value. +See <<./handler-advice.adoc#idempotent-receiver,Idempotent Receiver Enterprise Integration Pattern>> for more information. + [[x5.3-amqp]] === AMQP Changes