From 818be4cbe80688639ebcac1c06e155d67d0e8829 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 8 Apr 2019 16:42:57 -0400 Subject: [PATCH] GH-2765: Add discardChannel for splitter (#2883) * GH-2765: Add discardChannel for splitter Fixes https://github.com/spring-projects/spring-integration/issues/2765 When encountering empty collections, splitter should be able to send the result to a discard channel. Currently, when encountering an empty collection, the splitter ends the flow. Some use-cases may rely on a custom split function which may returns empty collections. These use-cases should be able to define a discard channel so they can proceed with a possible compensation flow. * Add `discardChannel` option to the `AbstractMessageSplitter` * Delegate `discardChannel` population from everywhere it is possible: DSL, XML, `AbstractMessageSplitter` extension like `FileSplitter` etc. * Fix `FileSplitterTests` for broken charset * Document new feature; fix some typos and out-dated code sample * * Fix `SplitterFactoryBean` for NPE on the `discardChannelName` propagation * * Check `this.discardChannel` first * `Assert.state()` in `doInit()` for mutual exclusiveness --- .../config/SplitterFactoryBean.java | 19 +++++ .../config/xml/SplitterParser.java | 2 + .../integration/dsl/SplitterEndpointSpec.java | 44 ++++++++++- .../splitter/AbstractMessageSplitter.java | 73 ++++++++++++++++++- .../config/spring-integration-5.2.xsd | 13 ++++ .../correlation/CorrelationHandlerTests.java | 43 +++++++---- .../file/config/FileSplitterParser.java | 1 + .../config/spring-integration-file-5.2.xsd | 13 ++++ .../FileSplitterParserTests-context.xml | 3 +- .../file/config/FileSplitterParserTests.java | 1 + .../file/splitter/FileSplitterTests.java | 47 ++++++------ .../config/XPathMessageSplitterParser.java | 7 +- .../xml/config/spring-integration-xml-5.2.xsd | 13 ++++ .../XPathSplitterParserTests-context.xml | 3 +- .../xml/config/XPathSplitterParserTests.java | 14 ++-- src/reference/asciidoc/dsl.adoc | 8 +- src/reference/asciidoc/splitter.adoc | 17 +++-- src/reference/asciidoc/whats-new.adoc | 3 + 18 files changed, 263 insertions(+), 61 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/SplitterFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/SplitterFactoryBean.java index 705ebff883..ec1c4b400d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/SplitterFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/SplitterFactoryBean.java @@ -23,6 +23,7 @@ import org.springframework.integration.splitter.AbstractMessageSplitter; import org.springframework.integration.splitter.DefaultMessageSplitter; import org.springframework.integration.splitter.ExpressionEvaluatingSplitter; import org.springframework.integration.splitter.MethodInvokingSplitter; +import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -42,6 +43,10 @@ public class SplitterFactoryBean extends AbstractStandardMessageHandlerFactoryBe private String delimiters; + private MessageChannel discardChannel; + + private String discardChannelName; + public void setApplySequence(boolean applySequence) { this.applySequence = applySequence; } @@ -50,6 +55,14 @@ public class SplitterFactoryBean extends AbstractStandardMessageHandlerFactoryBe this.delimiters = delimiters; } + public void setDiscardChannel(MessageChannel discardChannel) { + this.discardChannel = discardChannel; + } + + public void setDiscardChannelName(String discardChannelName) { + this.discardChannelName = discardChannelName; + } + @Override protected MessageHandler createMethodInvokingHandler(Object targetObject, String targetMethodName) { Assert.notNull(targetObject, "targetObject must not be null"); @@ -90,6 +103,12 @@ public class SplitterFactoryBean extends AbstractStandardMessageHandlerFactoryBe protected AbstractMessageSplitter configureSplitter(AbstractMessageSplitter splitter) { postProcessReplyProducer(splitter); + if (this.discardChannel != null) { + splitter.setDiscardChannel(this.discardChannel); + } + else if (StringUtils.hasText(this.discardChannelName)) { + splitter.setDiscardChannelName(this.discardChannelName); + } return splitter; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/SplitterParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/SplitterParser.java index 56c95b7c98..fd583f1e7e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/SplitterParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/SplitterParser.java @@ -28,6 +28,7 @@ import org.springframework.integration.config.SplitterFactoryBean; * @author Mark Fisher * @author Iwein Fuld * @author Gary Russell + * @author Artem Bilan */ public class SplitterParser extends AbstractDelegatingConsumerEndpointParser { @@ -45,6 +46,7 @@ public class SplitterParser extends AbstractDelegatingConsumerEndpointParser { void postProcess(BeanDefinitionBuilder builder, Element element, ParserContext parserContext) { IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "apply-sequence"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "delimiters"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "discard-channel", "discardChannelName"); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/SplitterEndpointSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/SplitterEndpointSpec.java index adac446014..6191bcb6ba 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/SplitterEndpointSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/SplitterEndpointSpec.java @@ -18,6 +18,7 @@ package org.springframework.integration.dsl; import org.springframework.integration.splitter.AbstractMessageSplitter; import org.springframework.integration.splitter.DefaultMessageSplitter; +import org.springframework.messaging.MessageChannel; /** * A {@link ConsumerEndpointSpec} for a {@link AbstractMessageSplitter} implementations. @@ -43,7 +44,7 @@ public final class SplitterEndpointSpec */ public SplitterEndpointSpec applySequence(boolean applySequence) { this.handler.setApplySequence(applySequence); - return _this(); + return this; } /** @@ -65,4 +66,45 @@ public final class SplitterEndpointSpec return this; } + /** + * Specify a channel where rejected Messages should be sent. If the discard + * channel is null (the default), rejected Messages will be dropped. + * A "Rejected Message" means that split function has returned an empty result (but not null): + * no items to iterate for sending. + * @param discardChannel The discard channel. + * @return the endpoint spec. + * @since 5.2 + * @see DefaultMessageSplitter#setDelimiters(String) + */ + public SplitterEndpointSpec discardChannel(MessageChannel discardChannel) { + this.handler.setDiscardChannel(discardChannel); + return this; + } + + /** + * Specify a channel bean name where rejected Messages should be sent. If the discard + * channel is null (the default), rejected Messages will be dropped. + * A "Rejected Message" means that split function has returned an empty result (but not null): + * no items to iterate for sending. + * @param discardChannelName The discard channel bean name. + * @return the endpoint spec. + * @since 5.2 + * @see DefaultMessageSplitter#setDelimiters(String) + */ + public SplitterEndpointSpec discardChannel(String discardChannelName) { + this.handler.setDiscardChannelName(discardChannelName); + return this; + } + + /** + * Configure a subflow to run for discarded messages instead of a + * {@link #discardChannel(MessageChannel)}. + * @param discardFlow the discard flow. + * @return the endpoint spec. + * @since 5.2 + */ + public SplitterEndpointSpec discardFlow(IntegrationFlow discardFlow) { + return discardChannel(obtainInputChannelFromFlow(discardFlow)); + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java b/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java index ec35e75579..06b2a6e341 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java @@ -30,13 +30,17 @@ import org.reactivestreams.Publisher; import org.springframework.integration.channel.ReactiveStreamsSubscribableChannel; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.integration.handler.DiscardingMessageHandler; import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.integration.support.json.JacksonPresent; import org.springframework.integration.util.FunctionIterator; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.util.Assert; import com.fasterxml.jackson.core.TreeNode; import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; /** * Base class for Message-splitting handlers. @@ -47,10 +51,15 @@ import reactor.core.publisher.Flux; * @author Ruslan Stelmachenko * @author Gary Russell */ -public abstract class AbstractMessageSplitter extends AbstractReplyProducingMessageHandler { +public abstract class AbstractMessageSplitter extends AbstractReplyProducingMessageHandler + implements DiscardingMessageHandler { private boolean applySequence = true; + private MessageChannel discardChannel; + + private String discardChannelName; + /** * Set the applySequence flag to the specified value. Defaults to true. * @param applySequence true to apply sequence information. @@ -59,10 +68,54 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess this.applySequence = applySequence; } + /** + * Specify a channel where rejected Messages should be sent. If the discard + * channel is null (the default), rejected Messages will be dropped. + * A "Rejected Message" means that split function has returned an empty result (but not null): + * no items to iterate for sending. + * @param discardChannel The discard channel. + * @since 5.2 + */ + public void setDiscardChannel(MessageChannel discardChannel) { + this.discardChannel = discardChannel; + } + + /** + * Specify a channel bean name (resolved to {@link MessageChannel} lazily) + * where rejected Messages should be sent. If the discard + * channel is null (the default), rejected Messages will be dropped. + * A "Rejected Message" means that split function has returned an empty result (but not null): + * no items to iterate for sending. + * @param discardChannelName The discard channel bean name. + * @since 5.2 + */ + public void setDiscardChannelName(String discardChannelName) { + Assert.hasText(discardChannelName, "'discardChannelName' must not be empty"); + this.discardChannelName = discardChannelName; + } + + @Override + public MessageChannel getDiscardChannel() { + if (this.discardChannel == null) { + String channelName = this.discardChannelName; + if (channelName != null) { + this.discardChannel = getChannelResolver().resolveDestination(channelName); + this.discardChannelName = null; + } + } + return this.discardChannel; + } + + @Override + protected void doInit() { + Assert.state(!(this.discardChannelName != null && this.discardChannel != null), + "'discardChannelName' and 'discardChannel' are mutually exclusive."); + } + @Override @SuppressWarnings("unchecked") protected final Object handleRequestMessage(Message message) { - Object result = this.splitMessage(message); + Object result = splitMessage(message); // return null if 'null' if (result == null) { return null; @@ -137,6 +190,10 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess } if (iterator != null && !iterator.hasNext()) { + MessageChannel discardingChannel = getDiscardChannel(); + if (discardingChannel != null) { + this.messagingTemplate.send(discardingChannel, message); + } return null; } @@ -154,7 +211,16 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess object -> createBuilder(object, headers, correlationId, sequenceNumber.getAndIncrement(), sequenceSize); if (reactive) { - return flux.map(messageBuilderFunction); + return flux + .map(messageBuilderFunction) + .switchIfEmpty( + Mono.defer(() -> { + MessageChannel discardingChannel = getDiscardChannel(); + if (discardingChannel != null) { + this.messagingTemplate.send(discardingChannel, message); + } + return Mono.empty(); + })); } else { return new FunctionIterator<>(result instanceof AutoCloseable && !result.equals(iterator) @@ -195,6 +261,7 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess private AbstractIntegrationMessageBuilder createBuilder(Object item, Map headers, Object correlationId, int sequenceNumber, int sequenceSize) { + AbstractIntegrationMessageBuilder builder; if (item instanceof Message) { builder = getMessageBuilderFactory().fromMessage((Message) item); diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.2.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.2.xsd index 70935e8728..62b2005384 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.2.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.2.xsd @@ -3669,6 +3669,19 @@ + + + + + + + + + The channel where the splitter will send the messages that return an empty container from + split function. + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java index 8bdf0f83b0..327be1b923 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java @@ -21,7 +21,6 @@ import static org.assertj.core.api.Assertions.assertThat; import java.util.ArrayList; import java.util.Arrays; import java.util.List; -import java.util.concurrent.Executor; import java.util.stream.Collectors; import org.junit.Test; @@ -31,6 +30,7 @@ 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.aggregator.HeaderAttributeCorrelationStrategy; import org.springframework.integration.channel.QueueChannel; @@ -47,10 +47,10 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.GenericMessage; -import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit4.SpringRunner; +import com.fasterxml.jackson.databind.node.ArrayNode; import com.fasterxml.jackson.databind.node.TextNode; /** @@ -90,6 +90,9 @@ public class CorrelationHandlerTests { @Autowired private PollableChannel releaseChannel; + @Autowired + private PollableChannel discardChannel; + @Test public void testSplitterResequencer() { QueueChannel replyChannel = new QueueChannel(); @@ -150,17 +153,24 @@ public class CorrelationHandlerTests { assertThat(out.getPayload()).isEqualTo("bar"); } + @Test + public void testSplitterDiscard() { + this.splitAggregateInput.send(new GenericMessage<>(new ArrayList<>())); + Message receive = this.discardChannel.receive(10_000); + assertThat(receive) + .isNotNull() + .extracting(Message::getPayload) + .isInstanceOf(ArrayNode.class) + .extracting("_children") + .element(0) + .asList() + .hasSize(0); + } + @Configuration @EnableIntegration public static class ContextConfiguration { - @Bean - public Executor taskExecutor() { - ThreadPoolTaskExecutor tpte = new ThreadPoolTaskExecutor(); - tpte.setCorePoolSize(50); - return tpte; - } - @Bean public TestSplitterPojo testSplitterData() { List first = new ArrayList<>(); @@ -175,22 +185,22 @@ public class CorrelationHandlerTests { } @Bean - public MessageChannelSpec executorChannel() { - return MessageChannels.executor(taskExecutor()); + public MessageChannelSpec executorChannel(TaskExecutor taskExecutor) { + return MessageChannels.executor(taskExecutor); } @Bean @SuppressWarnings("rawtypes") - public IntegrationFlow splitResequenceFlow(MessageChannel executorChannel) { + public IntegrationFlow splitResequenceFlow(MessageChannel executorChannel, TaskExecutor taskExecutor) { return f -> f.enrichHeaders(s -> s.header("FOO", "BAR")) .split("testSplitterData", "buildList", c -> c.applySequence(false)) .channel(executorChannel) .split(Message.class, Message::getPayload, c -> c.applySequence(false)) - .channel(MessageChannels.executor(taskExecutor())) + .channel(MessageChannels.executor(taskExecutor)) .split(s -> s .applySequence(false) .delimiters(",")) - .channel(MessageChannels.executor(taskExecutor())) + .channel(MessageChannels.executor(taskExecutor)) .transform(Integer::parseInt) .enrichHeaders(h -> h.headerFunction(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, Message::getPayload)) @@ -203,8 +213,9 @@ public class CorrelationHandlerTests { public IntegrationFlow splitAggregateFlow() { return IntegrationFlows.from("splitAggregateInput", true) .transform(Transformers.toJson(ObjectToJsonTransformer.ResultType.NODE)) - .split() - .channel(MessageChannels.executor(taskExecutor())) + .split((splitter) -> splitter + .discardFlow((subFlow) -> subFlow.channel((c) -> c.queue("discardChannel")))) + .channel(MessageChannels.flux()) .resequence() .aggregate() .get(); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileSplitterParser.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileSplitterParser.java index dee666d1be..fe16f60bb9 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileSplitterParser.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileSplitterParser.java @@ -45,6 +45,7 @@ public class FileSplitterParser extends AbstractConsumerEndpointParser { IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "apply-sequence"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "first-line-as-header"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "discard-channel", "discardChannelName"); return builder; } diff --git a/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-5.2.xsd b/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-5.2.xsd index 05134fe5bc..4079b228ce 100644 --- a/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-5.2.xsd +++ b/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-5.2.xsd @@ -806,6 +806,19 @@ Only files matching this regular expression will be picked up by this adapter. + + + + + + + + + The channel where the splitter will send the messages that return an empty container from + split function. + + + diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileSplitterParserTests-context.xml b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileSplitterParserTests-context.xml index bdf7ee985b..91915678ab 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileSplitterParserTests-context.xml +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileSplitterParserTests-context.xml @@ -22,6 +22,7 @@ send-timeout="5" auto-startup="false" order="2" - phase="1"/> + phase="1" + discard-channel="nullChannel"/> diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileSplitterParserTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileSplitterParserTests.java index b8b79dd2ae..2bfa8a7c90 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileSplitterParserTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileSplitterParserTests.java @@ -66,6 +66,7 @@ public class FileSplitterParserTests { assertThat(TestUtils.getPropertyValue(this.splitter, "charset")).isEqualTo(Charset.forName("UTF-8")); assertThat(TestUtils.getPropertyValue(this.splitter, "messagingTemplate.sendTimeout")).isEqualTo(5L); assertThat(TestUtils.getPropertyValue(this.splitter, "firstLineHeaderName")).isEqualTo("foo"); + assertThat(TestUtils.getPropertyValue(this.splitter, "discardChannelName")).isEqualTo("nullChannel"); assertThat(this.splitter.getOutputChannel()).isSameAs(this.out); assertThat(this.splitter.getOrder()).isEqualTo(2); assertThat(this.fullBoat.getInputChannel()).isSameAs(this.in); 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 cb85aba12a..d7c4db595c 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 @@ -28,6 +28,7 @@ import java.io.IOException; import java.io.InputStream; import java.io.Reader; import java.nio.charset.Charset; +import java.nio.charset.StandardCharsets; import java.util.Date; import java.util.List; @@ -80,7 +81,7 @@ public class FileSplitterTests { private static File file; - private static final String SAMPLE_CONTENT = "HelloWorld\n????"; + private static final String SAMPLE_CONTENT = "HelloWorld\näöüß"; @Autowired private MessageChannel input1; @@ -97,7 +98,7 @@ public class FileSplitterTests { @BeforeClass public static void setup() throws IOException { file = File.createTempFile("foo", ".txt"); - FileCopyUtils.copy(SAMPLE_CONTENT.getBytes("UTF-8"), + FileCopyUtils.copy(SAMPLE_CONTENT.getBytes(StandardCharsets.UTF_8), new FileOutputStream(file, false)); } @@ -108,24 +109,24 @@ public class FileSplitterTests { @Test public void testFileSplitter() throws Exception { - this.input1.send(new GenericMessage(file)); + this.input1.send(new GenericMessage<>(file)); Message receive = this.output.receive(10000); assertThat(receive).isNotNull(); //HelloWorld assertThat(receive.getPayload()).isEqualTo("HelloWorld"); assertThat(receive.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)).isEqualTo(2); receive = this.output.receive(10000); - assertThat(receive).isNotNull(); //???? - assertThat(receive.getPayload()).isEqualTo("????"); + 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(this.output.receive(1)).isNull(); - this.input1.send(new GenericMessage(file.getAbsolutePath())); + this.input1.send(new GenericMessage<>(file.getAbsolutePath())); receive = this.output.receive(10000); assertThat(receive).isNotNull(); //HelloWorld assertThat(receive.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)).isEqualTo(2); receive = this.output.receive(10000); - assertThat(receive).isNotNull(); //???? + assertThat(receive).isNotNull(); //äöüß assertThat(receive.getHeaders().get(FileHeaders.ORIGINAL_FILE)).isEqualTo(file); assertThat(receive.getHeaders().get(FileHeaders.FILENAME)).isEqualTo(file.getName()); assertThat(this.output.receive(1)).isNull(); @@ -135,54 +136,56 @@ public class FileSplitterTests { assertThat(receive).isNotNull(); //HelloWorld assertThat(receive.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)).isEqualTo(2); receive = this.output.receive(10000); - assertThat(receive).isNotNull(); //???? + assertThat(receive).isNotNull(); //äöüß assertThat(this.output.receive(1)).isNull(); - this.input2.send(new GenericMessage(file)); + this.input2.send(new GenericMessage<>(file)); receive = this.output.receive(10000); assertThat(receive).isNotNull(); //HelloWorld assertThat(receive.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)).isEqualTo(0); receive = this.output.receive(10000); - assertThat(receive).isNotNull(); //???? + assertThat(receive).isNotNull(); //äöüß assertThat(this.output.receive(1)).isNull(); - this.input2.send(new GenericMessage(new ByteArrayInputStream(SAMPLE_CONTENT.getBytes("UTF-8")))); + this.input2.send(new GenericMessage( + new ByteArrayInputStream(SAMPLE_CONTENT.getBytes(StandardCharsets.UTF_8)))); receive = this.output.receive(10000); assertThat(receive).isNotNull(); //HelloWorld assertThat(receive.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)).isEqualTo(0); receive = this.output.receive(10000); - assertThat(receive).isNotNull(); //???? + assertThat(receive).isNotNull(); //äöüß assertThat(this.output.receive(1)).isNull(); try { - this.input2.send(new GenericMessage("bar")); + this.input2.send(new GenericMessage<>("bar")); fail("FileNotFoundException expected"); } catch (Exception e) { assertThat(e.getCause()).isInstanceOf(FileNotFoundException.class); assertThat(e.getMessage()).contains("failed to read file [bar]"); } - this.input2.send(new GenericMessage(new Date())); + this.input2.send(new GenericMessage<>(new Date())); receive = this.output.receive(10000); assertThat(receive).isNotNull(); assertThat(receive.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)).isEqualTo(1); assertThat(receive.getPayload()).isInstanceOf(Date.class); assertThat(this.output.receive(1)).isNull(); - this.input3.send(new GenericMessage(file)); + this.input3.send(new GenericMessage<>(file)); receive = this.output.receive(10000); assertThat(receive).isNotNull(); //HelloWorld assertThat(receive.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)).isEqualTo(0); receive = this.output.receive(10000); - assertThat(receive).isNotNull(); //???? + assertThat(receive).isNotNull(); //äöüß assertThat(this.output.receive(1)).isNull(); - this.input3.send(new GenericMessage(new ByteArrayInputStream(SAMPLE_CONTENT.getBytes("UTF-8")))); + this.input3.send(new GenericMessage<>( + new ByteArrayInputStream(SAMPLE_CONTENT.getBytes(StandardCharsets.UTF_8)))); receive = this.output.receive(10000); assertThat(receive).isNotNull(); //HelloWorld assertThat(receive.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)).isEqualTo(0); receive = this.output.receive(10000); - assertThat(receive).isNotNull(); //???? + assertThat(receive).isNotNull(); //äöüß assertThat(this.output.receive(1)).isNull(); } @@ -191,7 +194,7 @@ public class FileSplitterTests { QueueChannel outputChannel = new QueueChannel(); FileSplitter splitter = new FileSplitter(true, true); splitter.setOutputChannel(outputChannel); - splitter.handleMessage(new GenericMessage(file)); + splitter.handleMessage(new GenericMessage<>(file)); Message received = outputChannel.receive(0); assertThat(received).isNotNull(); assertThat(received.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)).isNull(); @@ -218,7 +221,7 @@ public class FileSplitterTests { FileSplitter splitter = new FileSplitter(true, true); splitter.setOutputChannel(outputChannel); File file = File.createTempFile("empty", ".txt"); - splitter.handleMessage(new GenericMessage(file)); + splitter.handleMessage(new GenericMessage<>(file)); Message received = outputChannel.receive(0); assertThat(received).isNotNull(); assertThat(received.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)).isNull(); @@ -246,7 +249,7 @@ public class FileSplitterTests { QueueChannel outputChannel = new QueueChannel(); FileSplitter splitter = new FileSplitter(true, true, true); splitter.setOutputChannel(outputChannel); - splitter.handleMessage(new GenericMessage(file)); + splitter.handleMessage(new GenericMessage<>(file)); Message received = outputChannel.receive(0); assertThat(received).isNotNull(); assertThat(received.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)).isNull(); @@ -263,7 +266,7 @@ public class FileSplitterTests { assertThat(received).isNotNull(); assertThat(received.getHeaders().get(FileHeaders.MARKER)).isEqualTo("END"); assertThat(received.getPayload()).isInstanceOf(String.class); - fileMarker = objectMapper.fromJson((String) received.getPayload(), FileSplitter.FileMarker.class); + fileMarker = objectMapper.fromJson(received.getPayload(), FileSplitter.FileMarker.class); assertThat(fileMarker.getMark()).isEqualTo(FileSplitter.FileMarker.Mark.END); assertThat(fileMarker.getFilePath()).isEqualTo(file.getAbsolutePath()); assertThat(fileMarker.getLineCount()).isEqualTo(2); diff --git a/spring-integration-xml/src/main/java/org/springframework/integration/xml/config/XPathMessageSplitterParser.java b/spring-integration-xml/src/main/java/org/springframework/integration/xml/config/XPathMessageSplitterParser.java index a5150d0d08..79279dfb09 100644 --- a/spring-integration-xml/src/main/java/org/springframework/integration/xml/config/XPathMessageSplitterParser.java +++ b/spring-integration-xml/src/main/java/org/springframework/integration/xml/config/XPathMessageSplitterParser.java @@ -42,7 +42,8 @@ public class XPathMessageSplitterParser extends AbstractConsumerEndpointParser { Assert.isTrue(xPathExpressionNodes.getLength() <= 1, "At most one xpath-expression child may be specified."); boolean hasChild = xPathExpressionNodes.getLength() == 1; boolean hasReference = StringUtils.hasText(xPathExpressionRef); - Assert.isTrue(hasChild ^ hasReference, "Exactly one of 'xpath-expression' or 'xpath-expression-ref' is required."); + Assert.isTrue(hasChild ^ hasReference, + "Exactly one of 'xpath-expression' or 'xpath-expression-ref' is required."); if (hasChild) { Element xpathExpressionElement = (Element) xPathExpressionNodes.item(0); builder.addConstructorArgValue(xpathExpressionElement.getAttribute("expression")); @@ -51,12 +52,14 @@ public class XPathMessageSplitterParser extends AbstractConsumerEndpointParser { else { builder.addConstructorArgReference(xPathExpressionRef); } - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "doc-builder-factory", "documentBuilder"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "doc-builder-factory", + "documentBuilder"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "create-documents"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "apply-sequence"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "iterator"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "output-properties"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "discard-channel", "discardChannelName"); return builder; } diff --git a/spring-integration-xml/src/main/resources/org/springframework/integration/xml/config/spring-integration-xml-5.2.xsd b/spring-integration-xml/src/main/resources/org/springframework/integration/xml/config/spring-integration-xml-5.2.xsd index a76815b486..9478bfae8a 100644 --- a/spring-integration-xml/src/main/resources/org/springframework/integration/xml/config/spring-integration-xml-5.2.xsd +++ b/spring-integration-xml/src/main/resources/org/springframework/integration/xml/config/spring-integration-xml-5.2.xsd @@ -820,6 +820,19 @@ + + + + + + + + + The channel where the splitter will send the messages that return an empty container from + split function. + + + diff --git a/spring-integration-xml/src/test/java/org/springframework/integration/xml/config/XPathSplitterParserTests-context.xml b/spring-integration-xml/src/test/java/org/springframework/integration/xml/config/XPathSplitterParserTests-context.xml index a4fc6354fe..b1b2a367a9 100644 --- a/spring-integration-xml/src/test/java/org/springframework/integration/xml/config/XPathSplitterParserTests-context.xml +++ b/spring-integration-xml/src/test/java/org/springframework/integration/xml/config/XPathSplitterParserTests-context.xml @@ -30,7 +30,8 @@ apply-sequence="false" create-documents="true" output-properties="outputProperties" - iterator="false"> + iterator="false" + discard-channel="nullChannel"> diff --git a/spring-integration-xml/src/test/java/org/springframework/integration/xml/config/XPathSplitterParserTests.java b/spring-integration-xml/src/test/java/org/springframework/integration/xml/config/XPathSplitterParserTests.java index 775bbfc765..577741d7c9 100644 --- a/spring-integration-xml/src/test/java/org/springframework/integration/xml/config/XPathSplitterParserTests.java +++ b/spring-integration-xml/src/test/java/org/springframework/integration/xml/config/XPathSplitterParserTests.java @@ -45,13 +45,16 @@ import org.springframework.util.MultiValueMap; @DirtiesContext public class XPathSplitterParserTests { - @Autowired @Qualifier("xpathSplitter.handler") + @Autowired + @Qualifier("xpathSplitter.handler") private MessageHandler xpathSplitter; - @Autowired @Qualifier("xpathSplitter") + @Autowired + @Qualifier("xpathSplitter") private EventDrivenConsumer consumer; - @Autowired @Qualifier("outputProperties") + @Autowired + @Qualifier("outputProperties") private Properties outputProperties; @Autowired @@ -64,10 +67,11 @@ public class XPathSplitterParserTests { assertThat(TestUtils.getPropertyValue(this.xpathSplitter, "returnIterator", Boolean.class)).isFalse(); assertThat(TestUtils.getPropertyValue(this.xpathSplitter, "outputProperties")).isSameAs(this.outputProperties); assertThat(TestUtils.getPropertyValue(this.xpathSplitter, - "xpathExpression.xpathExpression.xpath.m_patternString", - String.class)).isEqualTo("/orders/order"); + "xpathExpression.xpathExpression.xpath.m_patternString", String.class)) + .isEqualTo("/orders/order"); assertThat(TestUtils.getPropertyValue(xpathSplitter, "order")).isEqualTo(2); assertThat(TestUtils.getPropertyValue(xpathSplitter, "messagingTemplate.sendTimeout")).isEqualTo(123L); + assertThat(TestUtils.getPropertyValue(this.xpathSplitter, "discardChannelName")).isEqualTo("nullChannel"); assertThat(TestUtils.getPropertyValue(consumer, "phase")).isEqualTo(-1); assertThat(TestUtils.getPropertyValue(consumer, "autoStartup", Boolean.class)).isFalse(); @SuppressWarnings("unchecked") diff --git a/src/reference/asciidoc/dsl.adoc b/src/reference/asciidoc/dsl.adoc index 50a2c6283d..a5dbd5b300 100644 --- a/src/reference/asciidoc/dsl.adoc +++ b/src/reference/asciidoc/dsl.adoc @@ -455,15 +455,13 @@ The following example shows how to use the `split()` method by providing a lambd @Bean public IntegrationFlow splitFlow() { return IntegrationFlows.from("splitInput") - .split(s -> - s.applySequence(false).get().getT2().setDelimiters(",")) - .channel(MessageChannels.executor(this.taskExecutor())) + .split(s -> s.applySequence(false).delimiters(",")) + .channel(MessageChannels.executor(taskExecutor())) .get(); } ---- The preceding example creates a splitter that splits a message containing a comma-delimited `String`. -Note: The `getT2()` method comes from a `Tuple` `Collection`, which is the result of `EndpointSpec.get()`, and represents a pair of `ConsumerEndpointFactoryBean` and `DefaultMessageSplitter` for the preceding example. Also see <>. @@ -1157,7 +1155,7 @@ Otherwise, creating such a configuration by using `IntegrationFlow` does not mak By default a `GatewayProxyFactoryBean` gets a conventional bean name, such as `[FLOW_BEAN_NAME.gateway]`. You can change that ID by using the `@MessagingGateway.name()` attribute or the overloaded `from(Class serviceInterface, String beanName)` factory method. -With Java 8, you can even create an integration fateway with the `java.util.function` interfaces, as the following example shows: +With Java 8, you can even create an integration gateway with the `java.util.function` interfaces, as the following example shows: ==== [source,java] diff --git a/src/reference/asciidoc/splitter.adoc b/src/reference/asciidoc/splitter.adoc index 05c033a4ef..019f0ffbab 100644 --- a/src/reference/asciidoc/splitter.adoc +++ b/src/reference/asciidoc/splitter.adoc @@ -64,6 +64,10 @@ In this case, the target `Iterator` is built on their iteration functionality. In addition, if the splitter's output channel is an instance of a `ReactiveStreamsSubscribableChannel`, the `AbstractMessageSplitter` produces a `Flux` result instead of an `Iterator`, and the output channel is subscribed to this `Flux` for back-pressure-based splitting on downstream flow demand. +Starting with version 5.2, the splitter supports a `discardChannel` option for sending those request messages for which a split function has returned an empty container (collection, array, stream, `Flux` etc.). +In this case there is just no item to iterate for sending to the `outputChannel`. +The `null` splitting result remains as an end of flow indicator. + ==== Configuring a Splitter with XML A splitter can be configured through XML as follows: @@ -73,11 +77,12 @@ A splitter can be configured through XML as follows: ---- - - ref="splitterBean" <2> - method="split" <3> - input-channel="inputChannel" <4> - output-channel="outputChannel" /> <5> + + ref="splitterBean" <2> + method="split" <3> + input-channel="inputChannel" <4> + output-channel="outputChannel" <5> + discard-channel="discardChannel" /> <6> @@ -94,6 +99,8 @@ Optional. Required. <5> The channel to which the splitter sends the results of splitting the incoming message. Optional (because incoming messages can specify a reply channel themselves). +<6> The channel to which the request message is sent in case of empty splitting result. +Optional (the will stop as in case of `null` result). ==== We recommend using a `ref` attribute if the custom splitter implementation can be referenced in other `` definitions. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 2effb32a7e..55d46bec04 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -19,6 +19,9 @@ See <> for more information. The `JsonToObjectTransformer` now supports generics for the target object to deserialize into. See <> for more information. +RThe `splitter` now supports a `discardChannel` configuration option. +See <> for more information. + [[x5.2-amqp]] ==== AMQP Changes