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

GH-2471: Close client on outbound adapter

Browse Source 
Resolves https://github.com/spring-projects/spring-integration/issues/2471

# Conflicts:
#	spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java
This commit is contained in:
Gary Russell
2018-10-10 09:38:23 -04:00
committed by Artem Bilan
parent ef9c7609d5
commit b1b31749ba
2 changed files with 61 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/MqttPahoMessageHandler.java
View File

@@ -159,15 +159,17 @@ public class MqttPahoMessageHandler extends AbstractMqttMessageHandler
private synchronized IMqttAsyncClient checkConnection() throws MqttException {
if (this.client != null && !this.client.isConnected()) {
this.client.setCallback(null);
this.client.close();
this.client = null;
}
if (this.client == null) {
IMqttAsyncClient client = null;
try {
MqttConnectOptions connectionOptions = this.clientFactory.getConnectionOptions();
Assert.state(this.getUrl() != null || connectionOptions.getServerURIs() != null,
"If no 'url' provided, connectionOptions.getServerURIs() must not be null");
IMqttAsyncClient client = this.clientFactory.getAsyncClientInstance(this.getUrl(), this.getClientId());
client = this.clientFactory.getAsyncClientInstance(this.getUrl(), this.getClientId());
incrementClientInstance();
client.setCallback(this);
client.connect(connectionOptions).waitForCompletion(this.completionTimeout);
@@ -177,6 +179,10 @@ public class MqttPahoMessageHandler extends AbstractMqttMessageHandler
}
}
catch (MqttException e) {
if (client != null) {
client.close();
client = null;
}
throw new MessagingException("Failed to connect", e);
}
}
@@ -209,6 +215,13 @@ public class MqttPahoMessageHandler extends AbstractMqttMessageHandler
@Override
public synchronized void connectionLost(Throwable cause) {
logger.error("Lost connection; will attempt reconnect on next request");
try {
this.client.setCallback(null);
this.client.close();
}
catch (MqttException e) {
// NOSONAR
}
this.client = null;
}

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

@@ -23,6 +23,7 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.willAnswer;
import static org.mockito.BDDMockito.willReturn;
@@ -50,6 +51,7 @@ import javax.net.SocketFactory;
import org.aopalliance.intercept.MethodInterceptor;
import org.apache.commons.logging.Log;
import org.eclipse.paho.client.mqttv3.IMqttAsyncClient;
import org.eclipse.paho.client.mqttv3.IMqttClient;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
@@ -77,6 +79,7 @@ import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannel
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
@@ -100,15 +103,32 @@ public class MqttAdapterTests {
}
@Test
public void testCloseOnBadConnect() throws Exception {
public void testCloseOnBadConnectIn() throws Exception {
final IMqttClient client = mock(IMqttClient.class);
willThrow(new MqttException(0)).given(client).connect(any());
MqttPahoMessageDrivenChannelAdapter adapter = buildAdapter(client, null, ConsumerStopAction.UNSUBSCRIBE_NEVER);
MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, null, ConsumerStopAction.UNSUBSCRIBE_NEVER);
adapter.start();
verify(client).close();
adapter.stop();
}
@Test
public void testCloseOnBadConnectOut() throws Exception {
final IMqttAsyncClient client = mock(IMqttAsyncClient.class);
willThrow(new MqttException(0)).given(client).connect(any());
MqttPahoMessageHandler adapter = buildAdapterOut(client);
adapter.start();
try {
adapter.handleMessage(new GenericMessage<>("foo"));
fail("exception expected");
}
catch (MessagingException e) {
// NOSONAR
}
verify(client).close();
adapter.stop();
}
@Test
public void testOutboundOptionsApplied() throws Exception {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
@@ -290,7 +310,7 @@ public class MqttAdapterTests {
@Test
public void testStopActionDefault() throws Exception {
final IMqttClient client = mock(IMqttClient.class);
MqttPahoMessageDrivenChannelAdapter adapter = buildAdapter(client, null, null);
MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, null, null);
adapter.start();
adapter.stop();
@@ -300,7 +320,7 @@ public class MqttAdapterTests {
@Test
public void testStopActionDefaultNotClean() throws Exception {
final IMqttClient client = mock(IMqttClient.class);
MqttPahoMessageDrivenChannelAdapter adapter = buildAdapter(client, false, null);
MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, false, null);
adapter.start();
adapter.stop();
@@ -310,7 +330,7 @@ public class MqttAdapterTests {
@Test
public void testStopActionAlways() throws Exception {
final IMqttClient client = mock(IMqttClient.class);
MqttPahoMessageDrivenChannelAdapter adapter = buildAdapter(client, false,
MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, false,
ConsumerStopAction.UNSUBSCRIBE_ALWAYS);
adapter.start();
@@ -328,7 +348,7 @@ public class MqttAdapterTests {
@Test
public void testStopActionNever() throws Exception {
final IMqttClient client = mock(IMqttClient.class);
MqttPahoMessageDrivenChannelAdapter adapter = buildAdapter(client, null, ConsumerStopAction.UNSUBSCRIBE_NEVER);
MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, null, ConsumerStopAction.UNSUBSCRIBE_NEVER);
adapter.start();
adapter.stop();
@@ -338,7 +358,7 @@ public class MqttAdapterTests {
@Test
public void testReconnect() throws Exception {
final IMqttClient client = mock(IMqttClient.class);
MqttPahoMessageDrivenChannelAdapter adapter = buildAdapter(client, null, ConsumerStopAction.UNSUBSCRIBE_NEVER);
MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, null, ConsumerStopAction.UNSUBSCRIBE_NEVER);
adapter.setRecoveryInterval(10);
Log logger = spy(TestUtils.getPropertyValue(adapter, "logger", Log.class));
new DirectFieldAccessor(adapter).setPropertyValue("logger", logger);
@@ -364,7 +384,7 @@ public class MqttAdapterTests {
taskScheduler.destroy();
}
private MqttPahoMessageDrivenChannelAdapter buildAdapter(final IMqttClient client, Boolean cleanSession,
private MqttPahoMessageDrivenChannelAdapter buildAdapterIn(final IMqttClient client, Boolean cleanSession,
ConsumerStopAction action) throws MqttException {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory() {
@@ -392,6 +412,25 @@ public class MqttAdapterTests {
return adapter;
}
private MqttPahoMessageHandler buildAdapterOut(final IMqttAsyncClient client) throws MqttException {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory() {
@Override
public IMqttAsyncClient getAsyncClientInstance(String uri, String clientId) throws MqttException {
return client;
}
};
MqttConnectOptions connectOptions = new MqttConnectOptions();
connectOptions.setServerURIs(new String[] { "tcp://localhost:1883" });
factory.setConnectionOptions(connectOptions);
MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("client", factory);
adapter.setDefaultTopic("foo");
adapter.setApplicationEventPublisher(mock(ApplicationEventPublisher.class));
adapter.afterPropertiesSet();
return adapter;
}
private void verifyUnsubscribe(IMqttClient client) throws Exception {
verify(client).connect(any(MqttConnectOptions.class));
verify(client).subscribe(any(String[].class), any(int[].class));