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

INT-3336 Add error-channel to MQTT m-d-c-a

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3336

Previously exceptions thrown in a flow downstream of a
message-driven-channel-adapter were not logged and thrown
back to the client, causing the connection to drop and
reconnect.

Add `error-channel` to the adapter to allow normal
error handling. If no error channel, catch and log
the unhandled exception.

Add `adapter.stop();` in the end of 'real' tests to close the mqtt-connection
This commit is contained in:
Gary Russell
2014-03-21 12:27:57 -04:00
committed by Artem Bilan
parent b92134fae7
commit 98d735d058
9 changed files with 221 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParser.java
View File

@@ -21,6 +21,7 @@ import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
/**
@@ -42,6 +43,7 @@ public class MqttMessageDrivenChannelAdapterParser extends AbstractChannelAdapte
MqttParserUtils.parseCommon(element, builder);
builder.addConstructorArgValue(element.getAttribute("topics"));
builder.addPropertyReference("outputChannel", channelName);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel");
return builder.getBeanDefinition();
}

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

@@ -15,6 +15,7 @@
*/
package org.springframework.integration.mqtt.inbound;
import java.util.Arrays;
import java.util.concurrent.ScheduledFuture;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
@@ -111,7 +112,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
this.cancelReconnect();
}
if (logger.isDebugEnabled()) {
logger.debug("Connected and subscribed to " + this.getTopic());
logger.debug("Connected and subscribed to " + Arrays.asList(this.getTopic()));
}
}
}
@@ -158,7 +159,13 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
@Override
public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
Message<?> message = this.getConverter().toMessage(topic, mqttMessage);
this.sendMessage(message);
try {
this.sendMessage(message);
}
catch (RuntimeException e) {
logger.error("Unhandled exception for " + message.toString(), e);
throw e;
}
}
@Override

24
spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/xml/spring-integration-mqtt-4.0.xsd
View File

@@ -47,19 +47,29 @@
<xsd:annotation>
<xsd:documentation><![CDATA[
Allows you to specify how long this inbound-channel-adapter
will wait for the message (containing the retrieved entities)
will wait for the message
to be sent successfully to the message channel, before throwing
an exception.
Keep in mind that when sending to a DirectChannel, the
invocation will occur in the sender's thread so the failing
of the send operation may be caused by other components
further downstream. By default the Inbound Channel Adapter
will wait indefinitely. The value is specified in milliseconds.
This only applies when the channel might block (such as a bounded QueueChannel
that is full).
]]>
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="error-channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.messaging.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
If a downstream exception is thrown and an error-channel is specified,
the MessagingException will be sent to this channel. Otherwise, any such exception
will be logged.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>

5
spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackTobackAdapterTests.java
View File

@@ -61,6 +61,7 @@ public class BackTobackAdapterTests {
inbound.stop();
assertEquals("foo", out.getPayload());
assertEquals("mqtt-foo", out.getHeaders().get(MqttHeaders.TOPIC));
adapter.stop();
}
@Test
@@ -90,6 +91,8 @@ public class BackTobackAdapterTests {
assertNotNull(out);
inbound.stop();
assertEquals("bar", out.getPayload());
assertEquals("mqtt-bar", out.getHeaders().get(MqttHeaders.TOPIC)); }
assertEquals("mqtt-bar", out.getHeaders().get(MqttHeaders.TOPIC));
adapter.stop();
}
}

29
spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/DownstreamExceptionTests-context.xml Normal file
View File

@@ -0,0 +1,29 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-mqtt="http://www.springframework.org/schema/integration/mqtt"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/mqtt http://www.springframework.org/schema/integration/mqtt/spring-integration-mqtt.xsd">
<int-mqtt:message-driven-channel-adapter id="noErrorChannel"
url="tcp://localhost:1883"
client-id="fooEx1" channel="foo" topics="mqtt-fooEx1" />
<int-mqtt:message-driven-channel-adapter id="withErrorChannel"
url="tcp://localhost:1883"
error-channel="errors"
client-id="fooEx2" channel="foo" topics="mqtt-fooEx2" />
<int:channel id="foo" />
<int:service-activator input-channel="foo" ref="service" />
<bean id="service" class="org.springframework.integration.mqtt.DownstreamExceptionTests$Service" />
<int:channel id="errors">
<int:queue />
</int:channel>
</beans>

139
spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/DownstreamExceptionTests.java Normal file
View File

