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
This commit is contained in:
Artem Bilan
2021-01-08 11:29:59 -05:00
committed by Gary Russell
parent ebe79aedfb
commit 732a1b87e1
3 changed files with 99 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
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.
@@ -42,6 +42,8 @@ import org.springframework.integration.mqtt.support.MqttUtils;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
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;
/**
@@ -205,7 +207,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
}
catch (Exception ex) {
logger.error(ex, "Exception while connecting and subscribing, retrying");
this.scheduleReconnect();
scheduleReconnect();
}
}
@@ -272,7 +274,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();
@@ -399,19 +401,52 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
@Override
public void messageArrived(String topic, MqttMessage mqttMessage) {
AbstractIntegrationMessageBuilder<?> builder = getConverter().toMessageBuilder(topic, mqttMessage);
if (this.manualAcks) {
builder.setHeader(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK,
new AcknowledgmentImpl(mqttMessage.getId(), mqttMessage.getQos(), this.client));
AbstractIntegrationMessageBuilder<?> builder = toMessageBuilder(topic, mqttMessage);
if (builder != null) {
if (this.manualAcks) {
builder.setHeader(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK,
new AcknowledgmentImpl(mqttMessage.getId(), mqttMessage.getQos(), this.client));
}
Message<?> message = builder.build();
try {
sendMessage(message);
}
catch (RuntimeException ex) {
logger.error(ex, () -> "Unhandled exception for " + message.toString());
throw ex;
}
}
Message<?> message = builder.build();
}
private AbstractIntegrationMessageBuilder<?> toMessageBuilder(String topic, MqttMessage mqttMessage) {
AbstractIntegrationMessageBuilder<?> builder = null;
Exception conversionError = null;
try {
sendMessage(message);
builder = getConverter().toMessageBuilder(topic, mqttMessage);
}
catch (RuntimeException ex) {
logger.error(ex, () -> "Unhandled exception for " + message.toString());
throw ex;
catch (Exception ex) {
conversionError = ex;
}
if (builder == null && conversionError == null) {
conversionError = new IllegalStateException("'MqttMessageConverter' returned 'null'");
}
if (conversionError != null) {
GenericMessage<MqttMessage> message = new GenericMessage<>(mqttMessage);
if (!sendErrorMessageIfNecessary(message, conversionError)) {
MessageConversionException conversionException;
if (conversionError instanceof MessageConversionException) {
conversionException = (MessageConversionException) conversionError;
}
else {
conversionException = new MessageConversionException(message, "Failed to convert from MQTT Message",
conversionError);
}
throw conversionException;
}
}
return builder;
}
@Override

58
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.
@@ -89,9 +89,13 @@ import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannel
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.integration.mqtt.support.MqttHeaderAccessor;
import org.springframework.integration.mqtt.support.MqttMessageConverter;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
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;
@@ -106,7 +110,7 @@ import org.springframework.util.ReflectionUtils;
*/
public class MqttAdapterTests {
private IMqttToken alwaysComplete;
private final IMqttToken alwaysComplete;
{
ProxyFactoryBean pfb = new ProxyFactoryBean();
@@ -198,12 +202,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();
@@ -258,7 +262,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;
@@ -271,12 +275,14 @@ public class MqttAdapterTests {
adapter.setManualAcks(true);
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;
@@ -304,6 +310,39 @@ 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 AbstractIntegrationMessageBuilder<?> toMessageBuilder(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(Message::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");
@@ -426,7 +465,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();
@@ -458,8 +497,7 @@ 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);
@@ -567,7 +605,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

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

@@ -154,7 +154,7 @@ Starting with version 5.3, you can set the `manualAcks` property to true.
Often used to asynchronously acknowledge delivery.
When set to `true`, header (`IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK`) is added to the message with the value being a `SimpleAcknowledgment`.
You must invoke the `acknowledge()` method to complete the delivery.
See the Javadocs for `IMqppClient` `setManualAcks()` and `messageArrivedComplete()` for more information.
See the Javadocs for `IMqttClient` `setManualAcks()` and `messageArrivedComplete()` for more information.
For convenience a header accessor is provided:
====
@@ -164,6 +164,9 @@ StaticMessageHeaderAccessor.acknowledgment(someMessage).acknowledge();
----
====
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: