Spring/spring-integration-extensions
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.

Disconnect the client after verifying the broker is available.
Change the rule to a class level rule in DownstreamExceptionTests.
This commit is contained in:
Gary Russell
2014-03-21 15:43:59 -04:00
committed by Artem Bilan
parent c583351bff
commit 592bf68086
11 changed files with 841 additions and 9 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

@@ -19,6 +19,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;
import org.w3c.dom.Element;
@@ -41,6 +42,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();
}

49
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParser.java~ Normal file
View File

@@ -0,0 +1,49 @@
/*
* Copyright 2002-2013 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.config.xml;
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.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.w3c.dom.Element;
/**
* The MqttAdapter Message Driven Channel adapter parser
*
* @author Gary Russell
* @since 1.0
*
*/
public class MqttMessageDrivenChannelAdapterParser extends AbstractChannelAdapterParser {
@Override
protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder
.genericBeanDefinition(MqttPahoMessageDrivenChannelAdapter.class);
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

174
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java~ Normal file
View File

@@ -0,0 +1,174 @@
/*
* Copyright 2002-2013 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.inbound;
import java.util.concurrent.ScheduledFuture;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.integration.Message;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
/**
* Eclipse Paho Implementation.
*
* @author Gary Russell
* @since 1.0
*
*/
public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDrivenChannelAdapter
implements MqttCallback {
private final MqttPahoClientFactory clientFactory;
private volatile MqttClient client;
private volatile ScheduledFuture<?> reconnectFuture;
private volatile boolean connected;
public MqttPahoMessageDrivenChannelAdapter(String url, String clientId, MqttPahoClientFactory clientFactory,
String... topic) {
super(url, clientId, topic);
this.clientFactory = clientFactory;
}
public MqttPahoMessageDrivenChannelAdapter(String url, String clientId, String... topic) {
this(url, clientId, new DefaultMqttPahoClientFactory(), topic);
}
@Override
protected void doStart() {
super.doStart();
try {
this.connectAndSubscribe();
}
catch (Exception e) {
logger.error("Exception while connecting and subscribing, retrying", e);
this.scheduleReconnect();
}
}
@Override
protected void doStop() {
this.cancelReconnect();
super.doStop();
try {
this.client.unsubscribe(this.getTopic());
}
catch (MqttException e) {
logger.error("Exception while unsubscribing", e);
}
try {
this.client.disconnect();
}
catch (MqttException e) {
logger.error("Exception while disconnecting", e);
}
try {
this.client.close();
}
catch (MqttException e) {
logger.error("Exception while closing", e);
}
this.connected = false;
this.client = null;
}
private void connectAndSubscribe() throws MqttException {
this.client = this.clientFactory.getClientInstance(this.getUrl(), this.getClientId());
this.client.connect(this.clientFactory.getConnectionOptions());
try {
this.client.subscribe(this.getTopic());
}
catch (MqttException e) {
this.client.disconnect();
throw e;
}
if (this.client.isConnected()) {
this.client.setCallback(this);
this.connected = true;
if (this.reconnectFuture != null) {
this.cancelReconnect();
}
if (logger.isDebugEnabled()) {
logger.debug("Connected and subscribed to " + Arrays.asList(this.getTopic()));
}
}
}
private synchronized void cancelReconnect() {
if (this.reconnectFuture != null) {
this.reconnectFuture.cancel(false);
this.reconnectFuture = null;
}
}
private void scheduleReconnect() {
try {
this.reconnectFuture = this.getTaskScheduler().scheduleWithFixedDelay(new Runnable() {
@Override
public void run() {
try {
if (logger.isDebugEnabled()) {
logger.debug("Attempting reconnect");
}
if (!connected) {
connectAndSubscribe();
}
}
catch (MqttException e) {
logger.error("Exception while connecting and subscribing", e);
}
}
}, 10000);
}
catch (Exception e) {
logger.error("Failed to schedule reconnect", e);
}
}
@Override
public void connectionLost(Throwable cause) {
this.logger.error("Lost connection:" + cause.getMessage() + "; retrying...");
this.connected = false;
this.scheduleReconnect();
}
@Override
public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
Message<?> message = this.getConverter().toMessage(topic, mqttMessage);
try {
this.sendMessage(message);
}
catch (RuntimeException e) {
logger.error("Unhandled exception for " + message.toString(), e);
throw e;
}
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
}
}

24
spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/xml/spring-integration-mqtt-1.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>

200
spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/xml/spring-integration-mqtt-1.0.xsd~ Normal file
View File

