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

GH-2471: MQTT: Fix Thread Leak

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

Call `close()` on the client whenever the connection is lost or can't be
established, to release resources in the client.

**cherry-pick to 5.0.x, 4.3.x**
This commit is contained in:
Gary Russell
2018-06-06 11:10:20 -04:00
committed by Artem Bilan
parent 88120a667e
commit f9cea64a79
2 changed files with 29 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -31,6 +31,7 @@ import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.willAnswer;
import static org.mockito.BDDMockito.willReturn;
import static org.mockito.BDDMockito.willThrow;
import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
@@ -111,6 +112,16 @@ public class MqttAdapterTests {
this.alwaysComplete = (IMqttToken) pfb.getObject();
}
@Test
public void testCloseOnBadConnect() throws Exception {
final IMqttClient client = mock(IMqttClient.class);
willThrow(new MqttException(0)).given(client).connect(any());
MqttPahoMessageDrivenChannelAdapter adapter = buildAdapter(client, null, ConsumerStopAction.UNSUBSCRIBE_NEVER);
adapter.start();
verify(client).close();
adapter.stop();
}
@Test
public void testOutboundOptionsApplied() throws Exception {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
@@ -383,6 +394,7 @@ public class MqttAdapterTests {
adapter.setTaskScheduler(taskScheduler);
adapter.start();
adapter.connectionLost(new RuntimeException("initial"));
verify(client).close();
Thread.sleep(1000);
// the following assertion should be equalTo, but leq to protect against a slow CI server
assertThat(attemptingReconnectCount.get(), lessThanOrEqualTo(2));