Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-3241: MetadataStoreSelector - compare old/new (#3242)

Browse Source 
Resolves: https://github.com/spring-projects/spring-integration/issues/3241

 * Docs and XML namespace support.
This commit is contained in:
Gary Russell
2020-04-06 18:12:36 -04:00
committed by GitHub
parent 1f2a33437d
commit 7d7c273515
10 changed files with 244 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
spring-integration-core/src/main/java/org/springframework/integration/config/xml/IdempotentReceiverInterceptorParser.java
View File

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

44
spring-integration-core/src/main/java/org/springframework/integration/selector/MetadataStoreSelector.java
View File

@@ -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<String> valueStrategy;
@Nullable
private BiPredicate<String, String> compareValues;
public MetadataStoreSelector(MessageProcessor<String> keyStrategy) {
this(keyStrategy, (MessageProcessor<String>) 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<String, String> compareValues) {
this.compareValues = compareValues;
}
/**
* Fluent version of {@link #setCompareValues(BiPredicate)}.
* @param compareValues the {@link BiPredicate}.
* @return this.
*/
public MetadataStoreSelector compareValues(@Nullable BiPredicate<String, String> 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;
}
}
}
}

13
spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd
View File

@@ -4853,6 +4853,19 @@ The list of component name patterns you want to track (e.g., tracked-components
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="compare-values">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="java.util.function.BiPredicate"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation><![CDATA[
A 'BiPredicate<String, String>' 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.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="discard-channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>

4
spring-integration-core/src/test/java/org/springframework/integration/config/xml/IdempotentReceiverParserTests-context.xml
View File

@@ -27,9 +27,13 @@
endpoint="foo"
key-strategy="keyStrategy"
value-strategy="valueStrategy"
compare-values="valueComparator"
discard-channel="nullChannel"
throw-exception-on-rejection="true"/>
<beans:bean id="valueComparator"
class="org.springframework.integration.config.xml.IdempotentReceiverParserTests$AlwaysAccept"/>
<beans:bean id="store" class="org.springframework.integration.metadata.SimpleMetadataStore"/>
<util:properties id="properties">

43
spring-integration-core/src/test/java/org/springframework/integration/config/xml/IdempotentReceiverParserTests.java
View File

@@ -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<String> 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<String, List<String>> idempotentEndpoints =
(Map<String, List<String>>) getPropertyValue(this.idempotentReceiverAutoProxyCreator,
"idempotentEndpoints", Map.class);
getPropertyValue(this.idempotentReceiverAutoProxyCreator,
"idempotentEndpoints", Map.class);
List<String> 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<String, List<String>> idempotentEndpoints =
(Map<String, List<String>>) getPropertyValue(this.idempotentReceiverAutoProxyCreator,
"idempotentEndpoints", Map.class);
getPropertyValue(this.idempotentReceiverAutoProxyCreator,
"idempotentEndpoints", Map.class);
List<String> 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<String, List<String>> idempotentEndpoints =
(Map<String, List<String>>) getPropertyValue(this.idempotentReceiverAutoProxyCreator,
"idempotentEndpoints", Map.class);
getPropertyValue(this.idempotentReceiverAutoProxyCreator,
"idempotentEndpoints", Map.class);
List<String> 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<String, String> {
@Override
public boolean test(String t, String u) {
return true;
}
}
}

50
spring-integration-core/src/test/java/org/springframework/integration/selector/MetadataStoreSelectorTests.java Normal file
View File

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

41
spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests.java
View File

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

44
src/reference/asciidoc/file.adoc
View File

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

13
src/reference/asciidoc/handler-advice.adoc
View File

@@ -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<String, String>`; 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 `<idempotent-receiver>` 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<String, String>` 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.
====

3
src/reference/asciidoc/whats-new.adoc
View File

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