@@ -0,0 +1,139 @@
/*
* Copyright 2014 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.mqtt;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyString;
import static org.mockito.Matchers.contains;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
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;
/**
* @author Gary Russell
* @since 4.0
*
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class DownstreamExceptionTests {
@Rule
public final BrokerRunning brokerRunning = BrokerRunning.isRunning(1883);
@Autowired
private Service service;
@Autowired
private MqttPahoMessageDrivenChannelAdapter noErrorChannel;
@Autowired
private MqttPahoMessageDrivenChannelAdapter withErrorChannel;
@Autowired
private PollableChannel errors;
@Test
public void testNoErrorChannel() throws Exception {
service.n = 0;
Log logger = spy(TestUtils.getPropertyValue(noErrorChannel, "logger", Log.class));
final CountDownLatch latch = new CountDownLatch(1);
doAnswer(new Answer<Void>() {
@Override
public Void answer(InvocationOnMock invocation) throws Throwable {
if (((String) invocation.getArguments()[0]).contains("Unhandled")) {
latch.countDown();
}
return null;
}
}).when(logger).error(anyString(), any(Throwable.class));
new DirectFieldAccessor(noErrorChannel).setPropertyValue("logger", logger);
MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out");
adapter.setDefaultTopic("mqtt-fooEx1");
adapter.afterPropertiesSet();
adapter.start();
adapter.handleMessage(new GenericMessage<String>("foo"));
service.barrier.await(10, TimeUnit.SECONDS);
service.barrier.reset();
adapter.handleMessage(new GenericMessage<String>("foo"));
service.barrier.await(10, TimeUnit.SECONDS);
assertTrue(latch.await(10, TimeUnit.SECONDS));
verify(logger).error(contains("Unhandled exception for"), any(Throwable.class));
service.barrier.reset();
adapter.stop();
}
@Test
public void testWithErrorChannel() throws Exception {
assertSame(this.errors, TestUtils.getPropertyValue(this.withErrorChannel, "errorChannel"));
service.n = 0;
MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out");
adapter.setDefaultTopic("mqtt-fooEx2");
adapter.afterPropertiesSet();
adapter.start();
adapter.handleMessage(new GenericMessage<String>("foo"));
service.barrier.await(10, TimeUnit.SECONDS);
service.barrier.reset();
adapter.handleMessage(new GenericMessage<String>("foo"));
service.barrier.await(10, TimeUnit.SECONDS);
assertNotNull(errors.receive(10000));
service.barrier.reset();
adapter.stop();
}
public static class Service {
public CyclicBarrier barrier = new CyclicBarrier(2);
public int n;
public void foo(String foo) throws Exception {
barrier.await(10, TimeUnit.SECONDS);
if (n++ > 0) {
throw new RuntimeException("bar");
}
}
}
}

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

@@ -19,6 +19,7 @@
converter="myConverter"
client-factory="clientFactory"
send-timeout="123"
error-channel="errors"
channel="out" />
<int-mqtt:message-driven-channel-adapter id="twoTopicsAdapter"
@@ -37,4 +38,6 @@
<bean id="myConverter" class="org.springframework.integration.mqtt.support.DefaultPahoMessageConverter" />
<bean id="clientFactory" class="org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory" />
<int:channel id="errors" />
</beans>

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

@@ -55,6 +55,9 @@ public class MqttMessageDrivenChannelAdapterParserTests {
@Autowired
private DefaultMqttPahoClientFactory clientFactory;
@Autowired
private MessageChannel errors;
@Test
public void testOneTopic() {
assertEquals("tcp://localhost:1883", TestUtils.getPropertyValue(oneTopicAdapter, "url"));
@@ -66,6 +69,7 @@ public class MqttMessageDrivenChannelAdapterParserTests {
assertEquals(123L, TestUtils.getPropertyValue(oneTopicAdapter, "messagingTemplate.sendTimeout"));
assertSame(out, TestUtils.getPropertyValue(oneTopicAdapter, "outputChannel"));
assertSame(clientFactory, TestUtils.getPropertyValue(oneTopicAdapter, "clientFactory"));
assertSame(errors, TestUtils.getPropertyValue(oneTopicAdapter, "errorChannel"));
}
@Test

22
src/reference/docbook/mqtt.xml
View File

@@ -48,6 +48,7 @@
converter="myConverter"]]> <co id="mqtt-i-04"/><![CDATA[
client-factory="clientFactory"]]> <co id="mqtt-i-05"/><![CDATA[
send-timeout="123"]]> <co id="mqtt-i-06"/><![CDATA[
error-channel="errors"]]> <co id="mqtt-i-07"/><![CDATA[
channel="out" />]]></programlisting>
<calloutlist>
<callout arearefs="mqtt-i-01">
@@ -79,6 +80,11 @@
The send timeout - only applies if the channel might block (such as a bounded <code>QueueChannel</code>
that is currently full).
</callout>
<callout arearefs="mqtt-i-07">
The error channel - downstream exceptions will be sent to this channel, if supplied, in an
<classname>ErrorMessage</classname>; the payload is a <classname>MessagingException</classname>
containing the failed message and cause.
</callout>
</calloutlist>
</section>
@@ -99,16 +105,16 @@
client-factory="clientFactory"]]> <co id="mqtt-o-04"/><![CDATA[
default-qos="1"]]> <co id="mqtt-o-05"/><![CDATA[
default-retained="true"]]> <co id="mqtt-o-06"/><![CDATA[
default-topic="bar"]]> <co id="mqtt-i-07"/><![CDATA[
default-topic="bar"]]> <co id="mqtt-o-07"/><![CDATA[
channel="target" />]]></programlisting>
<calloutlist>
<callout arearefs="mqtt-i-01">
<callout arearefs="mqtt-o-01">
The client id.
</callout>
<callout arearefs="mqtt-i-02">
<callout arearefs="mqtt-o-02">
The broker URL.
</callout>
<callout arearefs="mqtt-i-03">
<callout arearefs="mqtt-o-03">
An <interfacename>MqttMessageConverter</interfacename> (optional). The default
<classname>DefaultPahoMessageConverter</classname>
recognizes the following headers:
@@ -118,18 +124,18 @@
<listitem><code>mqtt_qos</code> - the quality of service</listitem>
</itemizedlist>
</callout>
<callout arearefs="mqtt-i-04">
<callout arearefs="mqtt-o-04">
The client factory.
</callout>
<callout arearefs="mqtt-i-05">
<callout arearefs="mqtt-o-05">
The default quality of service (used if no <code>mqtt_qos</code> header is found). Not allowed
if a custom <code>converter</code> is supplied.
</callout>
<callout arearefs="mqtt-i-06">
<callout arearefs="mqtt-o-06">
The default value of the retained flag (used if no <code>mqtt_retaind</code> header is found). Not allowed
if a custom <code>converter</code> is supplied.
</callout>
<callout arearefs="mqtt-i-07">
<callout arearefs="mqtt-o-07">
The default topic to which the message will be sent (used if no <code>mqtt_topic</code> header is found).
</callout>
</calloutlist>