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

INT-4138: MQTT: Outbound Adapter Improvements

Browse Source 
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
This commit is contained in:
Gary Russell
2016-11-14 16:04:17 -05:00
committed by Artem Bilan
parent d42357ba02
commit 3035bc716d
17 changed files with 471 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
spring-integration-core/src/main/java/org/springframework/integration/handler/ExpressionEvaluatingMessageProcessor.java
View File

@@ -27,6 +27,7 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
* @author Artem Bilan
* @author Gary Russell
* @since 2.0
*/
public class ExpressionEvaluatingMessageProcessor<T> extends AbstractMessageProcessor<T> {
@@ -61,6 +62,38 @@ public class ExpressionEvaluatingMessageProcessor<T> 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<T> 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.

5
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParser.java
View File

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

139
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java
View File

@@ -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<String> 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<String> topicProcessor = DEFAULT_TOPIC_PROCESSOR;
private volatile boolean defaultRetained = false;
private int defaultQos = 0;
private volatile MessageConverter converter;
private MessageProcessor<Integer> qosProcessor = MqttMessageConverter.defaultQosProcessor();
private volatile int clientInstance;
private boolean defaultRetained;
private MessageProcessor<Boolean> 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);
}

87
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/DefaultPahoMessageConverter.java
View File

@@ -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<Integer> qosProcessor;
private final boolean defaultRetained;
private final MessageProcessor<Boolean> 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<Integer> qosProcessor, boolean defaultRetained,
MessageProcessor<Boolean> 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<Integer> qosProcessor, boolean defaultRetained,
MessageProcessor<Boolean> 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<Object> 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;
}

7
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttHeaders.java
View File

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

10
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttMessageConverter.java
View File

@@ -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<Integer> defaultQosProcessor() {
return message -> message.getHeaders().get(MqttHeaders.QOS, Integer.class);
}
static MessageProcessor<Boolean> defaultRetainedProcessor() {
return message -> message.getHeaders().get(MqttHeaders.RETAINED, Boolean.class);
}
}

34
spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/spring-integration-mqtt-5.0.xsd
View File

@@ -133,22 +133,48 @@
<xsd:attribute name="default-topic">
<xsd:annotation>
<xsd:documentation>
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'
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="topic-expression">
<xsd:annotation>
<xsd:documentation>
Specifies an expression to evaluate to determine the destination topic.
Default "headers['mqtt_topic']".
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="default-qos">
<xsd:annotation>
<xsd:documentation>
Specifies the default quality of service. Default 0.
Specifies the default quality of service; used if the 'qos-expression'
evaluates to 'null'. Default 0.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="qos-expression">
<xsd:annotation>
<xsd:documentation>
Specifies an expression to evaluate to determine the message qos.
Default "headers['mqtt_qos']".
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="default-retained">
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="retained-expression">
<xsd:annotation>
<xsd:documentation>
Specifies an expression to evaluate to determine the message 'retained'
flag. Default "headers['mqtt_retained']".
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>

28
spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests.java
View File

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

6
spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/DownstreamExceptionTests.java
View File

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

70
spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java
View File

@@ -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<String> 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;
}
}
}

8
spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParserTests.java
View File

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

3
spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParserTests-context.xml
View File

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

33
spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParserTests.java
View File

@@ -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<String> 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<String> 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));

2
src/reference/asciidoc/ftp.adoc
View File

@@ -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]
----
<bean id="startDotTxtFilter" class="org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter">
<constructor-arg value="*.txt" />

123
src/reference/asciidoc/mqtt.adoc
View File

@@ -20,17 +20,17 @@ A minimal configuration might be:
[source,xml]
----
<bean id="clientFactory"
class="org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory">
<property name="userName" value="${mqtt.username}"/>
<property name="password" value="${mqtt.password}"/>
class="org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory">
<property name="userName" value="${mqtt.username}"/>
<property name="password" value="${mqtt.password}"/>
</bean>
<int-mqtt:message-driven-channel-adapter id="mqttInbound"
client-id="${mqtt.default.client.id}.src"
url="${mqtt.url}"
topics="sometopic"
client-factory="clientFactory"
channel="output"/>
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]
----
<int-mqtt:message-driven-channel-adapter id="oneTopicAdapter"
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" />
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]
----
<int-mqtt:outbound-channel-adapter id="withConverter"
client-id="foo" <1>
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");
}

2
src/reference/asciidoc/sftp.adoc
View File

@@ -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]
----
<bean id="startDotTxtFilter" class="org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter">
<constructor-arg value="*.txt" />

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

@@ -92,3 +92,11 @@ See <<http-header-mapping>> 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 <<aggregator>> 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 <<mqtt>> for more information.