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

GH-3454: From MQTT conversion error - to error ch (#3456)

Browse Source 
* GH-3454: From MQTT conversion error - to error ch

Fixes https://github.com/spring-projects/spring-integration/issues/3454

The message converter may return null when we try to covert from the
MQTT message.
The thrown exception may also reset the client connect.

* Fix `MqttPahoMessageDrivenChannelAdapter` to catch any conversion errors
(including `null` result) and try to send an `ErrorMessage` with that info
into the provided `errorChannel`.
Otherwise re-throw it as as

**Cherry-pick to `5.4.x` & `5.3.x`**

* * Apply review language-specific changes

# Conflicts:
#	spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java
#	spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java
#	src/reference/asciidoc/mqtt.adoc
This commit is contained in:
Artem Bilan
2021-01-08 11:29:59 -05:00
parent b5ef2f6f5a
commit 71d41757cd
3 changed files with 96 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2020 the original author or authors.
* Copyright 2002-2021 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -37,6 +37,8 @@ import org.springframework.integration.mqtt.event.MqttConnectionFailedEvent;
import org.springframework.integration.mqtt.event.MqttSubscribedEvent;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.converter.MessageConversionException;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.util.Assert;
/**
@@ -169,7 +171,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
}
catch (Exception e) {
logger.error("Exception while connecting and subscribing, retrying", e);
this.scheduleReconnect();
scheduleReconnect();
}
}
@@ -181,7 +183,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
try {
if (this.consumerStopAction.equals(ConsumerStopAction.UNSUBSCRIBE_ALWAYS)
|| (this.consumerStopAction.equals(ConsumerStopAction.UNSUBSCRIBE_CLEAN)
&& this.cleanSession)) {
&& this.cleanSession)) {
this.client.unsubscribe(getTopic());
}
@@ -237,7 +239,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
super.removeTopic(topic);
}
catch (MqttException e) {
throw new MessagingException("Failed to unsubscribe from topic " + Arrays.asList(topic), e);
throw new MessagingException("Failed to unsubscribe from topic(s) " + Arrays.toString(topic), e);
}
finally {
this.topicLock.unlock();
@@ -365,14 +367,47 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
@Override
public void messageArrived(String topic, MqttMessage mqttMessage) {
Message<?> message = this.getConverter().toMessage(topic, mqttMessage);
Message<?> message = toMessage(topic, mqttMessage);
if (message != null) {
try {
sendMessage(message);
}
catch (RuntimeException e) {
logger.error("Unhandled exception for " + message.toString(), e);
throw e;
}
}
}
private Message<?> toMessage(String topic, MqttMessage mqttMessage) {
Message<?> message = null;
RuntimeException conversionError = null;
try {
sendMessage(message);
message = getConverter().toMessage(topic, mqttMessage);
}
catch (RuntimeException e) {
logger.error("Unhandled exception for " + message.toString(), e);
throw e;
catch (RuntimeException ex) {
conversionError = ex;
}
if (message == null && conversionError == null) {
conversionError = new IllegalStateException("'MqttMessageConverter' returned 'null'");
}
if (conversionError != null) {
GenericMessage<MqttMessage> failedMessage = new GenericMessage<>(mqttMessage);
if (!sendErrorMessageIfNecessary(failedMessage, conversionError)) {
MessageConversionException conversionException;
if (conversionError instanceof MessageConversionException) {
conversionException = (MessageConversionException) conversionError;
}
else {
conversionException = new MessageConversionException(failedMessage,
"Failed to convert from MQTT Message", conversionError);
}
throw conversionException;
}
}
return message;
}
@Override

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2020 the original author or authors.
* Copyright 2002-2021 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -85,9 +85,12 @@ 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.mqtt.support.MqttMessageConverter;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
@@ -102,7 +105,7 @@ import org.springframework.util.ReflectionUtils;
*/
public class MqttAdapterTests {
private IMqttToken alwaysComplete;
private final IMqttToken alwaysComplete;
{
ProxyFactoryBean pfb = new ProxyFactoryBean();
@@ -194,12 +197,12 @@ public class MqttAdapterTests {
return deliveryToken;
}).given(client).publish(anyString(), any(MqttMessage.class));
handler.handleMessage(new GenericMessage<String>("Hello, world!"));
handler.handleMessage(new GenericMessage<>("Hello, world!"));
verify(client, times(1)).connect(any(MqttConnectOptions.class));
assertThat(connectCalled.get()).isTrue();
AtomicReference<Object> failed = new AtomicReference<>();
handler.setApplicationEventPublisher(event -> failed.set(event));
handler.setApplicationEventPublisher(failed::set);
handler.connectionLost(new IllegalStateException());
assertThat(failed.get()).isInstanceOf(MqttConnectionFailedEvent.class);
handler.stop();
@@ -254,7 +257,7 @@ public class MqttAdapterTests {
return null;
}).given(client).connect(any(MqttConnectOptions.class));
final AtomicReference<MqttCallback> callback = new AtomicReference<MqttCallback>();
final AtomicReference<MqttCallback> callback = new AtomicReference<>();
willAnswer(invocation -> {
callback.set(invocation.getArgument(0));
return null;
@@ -266,12 +269,14 @@ public class MqttAdapterTests {
"baz", "fix");
QueueChannel outputChannel = new QueueChannel();
adapter.setOutputChannel(outputChannel);
QueueChannel errorChannel = new QueueChannel();
adapter.setErrorChannel(errorChannel);
ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
taskScheduler.initialize();
adapter.setTaskScheduler(taskScheduler);
adapter.setBeanFactory(mock(BeanFactory.class));
ApplicationEventPublisher applicationEventPublisher = mock(ApplicationEventPublisher.class);
final BlockingQueue<MqttIntegrationEvent> events = new LinkedBlockingQueue<MqttIntegrationEvent>();
final BlockingQueue<MqttIntegrationEvent> events = new LinkedBlockingQueue<>();
willAnswer(invocation -> {
events.add(invocation.getArgument(0));
return null;
@@ -294,6 +299,33 @@ public class MqttAdapterTests {
assertThat(event).isInstanceOf(MqttSubscribedEvent.class);
assertThat(((MqttSubscribedEvent) event).getMessage()).isEqualTo("Connected and subscribed to [baz, fix]");
adapter.setConverter(new MqttMessageConverter() {
@Override public Message<?> toMessage(String topic, MqttMessage mqttMessage) {
return null;
}
@Override public Object fromMessage(Message<?> message, Class<?> targetClass) {
return null;
}
@Override public Message<?> toMessage(Object payload, MessageHeaders headers) {
return null;
}
});
callback.get().messageArrived("baz", message);
ErrorMessage errorMessage = (ErrorMessage) errorChannel.receive(0);
assertThat(errorMessage).isNotNull()
.extracting(ErrorMessage::getPayload)
.isInstanceOf(IllegalStateException.class);
IllegalStateException exception = (IllegalStateException) errorMessage.getPayload();
assertThat(exception).hasMessage("'MqttMessageConverter' returned 'null'");
assertThat(errorMessage.getOriginalMessage().getPayload()).isSameAs(message);
// lose connection and make first reconnect fail
failConnection.set(true);
RuntimeException e = new RuntimeException("foo");
@@ -416,7 +448,7 @@ public class MqttAdapterTests {
// the following assertion should be equalTo, but leq to protect against a slow CI server
assertThat(attemptingReconnectCount.get()).isLessThanOrEqualTo(2);
AtomicReference<Object> failed = new AtomicReference<>();
adapter.setApplicationEventPublisher(event -> failed.set(event));
adapter.setApplicationEventPublisher(failed::set);
adapter.connectionLost(new IllegalStateException());
assertThat(failed.get()).isInstanceOf(MqttConnectionFailedEvent.class);
adapter.stop();
@@ -448,12 +480,11 @@ public class MqttAdapterTests {
new DirectFieldAccessor(client).setPropertyValue("aClient", aClient);
willAnswer(new CallsRealMethods()).given(client).connect(any(MqttConnectOptions.class));
willAnswer(new CallsRealMethods()).given(client).subscribe(any(String[].class), any(int[].class));
willAnswer(new CallsRealMethods()).given(client).subscribe(any(String[].class), any(int[].class),
(IMqttMessageListener[]) isNull());
willAnswer(new CallsRealMethods()).given(client).subscribe(any(String[].class), any(int[].class), isNull());
willReturn(alwaysComplete).given(aClient).connect(any(MqttConnectOptions.class), any(), any());
IMqttToken token = mock(IMqttToken.class);
given(token.getGrantedQos()).willReturn(new int[] { 0x80 });
given(token.getGrantedQos()).willReturn(new int[]{ 0x80 });
willReturn(token).given(aClient).subscribe(any(String[].class), any(int[].class), isNull(), isNull(), any());
MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("foo", "bar", factory,
@@ -465,11 +496,11 @@ public class MqttAdapterTests {
}, m -> m.getName().equals("connectAndSubscribe"));
assertThat(method.get()).isNotNull();
Condition<InvocationTargetException> subscribeFailed = new Condition<>(ex ->
((MqttException) ex.getCause()).getReasonCode() == MqttException.REASON_CODE_SUBSCRIBE_FAILED,
"expected the reason code to be REASON_CODE_SUBSCRIBE_FAILED");
((MqttException) ex.getCause()).getReasonCode() == MqttException.REASON_CODE_SUBSCRIBE_FAILED,
"expected the reason code to be REASON_CODE_SUBSCRIBE_FAILED");
assertThatExceptionOfType(InvocationTargetException.class).isThrownBy(() -> method.get().invoke(adapter))
.withCauseInstanceOf(MqttException.class)
.is(subscribeFailed);
.withCauseInstanceOf(MqttException.class)
.is(subscribeFailed);
}
@Test
@@ -502,7 +533,7 @@ public class MqttAdapterTests {
willReturn(alwaysComplete).given(aClient).connect(any(MqttConnectOptions.class), any(), any());
IMqttToken token = mock(IMqttToken.class);
given(token.getGrantedQos()).willReturn(new int[] { 2, 0 });
given(token.getGrantedQos()).willReturn(new int[]{ 2, 0 });
willReturn(token).given(aClient).subscribe(any(String[].class), any(int[].class), isNull(), isNull(), any());
MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("foo", "bar", factory,
@@ -537,7 +568,7 @@ public class MqttAdapterTests {
};
MqttConnectOptions connectOptions = new MqttConnectOptions();
connectOptions.setServerURIs(new String[] { "tcp://localhost:1883" });
connectOptions.setServerURIs(new String[]{ "tcp://localhost:1883" });
if (cleanSession != null) {
connectOptions.setCleanSession(cleanSession);
}
@@ -554,7 +585,7 @@ public class MqttAdapterTests {
return adapter;
}
private MqttPahoMessageHandler buildAdapterOut(final IMqttAsyncClient client) throws MqttException {
private MqttPahoMessageHandler buildAdapterOut(final IMqttAsyncClient client) {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory() {
@Override
@@ -564,7 +595,7 @@ public class MqttAdapterTests {
};
MqttConnectOptions connectOptions = new MqttConnectOptions();
connectOptions.setServerURIs(new String[] { "tcp://localhost:1883" });
connectOptions.setServerURIs(new String[]{ "tcp://localhost:1883" });
factory.setConnectionOptions(connectOptions);
MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("client", factory);
adapter.setDefaultTopic("foo");

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

@@ -145,6 +145,9 @@ A new application context reverts to the configured settings.
Changing the topics while the adapter is stopped (or disconnected from the broker) takes effect the next time a connection is established.
Starting with version `5.2.11`, when the message converter throws an exception or returns `null` from the `MqttMessage` conversion, the `MqttPahoMessageDrivenChannelAdapter` sends an `ErrorMessage` into the `errorChannel`, if provided.
Re-throws this conversion error otherwise into an MQTT client callback.
==== Configuring with Java Configuration
The following Spring Boot application shows an example of how to configure the inbound adapter with Java configuration: