From 3035bc716d6bc8199a5f917ceb7d46c3074a42e3 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 14 Nov 2016 16:04:17 -0500 Subject: [PATCH] INT-4138: MQTT: Outbound Adapter Improvements JIRA:https://jira.spring.io/browse/INT-4138 Expressions for topic, qos, retained. Also change inbound mapping to `RECEIVED_...` headers. Fix some minor asciidoc problems in (s)ftp. Rework Qos/Retained Expressions/Defaults Encapsulate the logic entirely in the converter. Polishing - PR Comments --- .../ExpressionEvaluatingMessageProcessor.java | 33 +++++ .../xml/MqttOutboundChannelAdapterParser.java | 5 + .../outbound/AbstractMqttMessageHandler.java | 139 ++++++++++++++++-- .../support/DefaultPahoMessageConverter.java | 87 ++++++++--- .../integration/mqtt/support/MqttHeaders.java | 7 + .../mqtt/support/MqttMessageConverter.java | 10 ++ .../config/spring-integration-mqtt-5.0.xsd | 34 ++++- .../mqtt/BackToBackAdapterTests.java | 28 ++-- .../mqtt/DownstreamExceptionTests.java | 6 +- .../integration/mqtt/MqttAdapterTests.java | 70 +++++++++ ...essageDrivenChannelAdapterParserTests.java | 8 +- ...boundChannelAdapterParserTests-context.xml | 3 + ...MqttOutboundChannelAdapterParserTests.java | 33 +++-- src/reference/asciidoc/ftp.adoc | 2 +- src/reference/asciidoc/mqtt.adoc | 123 +++++++++------- src/reference/asciidoc/sftp.adoc | 2 +- src/reference/asciidoc/whats-new.adoc | 8 + 17 files changed, 471 insertions(+), 127 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/ExpressionEvaluatingMessageProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/ExpressionEvaluatingMessageProcessor.java index 4b89402af1..04955867e9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/ExpressionEvaluatingMessageProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/ExpressionEvaluatingMessageProcessor.java @@ -27,6 +27,7 @@ import org.springframework.util.Assert; * * @author Mark Fisher * @author Artem Bilan + * @author Gary Russell * @since 2.0 */ public class ExpressionEvaluatingMessageProcessor extends AbstractMessageProcessor { @@ -61,6 +62,38 @@ public class ExpressionEvaluatingMessageProcessor extends AbstractMessageProc } } + /** + * Create an {@link ExpressionEvaluatingMessageProcessor} for the given expression. + * @param expression a SpEL expression to evaluate. + * @since 5.0 + */ + public ExpressionEvaluatingMessageProcessor(String expression) { + try { + this.expression = EXPRESSION_PARSER.parseExpression(expression); + this.expectedType = null; + } + catch (ParseException e) { + throw new IllegalArgumentException("Failed to parse expression.", e); + } + } + + /** + * Construct {@link ExpressionEvaluatingMessageProcessor} for the provided + * SpEL expression and expected result type. + * @param expression a SpEL expression to evaluate. + * @param expectedType the expected result type. + * @since 5.0 + */ + public ExpressionEvaluatingMessageProcessor(String expression, Class expectedType) { + try { + this.expression = EXPRESSION_PARSER.parseExpression(expression); + this.expectedType = expectedType; + } + catch (ParseException e) { + throw new IllegalArgumentException("Failed to parse expression.", e); + } + } + /** * Processes the Message by evaluating the expression with that Message as the * root object. The expression evaluation result Object will be returned. diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParser.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParser.java index de510dad76..ac89823daf 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParser.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParser.java @@ -52,6 +52,8 @@ public class MqttOutboundChannelAdapterParser extends AbstractOutboundChannelAda MqttParserUtils.parseCommon(element, builder, parserContext); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "default-topic"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "topic-expression", + "topicExpressionString"); if (StringUtils.hasText(element.getAttribute("converter")) && (StringUtils.hasText(element.getAttribute("default-qos")) || StringUtils.hasText(element.getAttribute("default-retained")))) { @@ -59,7 +61,10 @@ public class MqttOutboundChannelAdapterParser extends AbstractOutboundChannelAda "'default-qos' or 'default-retained'", element); } IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "default-qos"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "qos-expression", "qosExpressionString"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "default-retained"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "retained-expression", + "retainedExpressionString"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "async"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "async-events"); diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java index c2152fd682..40dc499730 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java @@ -18,10 +18,15 @@ package org.springframework.integration.mqtt.outbound; import java.util.concurrent.atomic.AtomicBoolean; +import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.context.Lifecycle; +import org.springframework.expression.Expression; import org.springframework.integration.handler.AbstractMessageHandler; +import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor; +import org.springframework.integration.handler.MessageProcessor; import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter; import org.springframework.integration.mqtt.support.MqttHeaders; +import org.springframework.integration.mqtt.support.MqttMessageConverter; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.converter.MessageConverter; @@ -37,21 +42,30 @@ import org.springframework.util.Assert; */ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler implements Lifecycle { + private static final MessageProcessor DEFAULT_TOPIC_PROCESSOR = + m -> (String) m.getHeaders().get(MqttHeaders.TOPIC); + private final AtomicBoolean running = new AtomicBoolean(); private final String url; private final String clientId; - private volatile String defaultTopic; + private String defaultTopic; - private volatile int defaultQos = 0; + private MessageProcessor topicProcessor = DEFAULT_TOPIC_PROCESSOR; - private volatile boolean defaultRetained = false; + private int defaultQos = 0; - private volatile MessageConverter converter; + private MessageProcessor qosProcessor = MqttMessageConverter.defaultQosProcessor(); - private volatile int clientInstance; + private boolean defaultRetained; + + private MessageProcessor retainedProcessor = MqttMessageConverter.defaultRetainedProcessor(); + + private MessageConverter converter; + + private int clientInstance; public AbstractMqttMessageHandler(String url, String clientId) { Assert.hasText(clientId, "'clientId' cannot be null or empty"); @@ -59,18 +73,109 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler this.clientId = clientId; } + /** + * Set the topic to which the message will be published if the + * {@link #setTopicExpression(Expression) topicExpression} evaluates to `null`. + * @param defaultTopic the default topic. + */ public void setDefaultTopic(String defaultTopic) { this.defaultTopic = defaultTopic; } + /** + * Set the topic expression; default "headers['mqtt_topic']". + * @param topicExpression the expression. + * @since 5.0 + */ + public void setTopicExpression(Expression topicExpression) { + Assert.notNull(topicExpression, "'topicExpression' cannot be null"); + this.topicProcessor = new ExpressionEvaluatingMessageProcessor<>(topicExpression); + } + + /** + * Set the topic expression; default "headers['mqtt_topic']". + * @param topicExpression the expression. + * @since 5.0 + */ + public void setTopicExpressionString(String topicExpression) { + Assert.hasText(topicExpression, "'topicExpression' must not be null or empty"); + this.topicProcessor = new ExpressionEvaluatingMessageProcessor<>(topicExpression); + } + + /** + * Set the qos for messages if the {@link #setQosExpression(Expression) qosExpression} + * evaluates to null. Only applies if a message converter is not provided. + * @param defaultQos the default qos. + * @see #setConverter(MessageConverter) + */ public void setDefaultQos(int defaultQos) { this.defaultQos = defaultQos; } - public void setDefaultRetained(boolean defaultRetain) { - this.defaultRetained = defaultRetain; + /** + * Set the qos expression; default "headers['mqtt_qos']". + * Only applies if a message converter is not provided. + * @param qosExpression the expression. + * @see #setConverter(MessageConverter) + * @since 5.0 + */ + public void setQosExpression(Expression qosExpression) { + Assert.notNull(qosExpression, "'qosExpression' cannot be null"); + this.qosProcessor = new ExpressionEvaluatingMessageProcessor<>(qosExpression); } + /** + * Set the qos expression; default "headers['mqtt_qos']". + * Only applies if a message converter is not provided. + * @param qosExpression the expression. + * @see #setConverter(MessageConverter) + * @since 5.0 + */ + public void setQosExpressionString(String qosExpression) { + Assert.hasText(qosExpression, "'qosExpression' must not be null or empty"); + this.qosProcessor = new ExpressionEvaluatingMessageProcessor<>(qosExpression); + } + + /** + * Set the retained boolean for messages if the + * {@link #setRetainedExpression(Expression) retainedExpression} evaluates to null. + * Only applies if a message converter is not provided. + * @param defaultRetained the default defaultRetained. + * @see #setConverter(MessageConverter) + */ + public void setDefaultRetained(boolean defaultRetained) { + this.defaultRetained = defaultRetained; + } + + /** + * Set the retained expression; default "headers['mqtt_retained']". + * Only applies if a message converter is not provided. + * @param retainedExpression the expression. + * @see #setConverter(MessageConverter) + * @since 5.0 + */ + public void setRetainedExpression(Expression retainedExpression) { + Assert.notNull(retainedExpression, "'qosExpression' cannot be null"); + this.retainedProcessor = new ExpressionEvaluatingMessageProcessor<>(retainedExpression); + } + + /** + * Set the retained expression; default "headers['mqtt_retained']". + * Only applies if a message converter is not provided. + * @param retainedExpression the expression. + * @see #setConverter(MessageConverter) + * @since 5.0 + */ + public void setRetainedExpressionString(String retainedExpression) { + Assert.hasText(retainedExpression, "'qosExpression' must not be null or empty"); + this.retainedProcessor = new ExpressionEvaluatingMessageProcessor<>(retainedExpression); + } + + /** + * Set the message converter to use; if this is provided, the adapter qos and retained + * settings are ignored. + * @param converter the converter. + */ public void setConverter(MessageConverter converter) { Assert.notNull(converter, "'converter' cannot be null"); this.converter = converter; @@ -109,8 +214,22 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler @Override protected void onInit() throws Exception { super.onInit(); + if (this.topicProcessor instanceof BeanFactoryAware && getBeanFactory() != null) { + ((BeanFactoryAware) this.topicProcessor).setBeanFactory(getBeanFactory()); + } + if (this.qosProcessor instanceof BeanFactoryAware && getBeanFactory() != null) { + ((BeanFactoryAware) this.qosProcessor).setBeanFactory(getBeanFactory()); + } + if (this.retainedProcessor instanceof BeanFactoryAware && getBeanFactory() != null) { + ((BeanFactoryAware) this.retainedProcessor).setBeanFactory(getBeanFactory()); + } if (this.converter == null) { - this.converter = new DefaultPahoMessageConverter(this.defaultQos, this.defaultRetained); + DefaultPahoMessageConverter defaultConverter = new DefaultPahoMessageConverter(this.defaultQos, + this.qosProcessor, this.defaultRetained, this.retainedProcessor); + if (getBeanFactory() != null) { + defaultConverter.setBeanFactory(getBeanFactory()); + } + this.converter = defaultConverter; } } @@ -139,11 +258,11 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler @Override protected void handleMessageInternal(Message message) throws Exception { - String topic = (String) message.getHeaders().get(MqttHeaders.TOPIC); Object mqttMessage = this.converter.fromMessage(message, Object.class); + String topic = this.topicProcessor.processMessage(message); if (topic == null && this.defaultTopic == null) { throw new MessageHandlingException(message, - "No '" + MqttHeaders.TOPIC + "' header and no default topic defined"); + "No topic could be determined from the message and no default topic defined"); } this.publish(topic == null ? this.defaultTopic : topic, mqttMessage, message); } diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/DefaultPahoMessageConverter.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/DefaultPahoMessageConverter.java index 7dbf05f7a3..9603ef62b9 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/DefaultPahoMessageConverter.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/DefaultPahoMessageConverter.java @@ -20,6 +20,7 @@ import org.eclipse.paho.client.mqttv3.MqttMessage; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; +import org.springframework.integration.handler.MessageProcessor; import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.integration.support.DefaultMessageBuilderFactory; import org.springframework.integration.support.MessageBuilderFactory; @@ -42,9 +43,13 @@ public class DefaultPahoMessageConverter implements MqttMessageConverter, BeanFa private final String charset; - private final Integer defaultQos; + private final int defaultQos; - private final Boolean defaultRetained; + private final MessageProcessor qosProcessor; + + private final boolean defaultRetained; + + private final MessageProcessor retainedProcessor; private volatile boolean payloadAsBytes = false; @@ -63,14 +68,15 @@ public class DefaultPahoMessageConverter implements MqttMessageConverter, BeanFa } /** - * Construct a converter to create outbound messages with the supplied default qos and retain settings and - * a UTF-8 charset for converting outbound String payloads to {@code byte[]} and inbound - * {@code byte[]} to String (unless {@link #setPayloadAsBytes(boolean) payloadAdBytes} is true). + * Construct a converter to create outbound messages with the supplied default qos and + * retain settings and a UTF-8 charset for converting outbound String payloads to + * {@code byte[]} and inbound {@code byte[]} to String (unless + * {@link #setPayloadAsBytes(boolean) payloadAdBytes} is true). * @param defaultQos the default qos. - * @param defaultRetain the default retain. + * @param defaultRetained the default retained. */ - public DefaultPahoMessageConverter(int defaultQos, boolean defaultRetain) { - this(defaultQos, defaultRetain, "UTF-8"); + public DefaultPahoMessageConverter(int defaultQos, boolean defaultRetained) { + this(defaultQos, defaultRetained, "UTF-8"); } /** @@ -85,16 +91,55 @@ public class DefaultPahoMessageConverter implements MqttMessageConverter, BeanFa } /** - * Construct a converter to create outbound messages with the supplied default qos and retain settings and - * the supplied charset. + * Construct a converter to create outbound messages with the supplied default qos and + * retain settings and the supplied charset. * @param defaultQos the default qos. - * @param defaultRetained the default retain. - * @param charset the charset used to convert outbound String paylaods to {@code byte[]} and inbound - * {@code byte[]} to String (unless {@link #setPayloadAsBytes(boolean) payloadAdBytes} is true). + * @param defaultRetained the default retained. + * @param charset the charset used to convert outbound String paylaods to + * {@code byte[]} and inbound {@code byte[]} to String (unless + * {@link #setPayloadAsBytes(boolean) payloadAdBytes} is true). */ public DefaultPahoMessageConverter(int defaultQos, boolean defaultRetained, String charset) { + this(defaultQos, MqttMessageConverter.defaultQosProcessor(), defaultRetained, + MqttMessageConverter.defaultRetainedProcessor(), charset); + } + + /** + * Construct a converter to create outbound messages with the supplied default qos and + * retained message processors and a UTF-8 charset for converting outbound String payloads to + * {@code byte[]} and inbound {@code byte[]} to String (unless + * {@link #setPayloadAsBytes(boolean) payloadAdBytes} is true). + * @param defaultQos the default qos. + * @param qosProcessor a message processor to determine the qos. + * @param defaultRetained the default retained. + * @param retainedProcessor a message processor to determine the retained flag. + * @since 5.0 + */ + public DefaultPahoMessageConverter(int defaultQos, MessageProcessor qosProcessor, boolean defaultRetained, + MessageProcessor retainedProcessor) { + this(defaultQos, qosProcessor, defaultRetained, retainedProcessor, "UTF-8"); + } + + /** + * Construct a converter to create outbound messages with the supplied default qos and + * retain settings and the supplied charset. + * @param defaultQos the default qos. + * @param qosProcessor a message processor to determine the qos. + * @param defaultRetained the default retained. + * @param retainedProcessor a message processor to determine the retained flag. + * @param charset the charset used to convert outbound String paylaods to + * {@code byte[]} and inbound {@code byte[]} to String (unless + * {@link #setPayloadAsBytes(boolean) payloadAdBytes} is true). + * @since 5.0 + */ + public DefaultPahoMessageConverter(int defaultQos, MessageProcessor qosProcessor, boolean defaultRetained, + MessageProcessor retainedProcessor, String charset) { + Assert.notNull(qosProcessor, "'qosProcessor' cannot be null"); + Assert.notNull(retainedProcessor, "'retainedProcessor' cannot be null"); this.defaultQos = defaultQos; + this.qosProcessor = qosProcessor; this.defaultRetained = defaultRetained; + this.retainedProcessor = retainedProcessor; this.charset = charset; } @@ -141,11 +186,11 @@ public class DefaultPahoMessageConverter implements MqttMessageConverter, BeanFa try { AbstractIntegrationMessageBuilder messageBuilder = getMessageBuilderFactory() .withPayload(mqttBytesToPayload(mqttMessage)) - .setHeader(MqttHeaders.QOS, mqttMessage.getQos()) + .setHeader(MqttHeaders.RECEIVED_QOS, mqttMessage.getQos()) .setHeader(MqttHeaders.DUPLICATE, mqttMessage.isDuplicate()) - .setHeader(MqttHeaders.RETAINED, mqttMessage.isRetained()); + .setHeader(MqttHeaders.RECEIVED_RETAINED, mqttMessage.isRetained()); if (topic != null) { - messageBuilder.setHeader(MqttHeaders.TOPIC, topic); + messageBuilder.setHeader(MqttHeaders.RECEIVED_TOPIC, topic); } return messageBuilder.build(); } @@ -158,12 +203,10 @@ public class DefaultPahoMessageConverter implements MqttMessageConverter, BeanFa public MqttMessage fromMessage(Message message, Class targetClass) { byte[] payloadBytes = messageToMqttBytes(message); MqttMessage mqttMessage = new MqttMessage(payloadBytes); - Object header = message.getHeaders().get(MqttHeaders.RETAINED); - Assert.isTrue(header == null || header instanceof Boolean, MqttHeaders.RETAINED + " header must be Boolean"); - mqttMessage.setRetained(header == null ? this.defaultRetained : (Boolean) header); - header = message.getHeaders().get(MqttHeaders.QOS); - Assert.isTrue(header == null || header instanceof Integer, MqttHeaders.QOS + " header must be Integer"); - mqttMessage.setQos(header == null ? this.defaultQos : (Integer) header); + Integer qos = this.qosProcessor.processMessage(message); + mqttMessage.setQos(qos == null ? this.defaultQos : qos); + Boolean retained = this.retainedProcessor.processMessage(message); + mqttMessage.setRetained(retained == null ? this.defaultRetained : retained); return mqttMessage; } diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttHeaders.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttHeaders.java index c2f8856cf4..d1281929d6 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttHeaders.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttHeaders.java @@ -29,11 +29,18 @@ public final class MqttHeaders { public static final String QOS = prefix + "qos"; + public static final String RECEIVED_QOS = prefix + "receivedQos"; + public static final String DUPLICATE = prefix + "duplicate"; public static final String RETAINED = prefix + "retained"; + public static final String RECEIVED_RETAINED = prefix + "receivedRetained"; + public static final String TOPIC = prefix + "topic"; + + public static final String RECEIVED_TOPIC = prefix + "receivedTopic"; + private MqttHeaders() { throw new AssertionError(); } diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttMessageConverter.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttMessageConverter.java index 45377693bb..110aa278d0 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttMessageConverter.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttMessageConverter.java @@ -18,6 +18,7 @@ package org.springframework.integration.mqtt.support; import org.eclipse.paho.client.mqttv3.MqttMessage; +import org.springframework.integration.handler.MessageProcessor; import org.springframework.messaging.Message; import org.springframework.messaging.converter.MessageConverter; @@ -39,4 +40,13 @@ public interface MqttMessageConverter extends MessageConverter { * @return The Message. */ Message toMessage(String topic, MqttMessage mqttMessage); + + static MessageProcessor defaultQosProcessor() { + return message -> message.getHeaders().get(MqttHeaders.QOS, Integer.class); + } + + static MessageProcessor defaultRetainedProcessor() { + return message -> message.getHeaders().get(MqttHeaders.RETAINED, Boolean.class); + } + } diff --git a/spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/spring-integration-mqtt-5.0.xsd b/spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/spring-integration-mqtt-5.0.xsd index 7448452dc4..6837293d54 100644 --- a/spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/spring-integration-mqtt-5.0.xsd +++ b/spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/spring-integration-mqtt-5.0.xsd @@ -133,22 +133,48 @@ - Specifies the default topic to which messages will be sent. Required if an - outbound message does not have an 'mqtt_topic' header. + Specifies the default topic to which messages will be sent. Required if + the 'topic-expression' evaluates to 'null' + + + + + + + Specifies an expression to evaluate to determine the destination topic. + Default "headers['mqtt_topic']". - Specifies the default quality of service. Default 0. + Specifies the default quality of service; used if the 'qos-expression' + evaluates to 'null'. Default 0. + + + + + + + Specifies an expression to evaluate to determine the message qos. + Default "headers['mqtt_qos']". - Specifies the default value of the 'retained' flag. Default false. + Specifies the default value of the 'retained' flag; used if the + 'retained-expression' evaluates to 'null'. Default false. + + + + + + + Specifies an expression to evaluate to determine the message 'retained' + flag. Default "headers['mqtt_retained']". diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests.java b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests.java index 3774294fab..b8f4d87dca 100644 --- a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests.java +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests.java @@ -54,8 +54,7 @@ import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.GenericMessage; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.junit4.SpringRunner; /** * @author Gary Russell @@ -63,8 +62,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; * @since 4.0 * */ -@ContextConfiguration -@RunWith(SpringJUnit4ClassRunner.class) +@RunWith(SpringRunner.class) @DirtiesContext public class BackToBackAdapterTests { @@ -103,7 +101,7 @@ public class BackToBackAdapterTests { assertNotNull(out); inbound.stop(); assertEquals("foo", out.getPayload()); - assertEquals("mqtt-foo", out.getHeaders().get(MqttHeaders.TOPIC)); + assertEquals("mqtt-foo", out.getHeaders().get(MqttHeaders.RECEIVED_TOPIC)); } @Test @@ -124,21 +122,21 @@ public class BackToBackAdapterTests { inbound.start(); inbound.addTopic("mqtt-foo"); adapter.handleMessage(new GenericMessage("foo")); - Message out = outputChannel.receive(10000); + Message out = outputChannel.receive(10_000); assertNotNull(out); assertEquals("foo", out.getPayload()); - assertEquals("mqtt-foo", out.getHeaders().get(MqttHeaders.TOPIC)); + assertEquals("mqtt-foo", out.getHeaders().get(MqttHeaders.RECEIVED_TOPIC)); inbound.addTopic("mqtt-bar"); adapter.handleMessage(MessageBuilder.withPayload("bar").setHeader(MqttHeaders.TOPIC, "mqtt-bar").build()); - out = outputChannel.receive(10000); + out = outputChannel.receive(10_000); assertNotNull(out); assertEquals("bar", out.getPayload()); - assertEquals("mqtt-bar", out.getHeaders().get(MqttHeaders.TOPIC)); + assertEquals("mqtt-bar", out.getHeaders().get(MqttHeaders.RECEIVED_TOPIC)); inbound.removeTopic("mqtt-bar"); adapter.handleMessage(MessageBuilder.withPayload("bar").setHeader(MqttHeaders.TOPIC, "mqtt-bar").build()); - out = outputChannel.receive(10000); + out = outputChannel.receive(10_000); assertNull(out); try { @@ -183,12 +181,12 @@ public class BackToBackAdapterTests { assertNotNull(out); inbound.stop(); assertEquals("foo", out.getPayload()); - assertEquals("mqtt-foo", out.getHeaders().get(MqttHeaders.TOPIC)); + assertEquals("mqtt-foo", out.getHeaders().get(MqttHeaders.RECEIVED_TOPIC)); out = outputChannel.receive(10000); assertNotNull(out); inbound.stop(); assertEquals("bar", out.getPayload()); - assertEquals("mqtt-bar", out.getHeaders().get(MqttHeaders.TOPIC)); + assertEquals("mqtt-bar", out.getHeaders().get(MqttHeaders.RECEIVED_TOPIC)); } @Test @@ -220,7 +218,7 @@ public class BackToBackAdapterTests { assertNotNull(out); inbound.stop(); assertEquals("foo", out.getPayload()); - assertEquals("mqtt-foo", out.getHeaders().get(MqttHeaders.TOPIC)); + assertEquals("mqtt-foo", out.getHeaders().get(MqttHeaders.RECEIVED_TOPIC)); } @Test @@ -288,10 +286,10 @@ public class BackToBackAdapterTests { out = outputChannel.receive(10000); assertNotNull(out); if ("foo".equals(out.getPayload())) { - assertEquals("mqtt-foo", out.getHeaders().get(MqttHeaders.TOPIC)); + assertEquals("mqtt-foo", out.getHeaders().get(MqttHeaders.RECEIVED_TOPIC)); } else if ("bar".equals(out.getPayload())) { - assertEquals("mqtt-bar", out.getHeaders().get(MqttHeaders.TOPIC)); + assertEquals("mqtt-bar", out.getHeaders().get(MqttHeaders.RECEIVED_TOPIC)); } else { fail("unexpected payload " + out.getPayload()); diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/DownstreamExceptionTests.java b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/DownstreamExceptionTests.java index 318d0645c4..adf933663b 100644 --- a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/DownstreamExceptionTests.java +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/DownstreamExceptionTests.java @@ -45,16 +45,14 @@ import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.junit4.SpringRunner; /** * @author Gary Russell * @since 4.0 * */ -@ContextConfiguration -@RunWith(SpringJUnit4ClassRunner.class) +@RunWith(SpringRunner.class) @DirtiesContext public class DownstreamExceptionTests { diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java index df39ab9b7e..fbf89ee367 100644 --- a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java @@ -59,8 +59,12 @@ import org.junit.Test; import org.springframework.aop.framework.ProxyFactoryBean; import org.springframework.beans.factory.BeanFactory; import org.springframework.context.ApplicationEventPublisher; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; import org.springframework.integration.channel.NullChannel; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.handler.MessageProcessor; import org.springframework.integration.mqtt.core.ConsumerStopAction; import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory; import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory.Will; @@ -69,6 +73,8 @@ import org.springframework.integration.mqtt.event.MqttIntegrationEvent; import org.springframework.integration.mqtt.event.MqttSubscribedEvent; import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter; import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler; +import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter; +import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; @@ -337,6 +343,31 @@ public class MqttAdapterTests { verifyNotUnsubscribe(client); } + @SuppressWarnings("unchecked") + @Test + public void testCustomExpressions() { + AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(Config.class); + MqttPahoMessageHandler handler = ctx.getBean("handler", MqttPahoMessageHandler.class); + GenericMessage message = new GenericMessage<>("foo"); + assertEquals("fooTopic", + TestUtils.getPropertyValue(handler, "topicProcessor", MessageProcessor.class).processMessage(message)); + assertEquals(1, + TestUtils.getPropertyValue(handler, "converter.qosProcessor", MessageProcessor.class) + .processMessage(message)); + assertEquals(Boolean.TRUE, + TestUtils.getPropertyValue(handler, "converter.retainedProcessor", MessageProcessor.class) + .processMessage(message)); + + handler = ctx.getBean("handlerWithNullExpressions", MqttPahoMessageHandler.class); + assertEquals(1, + TestUtils.getPropertyValue(handler, "converter", DefaultPahoMessageConverter.class) + .fromMessage(message, null).getQos()); + assertEquals(Boolean.TRUE, + TestUtils.getPropertyValue(handler, "converter", DefaultPahoMessageConverter.class) + .fromMessage(message, null).isRetained()); + ctx.close(); + } + private MqttPahoMessageDrivenChannelAdapter buildAdapter(final MqttAsyncClient client, Boolean cleanSession, ConsumerStopAction action) throws MqttException, MqttSecurityException { DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory() { @@ -380,4 +411,43 @@ public class MqttAdapterTests { verify(client).disconnect(); } + @Configuration + public static class Config { + + @Bean + public MqttPahoMessageHandler handler() { + MqttPahoMessageHandler handler = new MqttPahoMessageHandler("tcp://localhost:1883", "bar"); + handler.setTopicExpressionString("@topic"); + handler.setQosExpressionString("@qos"); + handler.setRetainedExpressionString("@retained"); + return handler; + } + + @Bean + public String topic() { + return "fooTopic"; + } + + @Bean + public Integer qos() { + return 1; + } + + @Bean + public Boolean retained() { + return true; + } + + @Bean + public MqttPahoMessageHandler handlerWithNullExpressions() { + MqttPahoMessageHandler handler = new MqttPahoMessageHandler("tcp://localhost:1883", "bar"); + handler.setDefaultQos(1); + handler.setQosExpressionString("null"); + handler.setDefaultRetained(true); + handler.setRetainedExpressionString("null"); + return handler; + } + + } + } diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParserTests.java b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParserTests.java index a2289175e4..e7b1d01818 100644 --- a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParserTests.java +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParserTests.java @@ -32,8 +32,8 @@ import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannel import org.springframework.integration.mqtt.support.MqttMessageConverter; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.MessageChannel; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; /** * @author Gary Russell @@ -41,8 +41,8 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; * @since 4.0 * */ -@ContextConfiguration -@RunWith(SpringJUnit4ClassRunner.class) +@RunWith(SpringRunner.class) +@DirtiesContext public class MqttMessageDrivenChannelAdapterParserTests { @Autowired diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParserTests-context.xml b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParserTests-context.xml index a89caa1475..b4112861b0 100644 --- a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParserTests-context.xml +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParserTests-context.xml @@ -17,6 +17,9 @@ converter="myConverter" client-factory="clientFactory" default-topic="bar" + topic-expression="'bar'" + qos-expression="2" + retained-expression="true" phase="25" order="1" channel="target"> diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParserTests.java b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParserTests.java index 89d494825c..9f44391cb0 100644 --- a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParserTests.java +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParserTests.java @@ -31,6 +31,7 @@ import org.springframework.aop.support.AopUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.integration.endpoint.EventDrivenConsumer; +import org.springframework.integration.handler.MessageProcessor; import org.springframework.integration.handler.advice.RequestHandlerRetryAdvice; import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory; import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler; @@ -38,16 +39,17 @@ import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter; import org.springframework.integration.mqtt.support.MqttMessageConverter; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.MessageHandler; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; /** * @author Gary Russell * @author Artem Bilan * @since 4.0 */ -@ContextConfiguration -@RunWith(SpringJUnit4ClassRunner.class) +@RunWith(SpringRunner.class) +@DirtiesContext public class MqttOutboundChannelAdapterParserTests { @Autowired @Qualifier("withConverter") @@ -65,11 +67,21 @@ public class MqttOutboundChannelAdapterParserTests { @Autowired private DefaultMqttPahoClientFactory clientFactory; + @SuppressWarnings("unchecked") @Test public void testWithConverter() throws Exception { assertEquals("tcp://localhost:1883", TestUtils.getPropertyValue(withConverterHandler, "url")); assertEquals("foo", TestUtils.getPropertyValue(withConverterHandler, "clientId")); assertEquals("bar", TestUtils.getPropertyValue(withConverterHandler, "defaultTopic")); + GenericMessage message = new GenericMessage<>("foo"); + assertEquals("bar", + TestUtils.getPropertyValue(withConverterHandler, "topicProcessor", MessageProcessor.class) + .processMessage(message)); + assertEquals(2, TestUtils.getPropertyValue(withConverterHandler, "qosProcessor", MessageProcessor.class) + .processMessage(message)); + assertEquals(Boolean.TRUE, + TestUtils.getPropertyValue(withConverterHandler, "retainedProcessor", MessageProcessor.class) + .processMessage(message)); assertSame(converter, TestUtils.getPropertyValue(withConverterHandler, "converter")); assertSame(clientFactory, TestUtils.getPropertyValue(withConverterHandler, "clientFactory")); assertFalse(TestUtils.getPropertyValue(withConverterHandler, "async", Boolean.class)); @@ -87,16 +99,17 @@ public class MqttOutboundChannelAdapterParserTests { @Test public void testWithDefaultConverter() { + GenericMessage message = new GenericMessage<>("foo"); assertEquals("tcp://localhost:1883", TestUtils.getPropertyValue(withDefaultConverterHandler, "url")); assertEquals("foo", TestUtils.getPropertyValue(withDefaultConverterHandler, "clientId")); assertEquals("bar", TestUtils.getPropertyValue(withDefaultConverterHandler, "defaultTopic")); assertEquals(1, TestUtils.getPropertyValue(withDefaultConverterHandler, "defaultQos")); - assertTrue(TestUtils.getPropertyValue(withDefaultConverterHandler, "defaultRetained", Boolean.class)); - MqttMessageConverter defaultConverter = TestUtils.getPropertyValue(withDefaultConverterHandler, "converter", - MqttMessageConverter.class); - assertTrue(defaultConverter instanceof DefaultPahoMessageConverter); - assertEquals(1, TestUtils.getPropertyValue(defaultConverter, "defaultQos")); - assertTrue(TestUtils.getPropertyValue(defaultConverter, "defaultRetained", Boolean.class)); + assertEquals(Boolean.TRUE, + TestUtils.getPropertyValue(withDefaultConverterHandler, "defaultRetained", Boolean.class)); + DefaultPahoMessageConverter defaultConverter = TestUtils.getPropertyValue(withDefaultConverterHandler, + "converter", DefaultPahoMessageConverter.class); + assertEquals(1, defaultConverter.fromMessage(message, null).getQos()); + assertTrue(defaultConverter.fromMessage(message, null).isRetained()); assertSame(clientFactory, TestUtils.getPropertyValue(withDefaultConverterHandler, "clientFactory")); assertTrue(TestUtils.getPropertyValue(withDefaultConverterHandler, "async", Boolean.class)); assertTrue(TestUtils.getPropertyValue(withDefaultConverterHandler, "asyncEvents", Boolean.class)); diff --git a/src/reference/asciidoc/ftp.adoc b/src/reference/asciidoc/ftp.adoc index ec36d1eb4f..4cd3864274 100644 --- a/src/reference/asciidoc/ftp.adoc +++ b/src/reference/asciidoc/ftp.adoc @@ -802,7 +802,7 @@ Typically, you would use the `#remoteDirectory` variable in the `local-directory Starting with _version 5.0_, the `FtpSimplePatternFileListFilter` and `FtpRegexPatternFileListFilter` can be configured to always pass directories by setting the `alwaysAcceptDirectorties` to `true`. This allows recursion for a simple pattern; examples follow: -[code, xml] +[source, xml] ---- diff --git a/src/reference/asciidoc/mqtt.adoc b/src/reference/asciidoc/mqtt.adoc index ec87df5e7e..31c40fffd2 100644 --- a/src/reference/asciidoc/mqtt.adoc +++ b/src/reference/asciidoc/mqtt.adoc @@ -20,17 +20,17 @@ A minimal configuration might be: [source,xml] ---- - - + class="org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory"> + + + client-id="${mqtt.default.client.id}.src" + url="${mqtt.url}" + topics="sometopic" + client-factory="clientFactory" + channel="output"/> ---- Attributes: @@ -38,16 +38,16 @@ Attributes: [source,xml] ---- - url="tcp://localhost:1883" <2> - topics="bar,baz" <3> - qos="1,2" <4> - converter="myConverter" <5> - client-factory="clientFactory" <6> - send-timeout="123" <7> - error-channel="errors" <8> - recovery-interval="10000" <9> - channel="out" /> + client-id="foo" <1> + url="tcp://localhost:1883" <2> + topics="bar,baz" <3> + qos="1,2" <4> + converter="myConverter" <5> + client-factory="clientFactory" <6> + send-timeout="123" <7> + error-channel="errors" <8> + recovery-interval="10000" <9> + channel="out" /> ---- <1> The client id. @@ -109,6 +109,11 @@ The latter (the default) will unsubscribe only if the `cleanSession` property is To revert to the pre-4.2.3 behavior, use `UNSUBSCRIBE_ALWAYS`. ==== +[IMPORTANT] +==== +Starting with _version 5.0_, the `topic`, `qos` and `retained` properties are mapped to `.RECEIVED_...` headers (`MqttHeaders.RECEIVED_TOPIC`, `MqttHeaders.RECEIVED_QOS`, and `MqttHeaders.RECEIVED_RETAINED`), to avoid inadvertent propagation to an outbound message which (by default) uses the `MqttHeaders.TOPIC`, `MqttHeaders.QOS`, and `MqttHeaders.RETAINED` headers. +==== + ==== Adding/Removing Topics at Runtime Starting with _version 4.1_, it is possible to programmatically change the topics to which the adapter is subscribed. @@ -130,39 +135,39 @@ The following Spring Boot application provides an example of configuring the inb public class MqttJavaApplication { public static void main(String[] args) { - new SpringApplicationBuilder(MqttJavaApplication.class) - .web(false) - .run(args); + new SpringApplicationBuilder(MqttJavaApplication.class) + .web(false) + .run(args); } @Bean public MessageChannel mqttInputChannel() { - return new DirectChannel(); + return new DirectChannel(); } @Bean public MessageProducer inbound() { - MqttPahoMessageDrivenChannelAdapter adapter = - new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883", "testClient", - "topic1", "topic2"); - adapter.setCompletionTimeout(5000); - adapter.setConverter(new DefaultPahoMessageConverter()); - adapter.setQos(1); - adapter.setOutputChannel(mqttInputChannel()); - return adapter; + MqttPahoMessageDrivenChannelAdapter adapter = + new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883", "testClient", + "topic1", "topic2"); + adapter.setCompletionTimeout(5000); + adapter.setConverter(new DefaultPahoMessageConverter()); + adapter.setQos(1); + adapter.setOutputChannel(mqttInputChannel()); + return adapter; } @Bean @ServiceActivator(inputChannel = "mqttInputChannel") public MessageHandler handler() { - return new MessageHandler() { + return new MessageHandler() { - @Override - public void handleMessage(Message message) throws MessagingException { - System.out.println(message.getPayload()); - } + @Override + public void handleMessage(Message message) throws MessagingException { + System.out.println(message.getPayload()); + } - }; + }; } } @@ -182,16 +187,19 @@ Attributes: [source,xml] ---- - url="tcp://localhost:1883" <2> - converter="myConverter" <3> - client-factory="clientFactory" <4> - default-qos="1" <5> - default-retained="true" <6> - default-topic="bar" <7> - async="false" <8> - async-events="false" <9> - channel="target" /> + client-id="foo" <1> + url="tcp://localhost:1883" <2> + converter="myConverter" <3> + client-factory="clientFactory" <4> + default-qos="1" <5> + qos-expression="" <6> + default-retained="true" <7> + retained-expression="" <8> + default-topic="bar" <9> + topic-expression="" <10> + async="false" <11> + async-events="false" <12> + channel="target" /> ---- <1> The client id. @@ -210,22 +218,25 @@ The default `DefaultPahoMessageConverter` recognizes the following headers: + <4> The client factory. -<5> The default quality of service (used if no `mqtt_qos` header is found). -Not allowed if a custom `converter` is supplied. +<5> The default quality of service (used if no `mqtt_qos` header is found or the `qos-expression` returns `null`. +Not used if a custom `converter` is supplied. +<6> An expression to evaluate to determine the qos; default `headers[mqtt_qos]`. -<6> The default value of the retained flag (used if no `mqtt_retained` header is found). -Not allowed if a custom `converter` is supplied. +<7> The default value of the retained flag (used if no `mqtt_retained` header is found). +Not used if a custom `converter` is supplied. +<8> An expression to evaluate to determine the retained boolean; default `headers[mqtt_retained]`. -<7> The default topic to which the message will be sent (used if no `mqtt_topic` header is found). +<9> The default topic to which the message will be sent (used if no `mqtt_topic` header is found). +<10> An expression to evaluate to determine the destination topic; default `headers['topic']`. -<8> When `true`, the caller will not block waiting for delivery confirmation when a message is sent. +<11> When `true`, the caller will not block waiting for delivery confirmation when a message is sent. Default:false (the send blocks until delivery is confirmed). -<9> When `async` and `async-events` are both `true`, an `MqttMessageSentEvent` is emitted, containing the message, the topic, the `messageId` generated by the client library, the `clientId` and the `clientInstance` (incremented each time the client is connected). +<12> When `async` and `async-events` are both `true`, an `MqttMessageSentEvent` is emitted, containing the message, the topic, the `messageId` generated by the client library, the `clientId` and the `clientInstance` (incremented each time the client is connected). When the delivery is confirmed by the client library, an `MqttMessageDeliveredEvent` is emitted, containing the the `messageId`, `clientId` and the `clientInstance`, enabling delivery to be correlated with the send. These events can be received by any `ApplicationListener`, or by an event inbound channel adapter. Note that it is possible that the `MqttMessageDeliveredEvent` might be received before the `MqttMessageSentEvent`. @@ -245,9 +256,9 @@ public class MqttJavaApplication { public static void main(String[] args) { ConfigurableApplicationContext context = - new SpringApplicationBuilder(MqttJavaApplication.class) - .web(false) - .run(args); + new SpringApplicationBuilder(MqttJavaApplication.class) + .web(false) + .run(args); MyGateway gateway = context.getBean(MyGateway.class); gateway.sendToMqtt("foo"); } diff --git a/src/reference/asciidoc/sftp.adoc b/src/reference/asciidoc/sftp.adoc index e6bb0d3d60..2508982699 100644 --- a/src/reference/asciidoc/sftp.adoc +++ b/src/reference/asciidoc/sftp.adoc @@ -891,7 +891,7 @@ Typically, you would use the `#remoteDirectory` variable in the `local-directory Starting with _version 5.0_, the `SftpSimplePatternFileListFilter` and `SftpRegexPatternFileListFilter` can be configured to always pass directories by setting the `alwaysAcceptDirectorties` to `true`. This allows recursion for a simple pattern; examples follow: -[code, xml] +[source, xml] ---- diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 28787d5fd9..ab948b2fb0 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -92,3 +92,11 @@ See <> for more information. Aggregators now use a `SimpleSequenceSizeReleaseStrategy` by default, which is more efficient, especially with large groups. Empty groups are now scheduled for removal after `empty-group-min-timeout`. See <> for more information. + +==== MQTT Changes + +Inbound messages are now mapped with headers `RECEIVED_TOPIC`, `RECEIVED_QOS` and `RECEIVED_RETAINED` to avoid inadvertent propagation to outbound messages when an application is relaying messages. + +The outbound channel adapter now supports expressions for the topic, qos and retained properties; the defaults remain the same. + +See <> for more information.