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**

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<StompSession>() {
@@ -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) {

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<StompSession> doConnect(StompSessionHandler handler) {
+				if (!this.thrown.getAndSet(true)) {
+					throw new RuntimeException("intentional");
+				}
+				else {
+					SettableListenableFuture<StompSession> future = new SettableListenableFuture<StompSession>();
+					StompSession stompSession = mock(StompSession.class);
+					future.set(stompSession);
+					handler.afterConnected(stompSession, getConnectHeaders());
+					return future;
+				}
+			}
+
+		};
+
+		sessionManager.start();
+
+		final SettableListenableFuture<StompSession> stompSessionFuture = new SettableListenableFuture<StompSession>();
+		sessionManager.connect(new StompSessionHandlerAdapter() {
+
+			@Override
+			public void afterConnected(StompSession session, StompHeaders connectedHeaders) {
+				stompSessionFuture.set(session);
+			}
+
+		});
+
+		assertNotNull(stompSessionFuture.get(10, TimeUnit.SECONDS));
+
+		sessionManager.stop();
+	}
+
+}