@@ -0,0 +1,200 @@
<?xml version="1.0" encoding="UTF-8"?>
<xsd:schema xmlns="http://www.springframework.org/schema/integration/mqtt"
xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:tool="http://www.springframework.org/schema/tool"
xmlns:integration="http://www.springframework.org/schema/integration"
targetNamespace="http://www.springframework.org/schema/integration/mqtt"
elementFormDefault="qualified" attributeFormDefault="unqualified">
<xsd:import namespace="http://www.springframework.org/schema/beans" />
<xsd:import namespace="http://www.springframework.org/schema/tool" />
<xsd:import namespace="http://www.springframework.org/schema/integration"
schemaLocation="http://www.springframework.org/schema/integration/spring-integration.xsd" />
<xsd:annotation>
<xsd:documentation><![CDATA[
Defines the configuration elements for the Spring Integration
Mqtt Adapters.
]]></xsd:documentation>
</xsd:annotation>
<xsd:element name="message-driven-channel-adapter">
<xsd:annotation>
<xsd:documentation>
The definition for the Spring Integration MqttAdapter
Inbound Channel Adapter.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:attributeGroup ref="coreMqttComponentAttributes"/>
<xsd:attribute name="channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="topics">
<xsd:annotation>
<xsd:documentation>
Specifies one or more (comma-delimited) topics on which to listen for messages.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="send-timeout" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Allows you to specify how long this inbound-channel-adapter
will wait for the message (containing the retrieved entities)
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.
]]>
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
<xsd:element name="outbound-channel-adapter">
<xsd:annotation>
<xsd:documentation>
Defines an outbound Channel Adapter.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:sequence>
<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1" />
</xsd:sequence>
<xsd:attributeGroup ref="coreMqttComponentAttributes"/>
<xsd:attribute name="channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
Channel from which messages will be output.
When a message is sent to this channel it will
cause the query
to be executed.
</xsd:documentation>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.MessageChannel" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="order">
<xsd:annotation>
<xsd:documentation>
Specifies the order for invocation when this endpoint is connected as a
subscriber to a SubscribableChannel.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<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.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="default-qos">
<xsd:annotation>
<xsd:documentation>
Specifies the default quality of service. Default 0.
</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.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
<xsd:attributeGroup name="coreMqttComponentAttributes">
<xsd:attribute name="id" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation>
Identifies the underlying Spring bean definition, which is an
instance of either 'EventDrivenConsumer' or 'PollingConsumer',
depending on whether the component's input channel is a
'SubscribableChannel' or 'PollableChannel'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="auto-startup" default="true" use="optional">
<xsd:annotation>
<xsd:documentation>
Flag to indicate that the component should start automatically
on startup (default true).
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:boolean xsd:string" />
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="phase" use="optional">
<xsd:annotation>
<xsd:documentation>
Flag to indicate the phase in which the component should start automatically
on startup. See SmartLifecycle.
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:integer xsd:string" />
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="url" use="required" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
MQTT broker URL.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="client-id" use="required" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
MQTT client ID.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="converter" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation><![CDATA[
A message converter to convert Spring Integration Message<String> to/from
a paho MqttMessage. Default is DefaultMqttMessageConverter.
]]></xsd:documentation>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.mqtt.support.MqttMessageConverter" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="client-factory" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation><![CDATA[
An MqttClientFactory used to create clients and connection options if you wish to
override the defaults. Default is DefaultMqttClientFactory.
]]></xsd:documentation>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.mqtt.support.MqttMessageConverter" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:attributeGroup>
</xsd:schema>

1
spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BrokerRunning.java
View File

@@ -65,6 +65,7 @@ public class BrokerRunning extends TestWatcher {
finally {
if (client != null) {
try {
client.disconnect();
client.close();
}
catch (MqttException e) {

82
spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BrokerRunning.java~ Normal file
View File

@@ -0,0 +1,82 @@
/*
* Copyright 2002-2013 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.Assume.assumeNoException;
import static org.junit.Assume.assumeTrue;
import java.util.HashMap;
import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.junit.rules.TestWatcher;
import org.junit.runner.Description;
import org.junit.runners.model.Statement;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
/**
* @author Gary Russell
* @since 1.0
*
*/
public class BrokerRunning extends TestWatcher {
private static Log logger = LogFactory.getLog(BrokerRunning.class);
// Static so that we only test once on failure: speeds up test suite
private static Map<Integer,Boolean> brokerOnline = new HashMap<Integer, Boolean>();
private final int port;
private BrokerRunning(int port) {
this.port = port;
brokerOnline.put(port, true);
}
@Override
public Statement apply(Statement base, Description description) {
assumeTrue(brokerOnline.get(port));
String url = "tcp://localhost:" + port;
MqttClient client = null;
try {
client = new DefaultMqttPahoClientFactory().getClientInstance(url, "junit-" + System.currentTimeMillis());
client.connect();
}
catch (MqttException e) {
logger.warn("Tests not running because no broker on " + url + ":", e);
assumeNoException(e);
}
finally {
if (client != null) {
try {
client.close();
}
catch (MqttException e) {
}
}
}
return super.apply(base, description);
}
public static BrokerRunning isRunning(int port) {
return new BrokerRunning(port);
}
}

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.ClassRule;
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.integration.core.PollableChannel;
import org.springframework.integration.message.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 {
@ClassRule
public static 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");
}
}
}
}

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.ClassRule;
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.integration.channel.PollableChannel;
import org.springframework.integration.message.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 {
@ClassRule
public static 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");
}
}
}
}