diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java index f9fb6bd2d7..a0cc473920 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java @@ -2194,6 +2194,9 @@ public abstract class IntegrationFlowDefinition The full request {@link Message} will be logged. + *

When this operator is used in the end of flow, it is treated + * as one-way handler without any replies to continue. + * The {@link #logAndReply()} should be used for request-reply configuration. * @return the current {@link IntegrationFlowDefinition}. * @see #wireTap(WireTapSpec) */ @@ -2207,6 +2210,9 @@ public abstract class IntegrationFlowDefinition The full request {@link Message} will be logged. + *

When this operator is used in the end of flow, it is treated + * as one-way handler without any replies to continue. + * The {@link #logAndReply()} should be used for request-reply configuration. * @param level the {@link LoggingHandler.Level}. * @return the current {@link IntegrationFlowDefinition}. * @see #wireTap(WireTapSpec) @@ -2220,6 +2226,9 @@ public abstract class IntegrationFlowDefinition The full request {@link Message} will be logged. + *

When this operator is used in the end of flow, it is treated + * as one-way handler without any replies to continue. + * The {@link #logAndReply()} should be used for request-reply configuration. * @param category the logging category to use. * @return the current {@link IntegrationFlowDefinition}. * @see #wireTap(WireTapSpec) @@ -2233,6 +2242,9 @@ public abstract class IntegrationFlowDefinition The full request {@link Message} will be logged. + *

When this operator is used in the end of flow, it is treated + * as one-way handler without any replies to continue. + * The {@link #logAndReply()} should be used for request-reply configuration. * @param level the {@link LoggingHandler.Level}. * @param category the logging category to use. * @return the current {@link IntegrationFlowDefinition}. @@ -2247,6 +2259,9 @@ public abstract class IntegrationFlowDefinition When this operator is used in the end of flow, it is treated + * as one-way handler without any replies to continue. + * The {@link #logAndReply()} should be used for request-reply configuration. * @param level the {@link LoggingHandler.Level}. * @param category the logging category. * @param logExpression the SpEL expression to evaluate logger message at runtime @@ -2264,6 +2279,9 @@ public abstract class IntegrationFlowDefinition When this operator is used in the end of flow, it is treated + * as one-way handler without any replies to continue. + * The {@link #logAndReply()} should be used for request-reply configuration. * @param function the function to evaluate logger message at runtime * @param

the expected payload type. * against the request {@link Message}. @@ -2281,6 +2299,9 @@ public abstract class IntegrationFlowDefinition When this operator is used in the end of flow, it is treated + * as one-way handler without any replies to continue. + * The {@link #logAndReply()} should be used for request-reply configuration. * @param logExpression the {@link Expression} to evaluate logger message at runtime * against the request {@link Message}. * @return the current {@link IntegrationFlowDefinition}. @@ -2297,6 +2318,9 @@ public abstract class IntegrationFlowDefinition When this operator is used in the end of flow, it is treated + * as one-way handler without any replies to continue. + * The {@link #logAndReply()} should be used for request-reply configuration. * @param level the {@link LoggingHandler.Level}. * @param logExpression the {@link Expression} to evaluate logger message at runtime * against the request {@link Message}. @@ -2307,13 +2331,15 @@ public abstract class IntegrationFlowDefinition When this operator is used in the end of flow, it is treated + * as one-way handler without any replies to continue. + * The {@link #logAndReply()} should be used for request-reply configuration. * @param category the logging category. * @param logExpression the {@link Expression} to evaluate logger message at runtime * against the request {@link Message}. @@ -2330,6 +2356,9 @@ public abstract class IntegrationFlowDefinition When this operator is used in the end of flow, it is treated + * as one-way handler without any replies to continue. + * The {@link #logAndReply()} should be used for request-reply configuration. * @param level the {@link LoggingHandler.Level}. * @param function the function to evaluate logger message at runtime * @param

the expected payload type. @@ -2346,6 +2375,9 @@ public abstract class IntegrationFlowDefinition When this operator is used in the end of flow, it is treated + * as one-way handler without any replies to continue. + * The {@link #logAndReply()} should be used for request-reply configuration. * @param category the logging category. * @param function the function to evaluate logger message at runtime * @param

the expected payload type. @@ -2362,6 +2394,9 @@ public abstract class IntegrationFlowDefinition When this operator is used in the end of flow, it is treated + * as one-way handler without any replies to continue. + * The {@link #logAndReply()} should be used for request-reply configuration. * @param level the {@link LoggingHandler.Level}. * @param category the logging category. * @param function the function to evaluate logger message at runtime @@ -2375,12 +2410,14 @@ public abstract class IntegrationFlowDefinition(function)); } - /** * Populate a {@link WireTap} for the {@link #currentMessageChannel} * with the {@link LoggingHandler} subscriber for the provided * {@link LoggingHandler.Level} logging level, logging category * and SpEL expression for the log message. + *

When this operator is used in the end of flow, it is treated + * as one-way handler without any replies to continue. + * The {@link #logAndReply()} should be used for request-reply configuration. * @param level the {@link LoggingHandler.Level}. * @param category the logging category. * @param logExpression the {@link Expression} to evaluate logger message at runtime @@ -2406,6 +2443,264 @@ public abstract class IntegrationFlowDefinition The full request {@link Message} will be logged. + *

A {@link #bridge()} is added after this operator to make the flow reply-producing + * if the {@code replyChannel} header is present. + *

This operator can be used only in the end of flow. + * @return an {@link IntegrationFlow} instance based on this builder. + * @see #log() + * @see #bridge() + */ + public IntegrationFlow logAndReply() { + return logAndReply(LoggingHandler.Level.INFO); + } + + /** + * Populate a {@link WireTap} for the {@link #currentMessageChannel} + * with the {@link LoggingHandler} subscriber for provided {@link LoggingHandler.Level} + * logging level and {@code org.springframework.integration.handler.LoggingHandler} + * as a default logging category. + *

The full request {@link Message} will be logged. + *

A {@link #bridge()} is added after this operator to make the flow reply-producing + * if the {@code replyChannel} header is present. + *

This operator can be used only in the end of flow. + * @param level the {@link LoggingHandler.Level}. + * @return an {@link IntegrationFlow} instance based on this builder. + * @see #log() + * @see #bridge() + */ + public IntegrationFlow logAndReply(LoggingHandler.Level level) { + return logAndReply(level, (String) null); + } + + /** + * Populate a {@link WireTap} for the {@link #currentMessageChannel} + * with the {@link LoggingHandler} subscriber for the provided logging category + * and {@code INFO} logging level. + *

The full request {@link Message} will be logged. + *

A {@link #bridge()} is added after this operator to make the flow reply-producing + * if the {@code replyChannel} header is present. + *

This operator can be used only in the end of flow. + * @param category the logging category to use. + * @return an {@link IntegrationFlow} instance based on this builder. + * @see #log() + * @see #bridge() + */ + public IntegrationFlow logAndReply(String category) { + return logAndReply(LoggingHandler.Level.INFO, category); + } + + /** + * Populate a {@link WireTap} for the {@link #currentMessageChannel} + * with the {@link LoggingHandler} subscriber for the provided + * {@link LoggingHandler.Level} logging level and logging category. + *

The full request {@link Message} will be logged. + *

A {@link #bridge()} is added after this operator to make the flow reply-producing + * if the {@code replyChannel} header is present. + *

This operator can be used only in the end of flow. + * @param level the {@link LoggingHandler.Level}. + * @param category the logging category to use. + * @return an {@link IntegrationFlow} instance based on this builder. + * @see #log() + * @see #bridge() + */ + public IntegrationFlow logAndReply(LoggingHandler.Level level, String category) { + return logAndReply(level, category, (Expression) null); + } + + /** + * Populate a {@link WireTap} for the {@link #currentMessageChannel} + * with the {@link LoggingHandler} subscriber for the provided + * {@link LoggingHandler.Level} logging level, logging category + * and SpEL expression for the log message. + *

A {@link #bridge()} is added after this operator to make the flow reply-producing + * if the {@code replyChannel} header is present. + *

This operator can be used only in the end of flow. + * @param level the {@link LoggingHandler.Level}. + * @param category the logging category. + * @param logExpression the SpEL expression to evaluate logger message at runtime + * against the request {@link Message}. + * @return an {@link IntegrationFlow} instance based on this builder. + * @see #log() + * @see #bridge() + */ + public IntegrationFlow logAndReply(LoggingHandler.Level level, String category, String logExpression) { + Assert.hasText(logExpression, "'logExpression' must not be empty"); + return logAndReply(level, category, PARSER.parseExpression(logExpression)); + } + + /** + * Populate a {@link WireTap} for the {@link #currentMessageChannel} + * with the {@link LoggingHandler} subscriber for the {@code INFO} logging level, + * the {@code org.springframework.integration.handler.LoggingHandler} + * as a default logging category and {@link Function} for the log message. + *

A {@link #bridge()} is added after this operator to make the flow reply-producing + * if the {@code replyChannel} header is present. + *

This operator can be used only in the end of flow. + * @param function the function to evaluate logger message at runtime + * @param

the expected payload type. + * against the request {@link Message}. + * @return an {@link IntegrationFlow} instance based on this builder. + * @see #log() + * @see #bridge() + */ + public

IntegrationFlow logAndReply(Function, Object> function) { + Assert.notNull(function, "'function' must not be null"); + return logAndReply(new FunctionExpression<>(function)); + } + + /** + * Populate a {@link WireTap} for the {@link #currentMessageChannel} + * with the {@link LoggingHandler} subscriber for the {@code INFO} logging level, + * the {@code org.springframework.integration.handler.LoggingHandler} + * as a default logging category and SpEL expression to evaluate + * logger message at runtime against the request {@link Message}. + *

A {@link #bridge()} is added after this operator to make the flow reply-producing + * if the {@code replyChannel} header is present. + *

This operator can be used only in the end of flow. + * @param logExpression the {@link Expression} to evaluate logger message at runtime + * against the request {@link Message}. + * @return an {@link IntegrationFlow} instance based on this builder. + * @see #log() + * @see #bridge() + */ + public IntegrationFlow logAndReply(Expression logExpression) { + return logAndReply(LoggingHandler.Level.INFO, logExpression); + } + + /** + * Populate a {@link WireTap} for the {@link #currentMessageChannel} + * with the {@link LoggingHandler} subscriber for the provided + * {@link LoggingHandler.Level} logging level, + * the {@code org.springframework.integration.handler.LoggingHandler} + * as a default logging category and SpEL expression to evaluate + * logger message at runtime against the request {@link Message}. + *

A {@link #bridge()} is added after this operator to make the flow reply-producing + * if the {@code replyChannel} header is present. + *

This operator can be used only in the end of flow. + * @param level the {@link LoggingHandler.Level}. + * @param logExpression the {@link Expression} to evaluate logger message at runtime + * against the request {@link Message}. + * @return an {@link IntegrationFlow} instance based on this builder. + * @see #log() + * @see #bridge() + */ + public IntegrationFlow logAndReply(LoggingHandler.Level level, Expression logExpression) { + return logAndReply(level, null, logExpression); + } + + /** + * Populate a {@link WireTap} for the {@link #currentMessageChannel} + * with the {@link LoggingHandler} subscriber for the {@code INFO} + * {@link LoggingHandler.Level} logging level, + * the provided logging category and SpEL expression to evaluate + * logger message at runtime against the request {@link Message}. + *

A {@link #bridge()} is added after this operator to make the flow reply-producing + * if the {@code replyChannel} header is present. + *

This operator can be used only in the end of flow. + * @param category the logging category. + * @param logExpression the {@link Expression} to evaluate logger message at runtime + * against the request {@link Message}. + * @return an {@link IntegrationFlow} instance based on this builder. + * @see #log() + * @see #bridge() + */ + public IntegrationFlow logAndReply(String category, Expression logExpression) { + return logAndReply(LoggingHandler.Level.INFO, category, logExpression); + } + + /** + * Populate a {@link WireTap} for the {@link #currentMessageChannel} + * with the {@link LoggingHandler} subscriber for the provided + * {@link LoggingHandler.Level} logging level, + * the {@code org.springframework.integration.handler.LoggingHandler} + * as a default logging category and {@link Function} for the log message. + *

A {@link #bridge()} is added after this operator to make the flow reply-producing + * if the {@code replyChannel} header is present. + *

This operator can be used only in the end of flow. + * @param level the {@link LoggingHandler.Level}. + * @param function the function to evaluate logger message at runtime + * @param

the expected payload type. + * against the request {@link Message}. + * @return an {@link IntegrationFlow} instance based on this builder. + * @see #log() + * @see #bridge() + */ + public

IntegrationFlow logAndReply(LoggingHandler.Level level, Function, Object> function) { + return logAndReply(level, null, function); + } + + /** + * Populate a {@link WireTap} for the {@link #currentMessageChannel} + * with the {@link LoggingHandler} subscriber for the provided + * {@link LoggingHandler.Level} logging level, + * the provided logging category and {@link Function} for the log message. + *

A {@link #bridge()} is added after this operator to make the flow reply-producing + * if the {@code replyChannel} header is present. + *

This operator can be used only in the end of flow. + * @param category the logging category. + * @param function the function to evaluate logger message at runtime + * @param

the expected payload type. + * against the request {@link Message}. + * @return an {@link IntegrationFlow} instance based on this builder. + * @see #log() + * @see #bridge() + */ + public

IntegrationFlow logAndReply(String category, Function, Object> function) { + return logAndReply(LoggingHandler.Level.INFO, category, function); + } + + /** + * Populate a {@link WireTap} for the {@link #currentMessageChannel} + * with the {@link LoggingHandler} subscriber for the provided + * {@link LoggingHandler.Level} logging level, logging category + * and {@link Function} for the log message. + *

A {@link #bridge()} is added after this operator to make the flow reply-producing + * if the {@code replyChannel} header is present. + *

This operator can be used only in the end of flow. + * @param level the {@link LoggingHandler.Level}. + * @param category the logging category. + * @param function the function to evaluate logger message at runtime + * @param

the expected payload type. + * against the request {@link Message}. + * @return an {@link IntegrationFlow} instance based on this builder. + * @see #log() + * @see #bridge() + */ + public

IntegrationFlow logAndReply(LoggingHandler.Level level, String category, + Function, Object> function) { + + Assert.notNull(function, "'function' must not be null"); + return logAndReply(level, category, new FunctionExpression<>(function)); + } + + /** + * Populate a {@link WireTap} for the {@link #currentMessageChannel} + * with the {@link LoggingHandler} subscriber for the provided + * {@link LoggingHandler.Level} logging level, logging category + * and SpEL expression for the log message. + *

A {@link #bridge()} is added after this operator to make the flow reply-producing + * if the {@code replyChannel} header is present. + *

This operator can be used only in the end of flow. + * @param level the {@link LoggingHandler.Level}. + * @param category the logging category. + * @param logExpression the {@link Expression} to evaluate logger message at runtime + * against the request {@link Message}. + * @return an {@link IntegrationFlow} instance based on this builder. + * @see #log() + * @see #bridge() + */ + public IntegrationFlow logAndReply(LoggingHandler.Level level, String category, Expression logExpression) { + return log(level, category, logExpression) + .bridge() + .get(); + } + /** * Populate a {@link ScatterGatherHandler} to the current integration flow position * based on the provided {@link MessageChannel} for scattering function @@ -2603,6 +2898,7 @@ public abstract class IntegrationFlowDefinition> B register(S endpointSpec, Consumer endpointConfigurer) { + if (endpointConfigurer != null) { endpointConfigurer.accept(endpointSpec); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/HeaderEnricher.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/HeaderEnricher.java index 5b65bc3718..efaaa3e24b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transformer/HeaderEnricher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/HeaderEnricher.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 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. @@ -29,11 +29,11 @@ import org.springframework.beans.factory.BeanNameAware; import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.handler.MessageProcessor; +import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.integration.support.DefaultMessageBuilderFactory; import org.springframework.integration.transformer.support.HeaderValueMessageProcessor; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; -import org.springframework.messaging.MessagingException; /** * A Transformer that adds statically configured header values to a Message. @@ -64,12 +64,13 @@ public class HeaderEnricher extends IntegrationObjectSupport implements Transfor /** * Create a HeaderEnricher with the given map of headers. - * * @param headersToAdd The headers to add. */ public HeaderEnricher(Map> headersToAdd) { - this.headersToAdd = (headersToAdd != null) ? headersToAdd - : new HashMap>(); + this.headersToAdd = + headersToAdd != null + ? headersToAdd + : new HashMap>(); } public void setMessageProcessor(MessageProcessor messageProcessor) { @@ -86,7 +87,6 @@ public class HeaderEnricher extends IntegrationObjectSupport implements Transfor * true. Set this to false if a * null value should trigger removal of the * corresponding header instead. - * * @param shouldSkipNulls true when null values should be skipped. */ public void setShouldSkipNulls(boolean shouldSkipNulls) { @@ -96,52 +96,56 @@ public class HeaderEnricher extends IntegrationObjectSupport implements Transfor @Override public String getComponentType() { - return "transformer"; // backwards compatibility + return "header-enricher"; } @Override public Message transform(Message message) { - try { - Map headerMap = new HashMap(message.getHeaders()); - this.addHeadersFromMessageProcessor(message, headerMap); - for (Map.Entry> entry : this.headersToAdd.entrySet()) { - String key = entry.getKey(); - HeaderValueMessageProcessor valueProcessor = entry.getValue(); + MessageHeaders messageHeaders = message.getHeaders(); - Boolean shouldOverwrite = valueProcessor.isOverwrite(); - if (shouldOverwrite == null) { - shouldOverwrite = this.defaultOverwrite; - } + AbstractIntegrationMessageBuilder messageBuilder = + getMessageBuilderFactory() + .fromMessage(message); - boolean headerDoesNotExist = headerMap.get(key) == null; + addHeadersFromMessageProcessor(message, messageBuilder); + for (Map.Entry> entry : this.headersToAdd.entrySet()) { + String key = entry.getKey(); + HeaderValueMessageProcessor valueProcessor = entry.getValue(); - /* - * Only evaluate value expression if necessary - */ - if (headerDoesNotExist || shouldOverwrite) { - Object value = valueProcessor.processMessage(message); - if (value != null || !this.shouldSkipNulls) { - headerMap.put(key, value); - } + Boolean shouldOverwrite = valueProcessor.isOverwrite(); + if (shouldOverwrite == null) { + shouldOverwrite = this.defaultOverwrite; + } + + boolean headerDoesNotExist = messageHeaders.get(key) == null; + + /* + * Only evaluate value expression if necessary + */ + if (headerDoesNotExist || shouldOverwrite) { + Object value = valueProcessor.processMessage(message); + if (value != null || !this.shouldSkipNulls) { + messageBuilder.setHeader(key, value); } } - return this.getMessageBuilderFactory().withPayload(message.getPayload()).copyHeaders(headerMap).build(); - } - catch (Exception e) { - throw new MessagingException(message, "failed to transform message headers", e); } + + return messageBuilder.build(); } - private void addHeadersFromMessageProcessor(Message message, Map headerMap) { + private void addHeadersFromMessageProcessor(Message message, + AbstractIntegrationMessageBuilder messageBuilder) { + if (this.messageProcessor != null) { Object result = this.messageProcessor.processMessage(message); if (result instanceof Map) { + MessageHeaders messageHeaders = message.getHeaders(); Map resultMap = (Map) result; for (Entry entry : resultMap.entrySet()) { Object key = entry.getKey(); if (key instanceof String) { - if (this.defaultOverwrite || headerMap.get(key) == null) { - headerMap.put((String) key, entry.getValue()); + if (this.defaultOverwrite || messageHeaders.get(key) == null) { + messageBuilder.setHeader((String) key, entry.getValue()); } } else if (logger.isDebugEnabled()) { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/flowservices/FlowServiceTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/flowservices/FlowServiceTests.java index 9868d2ffd4..3226f9df8d 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/flowservices/FlowServiceTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/flowservices/FlowServiceTests.java @@ -133,8 +133,7 @@ public class FlowServiceTests { @Bean public IntegrationFlow testGateway() { return f -> f.gateway("processChannel", g -> g.replyChannel("replyChannel")) - .log() - .bridge(); + .logAndReply(); } @Bean diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/transformers/TransformerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/transformers/TransformerTests.java index 4f8e268dc0..8704e02497 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/transformers/TransformerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/transformers/TransformerTests.java @@ -266,7 +266,7 @@ public class TransformerTests { .propertyFunction("date", m -> new Date()) .headerExpression("foo", "payload['name']") ) - .get(); + .logAndReply(); } @Bean diff --git a/spring-integration-core/src/test/java/org/springframework/integration/history/MessageHistoryIntegrationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/history/MessageHistoryIntegrationTests.java index b2e3cf3d5f..f1b5cf1df2 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/history/MessageHistoryIntegrationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/history/MessageHistoryIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2018 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. @@ -61,7 +61,8 @@ public class MessageHistoryIntegrationTests { for (ConsumerEndpointFactoryBean cefBean : cefBeans.values()) { DirectFieldAccessor bridgeAccessor = new DirectFieldAccessor(cefBean); String handlerClassName = bridgeAccessor.getPropertyValue("handler").getClass().getName(); - assertFalse("org.springframework.integration.config.MessageHistoryWritingMessageHandler".equals(handlerClassName)); + assertFalse("org.springframework.integration.config.MessageHistoryWritingMessageHandler" + .equals(handlerClassName)); } ac.close(); } @@ -72,10 +73,12 @@ public class MessageHistoryIntegrationTests { MessageHistoryIntegrationTests.class); SampleGateway gateway = ac.getBean("sampleGateway", SampleGateway.class); DirectChannel endOfThePipeChannel = ac.getBean("endOfThePipeChannel", DirectChannel.class); - MessageHandler handler = Mockito.spy(new MessageHandler() { + MessageHandler handler = Mockito.spy(new MessageHandler() { // Not a lambda: Mockito can't mock final classes + @Override public void handleMessage(Message message) { - Iterator historyIterator = message.getHeaders().get(MessageHistory.HEADER_NAME, MessageHistory.class).iterator(); + Iterator historyIterator = message.getHeaders() + .get(MessageHistory.HEADER_NAME, MessageHistory.class).iterator(); Properties event = historyIterator.next(); assertEquals("sampleGateway", event.getProperty(MessageHistory.NAME_PROPERTY)); @@ -95,7 +98,7 @@ public class MessageHistoryIntegrationTests { event = historyIterator.next(); assertEquals("testHeaderEnricher", event.getProperty(MessageHistory.NAME_PROPERTY)); - assertEquals("transformer", event.getProperty(MessageHistory.TYPE_PROPERTY)); + assertEquals("header-enricher", event.getProperty(MessageHistory.TYPE_PROPERTY)); event = historyIterator.next(); assertEquals("chainChannel", event.getProperty(MessageHistory.NAME_PROPERTY)); @@ -106,7 +109,8 @@ public class MessageHistoryIntegrationTests { assertEquals("chain", event.getProperty(MessageHistory.TYPE_PROPERTY)); event = historyIterator.next(); - assertEquals("sampleChain$child.service-activator-within-chain", event.getProperty(MessageHistory.NAME_PROPERTY)); + assertEquals("sampleChain$child.service-activator-within-chain", event + .getProperty(MessageHistory.NAME_PROPERTY)); assertEquals("service-activator", event.getProperty(MessageHistory.TYPE_PROPERTY)); event = historyIterator.next(); @@ -155,7 +159,8 @@ public class MessageHistoryIntegrationTests { MessageHistoryIntegrationTests.class); SampleGateway gateway = ac.getBean("sampleGateway", SampleGateway.class); DirectChannel endOfThePipeChannel = ac.getBean("endOfThePipeChannel", DirectChannel.class); - MessageHandler handler = Mockito.spy(new MessageHandler() { + MessageHandler handler = Mockito.spy(new MessageHandler() { // Not a lambda: Mockito can't mock final classes + @Override public void handleMessage(Message message) { assertNull(message.getHeaders().get(MessageHistory.HEADER_NAME, MessageHistory.class)); @@ -175,7 +180,8 @@ public class MessageHistoryIntegrationTests { "messageHistoryWithHistoryWriterNamespace.xml", MessageHistoryIntegrationTests.class); SampleGateway gateway = ac.getBean("sampleGateway", SampleGateway.class); DirectChannel endOfThePipeChannel = ac.getBean("endOfThePipeChannel", DirectChannel.class); - MessageHandler handler = Mockito.spy(new MessageHandler() { + MessageHandler handler = Mockito.spy(new MessageHandler() { // Not a lambda: Mockito can't mock final classes + @Override public void handleMessage(Message message) { Iterator historyIterator = message.getHeaders() @@ -197,10 +203,12 @@ public class MessageHistoryIntegrationTests { "messageHistoryWithHistoryWriterNamespaceAndPatterns.xml", MessageHistoryIntegrationTests.class); SampleGateway gateway = ac.getBean("sampleGateway", SampleGateway.class); DirectChannel endOfThePipeChannel = ac.getBean("endOfThePipeChannel", DirectChannel.class); - MessageHandler handler = Mockito.spy(new MessageHandler() { + MessageHandler handler = Mockito.spy(new MessageHandler() { // Not a lambda: Mockito can't mock final classes + @Override public void handleMessage(Message message) { - Iterator historyIterator = message.getHeaders().get(MessageHistory.HEADER_NAME, MessageHistory.class).iterator(); + Iterator historyIterator = message.getHeaders() + .get(MessageHistory.HEADER_NAME, MessageHistory.class).iterator(); assertTrue(historyIterator.hasNext()); Properties gatewayHistory = historyIterator.next(); assertEquals("sampleGateway", gatewayHistory.get("name")); @@ -224,7 +232,8 @@ public class MessageHistoryIntegrationTests { MessageHistoryIntegrationTests.class).close(); } - @Test @Ignore + @Test + @Ignore public void testMessageHistoryWithHistoryPerformance() { ConfigurableApplicationContext acWithHistory = new ClassPathXmlApplicationContext("perfWithMessageHistory.xml", MessageHistoryIntegrationTests.class); @@ -264,7 +273,9 @@ public class MessageHistoryIntegrationTests { } public interface SampleGateway { + Message echo(String value); + } } diff --git a/src/reference/asciidoc/dsl.adoc b/src/reference/asciidoc/dsl.adoc index f140206366..4eb03c8877 100644 --- a/src/reference/asciidoc/dsl.adoc +++ b/src/reference/asciidoc/dsl.adoc @@ -511,14 +511,31 @@ public IntegrationFlow integerFlow() { ---- ==== -We also register a `BytesToIntegerConverter` within `ConversionService` to get rid of that additional `.transform()`. -// TODO We don't show how to register a `BytesToIntegerConverter` within `ConversionService` to get rid of that additional `.transform()`. +We also can register some `BytesToIntegerConverter` within `ConversionService` to get rid of that additional `.transform()`: + +==== +[source,java] +---- +@Bean +@IntegrationConverter +public BytesToIntegerConverter bytesToIntegerConverter() { + return new BytesToIntegerConverter(); +} + +@Bean +public IntegrationFlow integerFlow() { + return IntegrationFlows.from("input") + .handle(Integer.class, (p, h) -> p * 2) + .get(); +} +---- +==== [[java-dsl-log]] === Operator log() For convenience, to log the message journey through the Spring Integration flow (``), a `log()` operator is presented. -Internally, it is represented by the `WireTap` `ChannelInterceptor` and `LoggingHandler` as subscriber. +Internally, it is represented by the `WireTap` `ChannelInterceptor` with a `LoggingHandler` as its subscriber. It is responsible for logging the incoming message into the next endpoint or the current channel. The following example shows how to use `LoggingHandler`: @@ -533,6 +550,10 @@ The following example shows how to use `LoggingHandler`: In the preceding example, an `id` header is logged at the `ERROR` level onto `test.category` only for messages that passed the filter and before routing. +When this operator is used at the end of a flow, it is a one-way handler and the flow ends. +To make it as a reply-producing flow, you can either use a simple `bridge()` after the `log()` or, starting with version 5.1, you can use a `logAndReply()` operator instead. +`logAndReply` can only be used at the end of a flow. + [[java-dsl-wiretap]] === `MessageChannelSpec.wireTap()` @@ -589,26 +610,6 @@ The following example does not have any channel declaration: In the preceding example (and any time no channel has been declared), an implicit `DirectChannel` is injected in the current position of the `IntegrationFlow` and used as an output channel for the currently configured `ServiceActivatingHandler` (from the `.handle()`, <>). -[IMPORTANT] -==== -If `log()` or `wireTap()` are used in the end of the flow, they are considered to be one-way `MessageHandler` instances. -If you expect the integration flow to return a reply, you should add a `bridge()` should to the end, after `log()` or `wireTap()`, as the following example shows: - -[source,java] ----- -@Bean -public IntegrationFlow sseFlow() { - return IntegrationFlows - .from(WebFlux.inboundGateway("/sse") - .requestMapping(m -> - m.produces(MediaType.TEXT_EVENT_STREAM_VALUE))) - .handle((p, h) -> Flux.just("foo", "bar", "baz")) - .log(LoggingHandler.Level.WARN) - .bridge() - .get(); -} ----- -==== [[java-dsl-flows]] === Working With Message Flows diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index d28ee8ad50..aef02d411f 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -36,6 +36,9 @@ The following changes have been made in version 5.1: The `IntegrationFlowContext` is now an interface and `IntegrationFlowRegistration` is an inner interface of `IntegrationFlowContext`. +A new `logAndReply()` operator has been introduced for convenience when you wish to log at the end of a flow for request-reply configurations. +This avoid confusion with `log()` which is treated as a one-way end flow component. + [[x5.1-dispatcher-exceptions]] ==== Dispatcher Exceptions