From bce851576e954d2a0971c45999b817d36e43df2c Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 19 Jul 2016 17:39:29 -0400 Subject: [PATCH] INT-4075: Fix NPE in AbstractStompSessionManager JIRA: https://jira.spring.io/browse/INT-4075 Fixes GH-1854 (https://github.com/spring-projects/spring-integration/issues/1854) The `catch` block just swallows an Exception with the logger and leaves `stompSessionListenableFuture` as `null`. The next `stompSessionListenableFuture` usage over two code lines bellow ends up with `NPE`. * Add reconnect logic into that `catch` block and just return from there, since we don't have anything to go ahead. * Add `if (e != null) {` around `log.error` in the `scheduleReconnect` * Add `if...else` around `this.stompClient.getTaskScheduler()` and `log.info()` to notify end-user that there won't be reconnection if there is no `taskScheduler` * Add `StompSessionManagerTests` mock tests to verify solution **Cherry-pick to 4.2.x** --- .../stomp/AbstractStompSessionManager.java | 35 +++++--- .../stomp/StompSessionManagerTests.java | 83 +++++++++++++++++++ 2 files changed, 108 insertions(+), 10 deletions(-) create mode 100644 spring-integration-stomp/src/test/java/org/springframework/integration/stomp/StompSessionManagerTests.java diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java index 4f3c751690..465a3a3b90 100644 --- a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java @@ -139,6 +139,7 @@ public abstract class AbstractStompSessionManager implements StompSessionManager } /** + * Specify a reconnect interval in milliseconds in case of lost connection. * @param recoveryInterval the reconnect interval in milliseconds in case of lost connection. * @since 4.2.2 */ @@ -189,7 +190,13 @@ public abstract class AbstractStompSessionManager implements StompSessionManager this.stompSessionListenableFuture = doConnect(this.compositeStompSessionHandler); } catch (Exception e) { - this.logger.error("doConnect() error for " + this, e); + if (epoch == this.epoch.get()) { + scheduleReconnect(e); + } + else { + this.logger.error("STOMP doConnect() error for " + this, e); + } + return; } final CountDownLatch latch = new CountDownLatch(1); this.stompSessionListenableFuture.addCallback(new ListenableFutureCallback() { @@ -239,7 +246,9 @@ public abstract class AbstractStompSessionManager implements StompSessionManager private void scheduleReconnect(Throwable e) { this.epoch.incrementAndGet(); this.connecting = this.connected = false; - this.logger.error("STOMP connect error for " + this, e); + if (e != null) { + this.logger.error("STOMP connect error for " + this, e); + } if (this.applicationEventPublisher != null) { this.applicationEventPublisher.publishEvent( new StompConnectionFailedEvent(this, e)); @@ -250,15 +259,20 @@ public abstract class AbstractStompSessionManager implements StompSessionManager this.reconnectFuture = null; } - this.reconnectFuture = this.stompClient.getTaskScheduler() - .schedule(new Runnable() { + if (this.stompClient.getTaskScheduler() != null) { + this.reconnectFuture = this.stompClient.getTaskScheduler() + .schedule(new Runnable() { - @Override - public void run() { - connect(); - } + @Override + public void run() { + connect(); + } - }, new Date(System.currentTimeMillis() + this.recoveryInterval)); + }, new Date(System.currentTimeMillis() + this.recoveryInterval)); + } + else { + this.logger.info("For automatic reconnection the 'stompClient' should be configured with a TaskScheduler."); + } } @Override @@ -395,7 +409,8 @@ public abstract class AbstractStompSessionManager implements StompSessionManager @Override public void handleTransportError(StompSession session, Throwable exception) { - AbstractStompSessionManager.this.logger.error("STOMP transport error for session: [" + session + "]", exception); + AbstractStompSessionManager.this.logger.error("STOMP transport error for session: [" + session + "]", + exception); this.session = null; scheduleReconnect(exception); synchronized (this.delegates) { diff --git a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/StompSessionManagerTests.java b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/StompSessionManagerTests.java new file mode 100644 index 0000000000..0658cc039c --- /dev/null +++ b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/StompSessionManagerTests.java @@ -0,0 +1,83 @@ +/* + * Copyright 2016 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.stomp; + +import static org.junit.Assert.assertNotNull; +import static org.mockito.Mockito.mock; + +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.junit.Test; + +import org.springframework.messaging.simp.stomp.StompClientSupport; +import org.springframework.messaging.simp.stomp.StompHeaders; +import org.springframework.messaging.simp.stomp.StompSession; +import org.springframework.messaging.simp.stomp.StompSessionHandler; +import org.springframework.messaging.simp.stomp.StompSessionHandlerAdapter; +import org.springframework.scheduling.concurrent.ConcurrentTaskScheduler; +import org.springframework.util.concurrent.ListenableFuture; +import org.springframework.util.concurrent.SettableListenableFuture; + +/** + * @author Artem Bilan + * @since 4.2.9 + */ +public class StompSessionManagerTests { + + @Test + public void testDoConnectFailure() throws Exception { + StompClientSupport stompClient = mock(StompClientSupport.class); + stompClient.setTaskScheduler(new ConcurrentTaskScheduler()); + AbstractStompSessionManager sessionManager = new AbstractStompSessionManager(stompClient) { + + private final AtomicBoolean thrown = new AtomicBoolean(); + + @Override + protected ListenableFuture doConnect(StompSessionHandler handler) { + if (!this.thrown.getAndSet(true)) { + throw new RuntimeException("intentional"); + } + else { + SettableListenableFuture future = new SettableListenableFuture(); + StompSession stompSession = mock(StompSession.class); + future.set(stompSession); + handler.afterConnected(stompSession, getConnectHeaders()); + return future; + } + } + + }; + + sessionManager.start(); + + final SettableListenableFuture stompSessionFuture = new SettableListenableFuture(); + sessionManager.connect(new StompSessionHandlerAdapter() { + + @Override + public void afterConnected(StompSession session, StompHeaders connectedHeaders) { + stompSessionFuture.set(session); + } + + }); + + assertNotNull(stompSessionFuture.get(10, TimeUnit.SECONDS)); + + sessionManager.stop(); + } + +}