From b1b31749bae345db899e7d1f51fb92e2a55d3602 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 10 Oct 2018 09:38:23 -0400 Subject: [PATCH] GH-2471: Close client on outbound adapter Resolves https://github.com/spring-projects/spring-integration/issues/2471 # Conflicts: # spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java --- .../mqtt/outbound/MqttPahoMessageHandler.java | 15 ++++- .../integration/mqtt/MqttAdapterTests.java | 55 ++++++++++++++++--- 2 files changed, 61 insertions(+), 9 deletions(-) diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/MqttPahoMessageHandler.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/MqttPahoMessageHandler.java index aa0b53fb60..a1e15f5ede 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/MqttPahoMessageHandler.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/MqttPahoMessageHandler.java @@ -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; } diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java index 7f8a06db32..38bcbde836 100644 --- a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java @@ -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));