INT-3882: StompSession Reconnection Logic

JIRA: https://jira.spring.io/browse/INT-3882

Add reconnect support for the StompSession:
* Introduce `recoveryInterval` for the `AbstractStompSessionManager`
* Add reconnect scheduled task
* Add handling of the `ConnectionLostException` into the `AbstractStompSessionManager`,
as well as for the `StompInboundChannelAdapter` and `StompMessageHandler`
* Cover adapters reconnection feature with tests
* Documentation polishing

Rework logic according PR comments

* Make `StompMessageHandler` as a "lazy-load" for the connection
* Add "direct" connect interaction for the `AbstractStompSessionManager`
* Polishing tests

Polishing

Fix `StompAdaptersParserTests#testStompSessionManagerReconnect()` to use "fake" server port

Address PR comments

The further polishing

Some further polishing

* `AbstractStompSessionManager`: `reconnectFuture.cancel(true)` on each `scheduleReconnect()` to avoid something like "DDoS attack"
* Add reconnect feature test for the `StompInboundChannelAdapterWebSocketIntegrationTests`, closing and then refreshing again the `serverContext`

spring-integration-mqtt/src/main/resources/META-INF/spring.tooling

@@ -1,4 +1,4 @@
-# Tooling related information for the integration MqttAdapter namespace
-http\://www.springframework.org/schema/integration/mqttadapter@name=integration MqttAdapter Namespace
-http\://www.springframework.org/schema/integration/mqttadapter@prefix=int-mqttadapter
-http\://www.springframework.org/schema/integration/mqttadapter@icon=org/springframework/integration/config/xml/spring-integration-mqttadapter.gif
+# Tooling related information for the integration mqtt namespace
+http\://www.springframework.org/schema/integration/mqttadapter@name=integration mqtt Namespace
+http\://www.springframework.org/schema/integration/mqttadapter@prefix=int-mqtt
+http\://www.springframework.org/schema/integration/mqttadapter@icon=org/springframework/integration/config/xml/spring-integration-mqtt.gif

spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java

@@ -19,16 +19,19 @@ package org.springframework.integration.stomp;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.ScheduledFuture;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
 import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.beans.factory.DisposableBean;
-import org.springframework.beans.factory.InitializingBean;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
-import org.springframework.integration.stomp.event.StompExceptionEvent;
+import org.springframework.context.SmartLifecycle;
+import org.springframework.integration.stomp.event.StompConnectionFailedEvent;
+import org.springframework.integration.stomp.event.StompSessionConnectedEvent;
+import org.springframework.messaging.simp.stomp.ConnectionLostException;
 import org.springframework.messaging.simp.stomp.StompClientSupport;
 import org.springframework.messaging.simp.stomp.StompCommand;
 import org.springframework.messaging.simp.stomp.StompHeaders;
@@ -43,9 +46,9 @@ import org.springframework.util.concurrent.ListenableFutureCallback;
  * Base {@link StompSessionManager} implementation to manage a single {@link StompSession}
  * over its {@link ListenableFuture} from the target implementation of this class.
  * <p>
- * The connection to the {@link StompSession} is made during {@link #afterPropertiesSet()}.
+ * The connection to the {@link StompSession} is made during {@link #start()}.
  * <p>
- * The {@link #destroy()} lifecycle method manages {@link StompSession#disconnect()}.
+ * The {@link #stop()} lifecycle method manages {@link StompSession#disconnect()}.
  * <p>
  * The {@link #connect(StompSessionHandler)} and {@link #disconnect(StompSessionHandler)} method
  * implementations populate/remove the provided {@link StompSessionHandler} to/from an internal
@@ -53,19 +56,28 @@ import org.springframework.util.concurrent.ListenableFutureCallback;
  * to the provided {@link StompSessionHandler}s.
  * This {@link AbstractStompSessionManager.CompositeStompSessionHandler} is used for the
  * {@link StompSession} connection.
- *
  * @author Artem Bilan
  * @since 4.2
  */
 public abstract class AbstractStompSessionManager implements StompSessionManager, ApplicationEventPublisherAware,
-		InitializingBean, DisposableBean, BeanNameAware {
+		SmartLifecycle, DisposableBean, BeanNameAware {
+
+	private static final long DEFAULT_RECOVERY_INTERVAL = 10000;
 
 	protected final Log logger = LogFactory.getLog(getClass());
 
 	private final CompositeStompSessionHandler compositeStompSessionHandler = new CompositeStompSessionHandler();
 
+	private final Object lifecycleMonitor = new Object();
+
 	protected final StompClientSupport stompClient;
 
+	private boolean autoStartup = false;
+
+	private boolean running = false;
+
+	private int phase = Integer.MAX_VALUE / 2;
+
 	private ApplicationEventPublisher applicationEventPublisher;
 
 	private volatile StompHeaders connectHeaders;
@@ -74,8 +86,14 @@ public abstract class AbstractStompSessionManager implements StompSessionManager
 
 	private volatile boolean autoReceipt;
 
+	private volatile boolean connecting;
+
 	private volatile boolean connected;
 
+	private volatile long recoveryInterval = DEFAULT_RECOVERY_INTERVAL;
+
+	private volatile ScheduledFuture<?> reconnectFuture;
+
 	private String name;
 
 	public AbstractStompSessionManager(StompClientSupport stompClient) {
@@ -111,50 +129,160 @@ public abstract class AbstractStompSessionManager implements StompSessionManager
 		this.name = name;
 	}
 
+	/**
+	 * @param recoveryInterval the reconnect interval in milliseconds in case of lost connection.
+	 * @since 4.2.2
+	 */
+	public void setRecoveryInterval(int recoveryInterval) {
+		this.recoveryInterval = recoveryInterval;
+	}
+
+	public void setAutoStartup(boolean autoStartup) {
+		this.autoStartup = autoStartup;
+	}
+
+	public void setPhase(int phase) {
+		this.phase = phase;
+	}
+
+	public long getRecoveryInterval() {
+		return recoveryInterval;
+	}
+
 	@Override
-	public void afterPropertiesSet() throws Exception {
+	public boolean isAutoStartup() {
+		return this.autoStartup;
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.running;
+	}
+
+	@Override
+	public int getPhase() {
+		return this.phase;
+	}
+
+	private void connect() {
+		this.connecting = true;
 		this.stompSessionListenableFuture = doConnect(this.compositeStompSessionHandler);
 		this.stompSessionListenableFuture.addCallback(new ListenableFutureCallback<StompSession>() {
 
 			@Override
 			public void onFailure(Throwable e) {
-				logger.error("STOMP connect error.", e);
-				if (applicationEventPublisher != null) {
-					applicationEventPublisher.publishEvent(
-							new StompExceptionEvent(AbstractStompSessionManager.this, e));
-				}
+				scheduleReconnect(e);
 			}
 
 			@Override
 			public void onSuccess(StompSession stompSession) {
-				stompSession.setAutoReceipt(autoReceipt);
 				AbstractStompSessionManager.this.connected = true;
+				AbstractStompSessionManager.this.connecting = false;
+				stompSession.setAutoReceipt(isAutoReceiptEnabled());
+				if (AbstractStompSessionManager.this.applicationEventPublisher != null) {
+					AbstractStompSessionManager.this.applicationEventPublisher.publishEvent(
+							new StompSessionConnectedEvent(this));
+				}
+				AbstractStompSessionManager.this.reconnectFuture = null;
 			}
 
 		});
 	}
 
+	private void scheduleReconnect(Throwable e) {
+		if (this.reconnectFuture != null) {
+			this.reconnectFuture.cancel(true);
+			this.reconnectFuture = null;
+		}
+		this.connecting = this.connected = false;
+		logger.error("STOMP connect error.", e);
+		if (this.applicationEventPublisher != null) {
+			this.applicationEventPublisher.publishEvent(
+					new StompConnectionFailedEvent(this, e));
+		}
+
+		this.reconnectFuture = this.stompClient.getTaskScheduler()
+				.scheduleWithFixedDelay(new Runnable() {
+
+					@Override
+					public void run() {
+						connect();
+					}
+
+				}, this.recoveryInterval);
+	}
+
 	@Override
-	public void destroy() throws Exception {
-		this.stompSessionListenableFuture.addCallback(new ListenableFutureCallback<StompSession>() {
-
-			@Override
-			public void onFailure(Throwable ex) {
-				AbstractStompSessionManager.this.connected = false;
+	public void destroy() {
+		if (this.stompSessionListenableFuture != null) {
+			if (this.reconnectFuture != null) {
+				this.reconnectFuture.cancel(false);
+				this.reconnectFuture = null;
 			}
+			this.stompSessionListenableFuture.addCallback(new ListenableFutureCallback<StompSession>() {
 
-			@Override
-			public void onSuccess(StompSession session) {
-				session.disconnect();
-				AbstractStompSessionManager.this.connected = false;
+				@Override
+				public void onFailure(Throwable ex) {
+					AbstractStompSessionManager.this.connected = false;
+				}
+
+				@Override
+				public void onSuccess(StompSession session) {
+					session.disconnect();
+					AbstractStompSessionManager.this.connected = false;
+				}
+
+			});
+			this.stompSessionListenableFuture = null;
+		}
+	}
+
+	@Override
+	public void start() {
+		synchronized (this.lifecycleMonitor) {
+			if (!isRunning()) {
+				if (logger.isInfoEnabled()) {
+					logger.info("Starting " + getClass().getSimpleName());
+				}
+				connect();
+				this.running = true;
 			}
+		}
+	}
 
-		});
+	@Override
+	public void stop(Runnable callback) {
+		synchronized (this.lifecycleMonitor) {
+			stop();
+			if (callback != null) {
+				callback.run();
+			}
+		}
+	}
+
+	@Override
+	public void stop() {
+		synchronized (this.lifecycleMonitor) {
+			if (isRunning()) {
+				this.running = false;
+				if (logger.isInfoEnabled()) {
+					logger.info("Stopping " + getClass().getSimpleName());
+				}
+				destroy();
+			}
+		}
 	}
 
 	@Override
 	public void connect(StompSessionHandler handler) {
 		this.compositeStompSessionHandler.addHandler(handler);
+		if (!isConnected() && !this.connecting) {
+			if (this.reconnectFuture != null) {
+				this.reconnectFuture.cancel(true);
+				this.reconnectFuture = null;
+			}
+			connect();
+		}
 	}
 
 	@Override
@@ -177,18 +305,16 @@ public abstract class AbstractStompSessionManager implements StompSessionManager
 	protected abstract ListenableFuture<StompSession> doConnect(StompSessionHandler handler);
 
 
-	private static class CompositeStompSessionHandler extends StompSessionHandlerAdapter {
+	private class CompositeStompSessionHandler extends StompSessionHandlerAdapter {
 
 		private final List<StompSessionHandler> delegates =
 				Collections.synchronizedList(new ArrayList<StompSessionHandler>());
 
 		private volatile StompSession session;
 
-		private volatile StompHeaders connectedHeaders;
-
 		void addHandler(StompSessionHandler delegate) {
 			if (this.session != null) {
-				delegate.afterConnected(this.session, this.connectedHeaders);
+				delegate.afterConnected(this.session, getConnectHeaders());
 			}
 			synchronized (this.delegates) {
 				this.delegates.add(delegate);
@@ -204,7 +330,6 @@ public abstract class AbstractStompSessionManager implements StompSessionManager
 		@Override
 		public void afterConnected(StompSession session, StompHeaders connectedHeaders) {
 			this.session = session;
-			this.connectedHeaders = connectedHeaders;
 			synchronized (this.delegates) {
 				for (StompSessionHandler delegate : this.delegates) {
 					delegate.afterConnected(session, connectedHeaders);
@@ -214,7 +339,7 @@ public abstract class AbstractStompSessionManager implements StompSessionManager
 
 		@Override
 		public void handleException(StompSession session, StompCommand command, StompHeaders headers, byte[] payload,
-				Throwable exception) {
+		                            Throwable exception) {
 			synchronized (this.delegates) {
 				for (StompSessionHandler delegate : this.delegates) {
 					delegate.handleException(session, command, headers, payload, exception);
@@ -224,6 +349,10 @@ public abstract class AbstractStompSessionManager implements StompSessionManager
 
 		@Override
 		public void handleTransportError(StompSession session, Throwable exception) {
+			if (exception instanceof ConnectionLostException) {
+				this.session = null;
+				scheduleReconnect(exception);
+			}
 			synchronized (this.delegates) {
 				for (StompSessionHandler delegate : this.delegates) {
 					delegate.handleTransportError(session, exception);

spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/StompConnectionFailedEvent.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2015 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.event;
+
+/**
+ * The {@link StompIntegrationEvent} implementation for the failed connection exceptions.
+ *
+ * @author Artem Bilan
+ * @since 4.2.2
+ */
+@SuppressWarnings("serial")
+public class StompConnectionFailedEvent extends StompIntegrationEvent {
+
+	public StompConnectionFailedEvent(Object source, Throwable cause) {
+		super(source, cause);
+	}
+
+}

spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/StompSessionConnectedEvent.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2015 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.event;
+
+/**
+ * The {@link StompIntegrationEvent} indicating the STOMP session establishment.
+ *
+ * @author Artem Bilan
+ * @since 4.2.2
+ */
+@SuppressWarnings("serial")
+public class StompSessionConnectedEvent extends StompIntegrationEvent {
+
+	public StompSessionConnectedEvent(Object source) {
+		super(source);
+	}
+
+}

spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java

@@ -26,6 +26,7 @@ import java.util.concurrent.locks.ReentrantLock;
 
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
+import org.springframework.context.Lifecycle;
 import org.springframework.integration.endpoint.MessageProducerSupport;
 import org.springframework.integration.mapping.HeaderMapper;
 import org.springframework.integration.stomp.StompSessionManager;
@@ -38,6 +39,7 @@ import org.springframework.jmx.export.annotation.ManagedResource;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandlingException;
+import org.springframework.messaging.simp.stomp.ConnectionLostException;
 import org.springframework.messaging.simp.stomp.StompCommand;
 import org.springframework.messaging.simp.stomp.StompFrameHandler;
 import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
@@ -190,16 +192,24 @@ public class StompInboundChannelAdapter extends MessageProducerSupport implement
 
 	@Override
 	protected void doStart() {
+		if (this.stompSessionManager instanceof Lifecycle) {
+			((Lifecycle) this.stompSessionManager).start();
+		}
 		this.stompSessionManager.connect(this.stompSessionHandler);
 	}
 
 	@Override
 	protected void doStop() {
-		for (StompSession.Subscription subscription : this.subscriptions.values()) {
-			subscription.unsubscribe();
+		this.stompSessionManager.disconnect(this.stompSessionHandler);
+		try {
+			for (StompSession.Subscription subscription : this.subscriptions.values()) {
+				subscription.unsubscribe();
+			}
+		}
+		catch (Exception e) {
+			logger.warn("The exception during unsubscribtion.", e);
 		}
 		this.subscriptions.clear();
-		this.stompSessionManager.disconnect(this.stompSessionHandler);
 	}
 
 	private void subscribeDestination(final String destination) {
@@ -262,7 +272,7 @@ public class StompInboundChannelAdapter extends MessageProducerSupport implement
 		}
 		else {
 			logger.warn("The StompInboundChannelAdapter [" + getComponentName() +
-					"] hasn't been connected to StompSession. Check the state of [" + this.stompSessionManager + "]");
+					"] ins't connected to StompSession. Check the state of [" + this.stompSessionManager + "]");
 		}
 	}
 
@@ -294,6 +304,9 @@ public class StompInboundChannelAdapter extends MessageProducerSupport implement
 
 		@Override
 		public void handleTransportError(StompSession session, Throwable exception) {
+			if (exception instanceof ConnectionLostException) {
+				StompInboundChannelAdapter.this.stompSession = null;
+			}
 			logger.error("STOMP transport error for session: [" + session + "]", exception);
 		}
 

spring-integration-stomp/src/main/java/org/springframework/integration/stomp/outbound/StompMessageHandler.java

@@ -16,6 +16,9 @@
 
 package org.springframework.integration.stomp.outbound;
 
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.context.Lifecycle;
@@ -32,28 +35,35 @@ import org.springframework.integration.stomp.support.StompHeaderMapper;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageDeliveryException;
 import org.springframework.messaging.MessagingException;
+import org.springframework.messaging.simp.stomp.ConnectionLostException;
 import org.springframework.messaging.simp.stomp.StompCommand;
 import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
 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.messaging.support.MessageBuilder;
 import org.springframework.util.Assert;
 
 /**
  * The {@link AbstractMessageHandler} implementation to send messages to STOMP destinations.
- *
  * @author Artem Bilan
  * @since 4.2
  */
 public class StompMessageHandler extends AbstractMessageHandler implements ApplicationEventPublisherAware, Lifecycle {
 
+	private static final int DEFAULT_CONNECT_TIMEOUT = 3000;
+
 	private final StompSessionHandler sessionHandler = new IntegrationOutboundStompSessionHandler();
 
 	private final StompSessionManager stompSessionManager;
 
+	private final Semaphore connectSemaphore = new Semaphore(0);
+
 	private volatile StompSession stompSession;
 
+	private volatile Throwable transportError;
+
 	private volatile boolean running;
 
 	private volatile HeaderMapper<StompHeaders> headerMapper = new StompHeaderMapper();
@@ -64,6 +74,8 @@ public class StompMessageHandler extends AbstractMessageHandler implements Appli
 
 	private ApplicationEventPublisher applicationEventPublisher;
 
+	private volatile long connectTimeout = DEFAULT_CONNECT_TIMEOUT;
+
 	public StompMessageHandler(StompSessionManager stompSessionManager) {
 		Assert.notNull(stompSessionManager, "'stompSessionManager' is required.");
 		this.stompSessionManager = stompSessionManager;
@@ -84,6 +96,17 @@ public class StompMessageHandler extends AbstractMessageHandler implements Appli
 		this.headerMapper = headerMapper;
 	}
 
+	/**
+	 * Specify the the timeout in milliseconds to wait for the STOMP session establishment.
+	 * Must be greater than
+	 * {@link org.springframework.integration.stomp.AbstractStompSessionManager#setRecoveryInterval(int)}.
+	 * @param connectTimeout the timeout to use.
+	 * @since 4.2.2
+	 */
+	public void setConnectTimeout(long connectTimeout) {
+		this.connectTimeout = connectTimeout;
+	}
+
 	public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) {
 		this.evaluationContext = evaluationContext;
 	}
@@ -104,10 +127,13 @@ public class StompMessageHandler extends AbstractMessageHandler implements Appli
 
 	@Override
 	protected void handleMessageInternal(final Message<?> message) throws Exception {
-		if (!this.isRunning()) {
-			throw new MessageDeliveryException(message, "The StompMessageHandler [" + getComponentName() +
-					"] hasn't been connected to StompSession. Check the state of [" + this.stompSessionManager + "]");
+		try {
+			connectIfNecessary();
 		}
+		catch (Exception e) {
+			throw new MessageDeliveryException(message, "The [" + this + "] could not deliver message." , e);
+		}
+		StompSession stompSession = this.stompSession;
 
 		StompHeaders stompHeaders = new StompHeaders();
 		this.headerMapper.fromHeaders(message.getHeaders(), stompHeaders);
@@ -118,7 +144,7 @@ public class StompMessageHandler extends AbstractMessageHandler implements Appli
 			stompHeaders.setDestination(destination);
 		}
 
-		final StompSession.Receiptable receiptable = this.stompSession.send(stompHeaders, message.getPayload());
+		final StompSession.Receiptable receiptable = stompSession.send(stompHeaders, message.getPayload());
 		if (receiptable.getReceiptId() != null) {
 			final String destination = stompHeaders.getDestination();
 			if (this.applicationEventPublisher != null) {
@@ -154,15 +180,40 @@ public class StompMessageHandler extends AbstractMessageHandler implements Appli
 		}
 	}
 
+	private StompSession connectIfNecessary() throws Exception {
+		synchronized (this.sessionHandler) {
+			if (this.stompSession == null || !this.stompSessionManager.isConnected()) {
+				this.stompSessionManager.disconnect(this.sessionHandler);
+				this.stompSessionManager.connect(this.sessionHandler);
+				if (!this.connectSemaphore.tryAcquire(this.connectTimeout, TimeUnit.MILLISECONDS)
+						|| this.stompSession == null) {
+					if (this.transportError != null) {
+						if (this.transportError instanceof ConnectionLostException) {
+							throw (ConnectionLostException) this.transportError;
+						}
+						else {
+							throw new ConnectionLostException(this.transportError.getMessage());
+						}
+					}
+					else {
+						throw new ConnectionLostException("Failed to obtain StompSession during timeout: "
+								+ this.connectTimeout);
+					}
+				}
+			}
+			return this.stompSession;
+		}
+	}
+
 	@Override
 	public void start() {
-		this.stompSessionManager.connect(this.sessionHandler);
+		this.running = true;
 	}
 
 	@Override
 	public void stop() {
-		this.stompSessionManager.disconnect(this.sessionHandler);
 		this.running = false;
+		this.stompSessionManager.disconnect(this.sessionHandler);
 	}
 
 	@Override
@@ -170,13 +221,13 @@ public class StompMessageHandler extends AbstractMessageHandler implements Appli
 		return this.running;
 	}
 
-
 	private class IntegrationOutboundStompSessionHandler extends StompSessionHandlerAdapter {
 
 		@Override
 		public void afterConnected(StompSession session, StompHeaders connectedHeaders) {
+			StompMessageHandler.this.transportError = null;
 			StompMessageHandler.this.stompSession = session;
-			StompMessageHandler.this.running = true;
+			StompMessageHandler.this.connectSemaphore.release();
 		}
 
 		@Override
@@ -200,8 +251,22 @@ public class StompMessageHandler extends AbstractMessageHandler implements Appli
 		}
 
 		@Override
-		public void handleTransportError(StompSession session, Throwable exception) {
-			logger.error("STOMP transport error for session: [" + session + "]", exception);
+		public void handleException(StompSession session, StompCommand command, StompHeaders headers, byte[] payload,
+		                            Throwable exception) {
+			Message<byte[]> message = MessageBuilder.createMessage(payload,
+					StompHeaderAccessor.create(command, headers).getMessageHeaders());
+			logger.error("The exception for session [" + session + "] on message [" + message + "]", exception);
+		}
+
+		@Override
+		public synchronized void handleTransportError(StompSession session, Throwable exception) {
+			StompMessageHandler.this.transportError = exception;
+			if (exception instanceof ConnectionLostException) {
+				StompMessageHandler.this.stompSession = null;
+			}
+			else {
+				logger.error("STOMP transport error for session: [" + session + "]", exception);
+			}
 		}
 
 	}

spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java

@@ -16,17 +16,18 @@
 
 package org.springframework.integration.stomp.client;
 
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.core.IsInstanceOf.instanceOf;
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
 
 import org.apache.activemq.broker.BrokerService;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
-import org.junit.Rule;
 import org.junit.Test;
 
 import org.springframework.context.ApplicationEvent;
@@ -42,14 +43,17 @@ import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.event.inbound.ApplicationEventListeningMessageProducer;
 import org.springframework.integration.stomp.Reactor2TcpStompSessionManager;
 import org.springframework.integration.stomp.StompSessionManager;
+import org.springframework.integration.stomp.event.StompConnectionFailedEvent;
 import org.springframework.integration.stomp.event.StompIntegrationEvent;
 import org.springframework.integration.stomp.event.StompReceiptEvent;
+import org.springframework.integration.stomp.event.StompSessionConnectedEvent;
 import org.springframework.integration.stomp.inbound.StompInboundChannelAdapter;
 import org.springframework.integration.stomp.outbound.StompMessageHandler;
 import org.springframework.integration.support.converter.PassThruMessageConverter;
-import org.springframework.integration.test.support.LongRunningIntegrationTest;
+import org.springframework.integration.test.support.LogAdjustingTestSupport;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageDeliveryException;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.simp.stomp.Reactor2TcpStompClient;
@@ -63,15 +67,17 @@ import org.springframework.util.SocketUtils;
  * @author Gary Russell
  * @since 4.2
  */
-public class StompServerIntegrationTests {
-
-	@Rule
-	public LongRunningIntegrationTest longTests = new LongRunningIntegrationTest();
+public class StompServerIntegrationTests extends LogAdjustingTestSupport {
 
 	private static BrokerService activeMQBroker;
 
 	private static Reactor2TcpStompClient stompClient;
 
+
+	public StompServerIntegrationTests() {
+		super("org.springframework", "org.springframework.integration.stomp");
+	}
+
 	@BeforeClass
 	public static void setup() throws Exception {
 		int port = SocketUtils.findAvailableTcpPort(61613);
@@ -97,7 +103,7 @@ public class StompServerIntegrationTests {
 	}
 
 	@Test
-	public void testStompAdapters() {
+	public void testStompAdapters() throws Exception {
 		ConfigurableApplicationContext context1 = new AnnotationConfigApplicationContext(ContextConfiguration.class);
 		ConfigurableApplicationContext context2 = new AnnotationConfigApplicationContext(ContextConfiguration.class);
 
@@ -112,11 +118,19 @@ public class StompServerIntegrationTests {
 
 		Message<?> eventMessage = stompEvents1.receive(10000);
 		assertNotNull(eventMessage);
+		assertThat(eventMessage.getPayload(), instanceOf(StompSessionConnectedEvent.class));
+
+		eventMessage = stompEvents1.receive(10000);
+		assertNotNull(eventMessage);
 		assertThat(eventMessage.getPayload(), instanceOf(StompReceiptEvent.class));
 		StompReceiptEvent stompReceiptEvent = (StompReceiptEvent) eventMessage.getPayload();
 		assertEquals(StompCommand.SUBSCRIBE, stompReceiptEvent.getStompCommand());
 		assertEquals("/topic/myTopic", stompReceiptEvent.getDestination());
 
+		eventMessage = stompEvents2.receive(10000);
+		assertNotNull(eventMessage);
+		assertThat(eventMessage.getPayload(), instanceOf(StompSessionConnectedEvent.class));
+
 		eventMessage = stompEvents2.receive(10000);
 		assertNotNull(eventMessage);
 		assertThat(eventMessage.getPayload(), instanceOf(StompReceiptEvent.class));
@@ -180,6 +194,42 @@ public class StompServerIntegrationTests {
 		assertNotNull(receive24);
 		assertArrayEquals("???".getBytes(), (byte[]) receive24.getPayload());
 
+		activeMQBroker.stop();
+
+		do {
+			eventMessage = stompEvents1.receive(10000);
+			assertNotNull(eventMessage);
+		}
+		while (!(eventMessage.getPayload() instanceof StompConnectionFailedEvent));
+
+		try {
+			stompOutputChannel1.send(new GenericMessage<byte[]>("foo".getBytes()));
+			fail("MessageDeliveryException is expected");
+		}
+		catch (Exception e) {
+			assertThat(e, instanceOf(MessageDeliveryException.class));
+			assertThat(e.getMessage(), containsString("could not deliver message"));
+		}
+
+		activeMQBroker.start(false);
+
+		do {
+			eventMessage = stompEvents1.receive(10000);
+			assertNotNull(eventMessage);
+		}
+		while (!(eventMessage.getPayload() instanceof StompReceiptEvent));
+
+		do {
+			eventMessage = stompEvents2.receive(10000);
+			assertNotNull(eventMessage);
+		}
+		while (!(eventMessage.getPayload() instanceof StompReceiptEvent));
+
+		stompOutputChannel1.send(new GenericMessage<byte[]>("foo".getBytes()));
+		Message<?> receive25 = stompInputChannel2.receive(10000);
+		assertNotNull(receive25);
+		assertArrayEquals("foo".getBytes(), (byte[]) receive25.getPayload());
+
 		context1.close();
 		context2.close();
 	}
@@ -192,6 +242,7 @@ public class StompServerIntegrationTests {
 		public StompSessionManager stompSessionManager() {
 			Reactor2TcpStompSessionManager stompSessionManager = new Reactor2TcpStompSessionManager(stompClient);
 			stompSessionManager.setAutoReceipt(true);
+			stompSessionManager.setRecoveryInterval(500);
 			return stompSessionManager;
 		}
 
@@ -213,6 +264,7 @@ public class StompServerIntegrationTests {
 		public MessageHandler stompMessageHandler() {
 			StompMessageHandler handler = new StompMessageHandler(stompSessionManager());
 			handler.setDestination("/topic/myTopic");
+			handler.setConnectTimeout(1000);
 			return handler;
 		}
 

spring-integration-stomp/src/test/java/org/springframework/integration/stomp/config/StompAdaptersParserTests.java

@@ -164,7 +164,6 @@ public class StompAdaptersParserTests {
 		List<SmartLifecycle> bars = lifecycles.get("bar");
 		bars.contains(this.customInboundAdapter);
 		assertTrue(lifecycles.containsKey("foo"));
-		List<SmartLifecycle> foos = lifecycles.get("bar");
 		bars.contains(this.customOutboundAdapter);
 	}
 

spring-integration-stomp/src/test/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapterWebSocketIntegrationTests.java

@@ -38,6 +38,7 @@ import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationEvent;
 import org.springframework.context.ApplicationListener;
+import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.http.MediaType;
@@ -46,8 +47,10 @@ import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.event.inbound.ApplicationEventListeningMessageProducer;
 import org.springframework.integration.stomp.StompSessionManager;
 import org.springframework.integration.stomp.WebSocketStompSessionManager;
+import org.springframework.integration.stomp.event.StompConnectionFailedEvent;
 import org.springframework.integration.stomp.event.StompIntegrationEvent;
 import org.springframework.integration.stomp.event.StompReceiptEvent;
+import org.springframework.integration.stomp.event.StompSessionConnectedEvent;
 import org.springframework.integration.test.rule.Log4jLevelAdjuster;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.integration.websocket.TomcatWebSocketTestServer;
@@ -96,7 +99,7 @@ public class StompInboundChannelAdapterWebSocketIntegrationTests {
 	public Log4jLevelAdjuster adjuster = new Log4jLevelAdjuster(Level.TRACE, "org.springframework");
 
 	@Value("#{server.serverContext}")
-	private ApplicationContext serverContext;
+	private ConfigurableApplicationContext serverContext;
 
 	@Autowired
 	@Qualifier("stompInputChannel")
@@ -114,7 +117,11 @@ public class StompInboundChannelAdapterWebSocketIntegrationTests {
 	private StompInboundChannelAdapter stompInboundChannelAdapter;
 
 	@Test
-	public void testWebSocketStompClient() throws InterruptedException {
+	public void testWebSocketStompClient() throws Exception {
+		Message<?> eventMessage = this.stompEvents.receive(10000);
+		assertNotNull(eventMessage);
+		assertThat(eventMessage.getPayload(), instanceOf(StompSessionConnectedEvent.class));
+
 		Message<?> receive = this.stompEvents.receive(10000);
 		assertNotNull(receive);
 		assertThat(receive.getPayload(), instanceOf(StompReceiptEvent.class));
@@ -165,6 +172,27 @@ public class StompInboundChannelAdapterWebSocketIntegrationTests {
 		assertThat(throwable, instanceOf(MessageHandlingException.class));
 		assertThat(throwable.getCause(), instanceOf(MessageConversionException.class));
 		assertThat(throwable.getMessage(), containsString("No suitable converter, payloadType=interface java.util.Map"));
+
+		this.serverContext.close();
+
+		eventMessage = this.stompEvents.receive(10000);
+		assertNotNull(eventMessage);
+		assertThat(eventMessage.getPayload(), instanceOf(StompConnectionFailedEvent.class));
+
+		this.serverContext.refresh();
+
+		do {
+			eventMessage = this.stompEvents.receive(10000);
+			assertNotNull(eventMessage);
+		}
+		while (!(eventMessage.getPayload() instanceof StompSessionConnectedEvent));
+
+		waitForSubscribe("myTopic");
+
+		messagingTemplate = this.serverContext.getBean("brokerMessagingTemplate", SimpMessagingTemplate.class);
+		messagingTemplate.convertAndSend("/topic/myTopic", "foo");
+		receive = this.errorChannel.receive(10000);
+		assertNotNull(receive);
 	}
 
 	private void waitForSubscribe(String destination) throws InterruptedException {
@@ -225,6 +253,7 @@ public class StompInboundChannelAdapterWebSocketIntegrationTests {
 			WebSocketStompSessionManager webSocketStompSessionManager =
 					new WebSocketStompSessionManager(stompClient, server().getWsBaseUrl() + "/ws");
 			webSocketStompSessionManager.setAutoReceipt(true);
+			webSocketStompSessionManager.setRecoveryInterval(1000);
 			StompHeaders stompHeaders = new StompHeaders();
 			stompHeaders.setHeartbeat(new long[] {10000, 10000});
 			webSocketStompSessionManager.setConnectHeaders(stompHeaders);

spring-integration-stomp/src/test/java/org/springframework/integration/stomp/outbound/StompMessageHandlerWebSocketIntegrationTests.java

@@ -54,6 +54,7 @@ import org.springframework.integration.stomp.WebSocketStompSessionManager;
 import org.springframework.integration.stomp.event.StompExceptionEvent;
 import org.springframework.integration.stomp.event.StompIntegrationEvent;
 import org.springframework.integration.stomp.event.StompReceiptEvent;
+import org.springframework.integration.stomp.event.StompSessionConnectedEvent;
 import org.springframework.integration.test.support.LongRunningIntegrationTest;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.integration.websocket.TomcatWebSocketTestServer;
@@ -103,10 +104,6 @@ public class StompMessageHandlerWebSocketIntegrationTests {
 	@Value("#{server.serverContext}")
 	private ApplicationContext serverContext;
 
-	@Autowired
-	@Qualifier("stompMessageHandler")
-	private MessageHandler stompMessageHandler;
-
 	@Autowired
 	@Qualifier("webSocketOutputChannel")
 	private MessageChannel webSocketOutputChannel;
@@ -117,12 +114,6 @@ public class StompMessageHandlerWebSocketIntegrationTests {
 
 	@Test
 	public void testStompMessageHandler() throws InterruptedException {
-		int n = 0;
-		while (TestUtils.getPropertyValue(this.stompMessageHandler, "stompSession") == null && n++ < 100) {
-			Thread.sleep(100);
-		}
-		assertTrue(n < 100);
-
 		StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.SEND);
 		headers.setDestination("/app/simple");
 		Message<String> message = MessageBuilder.withPayload("foo").setHeaders(headers).build();
@@ -131,9 +122,13 @@ public class StompMessageHandlerWebSocketIntegrationTests {
 		SimpleController controller = this.serverContext.getBean(SimpleController.class);
 		assertTrue(controller.latch.await(10, TimeUnit.SECONDS));
 
-		// Simple Broker Relay doesn't support RECEIPT Frame, so we check here the 'lost' StompReceiptEvent
 		Message<?> receive = this.stompEvents.receive(10000);
 		assertNotNull(receive);
+		assertThat(receive.getPayload(), instanceOf(StompSessionConnectedEvent.class));
+
+		// Simple Broker Relay doesn't support RECEIPT Frame, so we check here the 'lost' StompReceiptEvent
+		receive = this.stompEvents.receive(10000);
+		assertNotNull(receive);
 		assertThat(receive.getPayload(), instanceOf(StompReceiptEvent.class));
 		StompReceiptEvent stompReceiptEvent = (StompReceiptEvent) receive.getPayload();
 		assertEquals(StompCommand.SEND, stompReceiptEvent.getStompCommand());
@@ -195,13 +190,16 @@ public class StompMessageHandlerWebSocketIntegrationTests {
 			WebSocketStompSessionManager webSocketStompSessionManager =
 					new WebSocketStompSessionManager(stompClient, server().getWsBaseUrl() + "/ws");
 			webSocketStompSessionManager.setAutoReceipt(true);
+			webSocketStompSessionManager.setRecoveryInterval(1000);
 			return webSocketStompSessionManager;
 		}
 
 		@Bean
 		@ServiceActivator(inputChannel = "webSocketOutputChannel")
 		public MessageHandler stompMessageHandler(StompSessionManager stompSessionManager) {
-			return new StompMessageHandler(stompSessionManager);
+			StompMessageHandler stompMessageHandler = new StompMessageHandler(stompSessionManager);
+			stompMessageHandler.setConnectTimeout(10000);
+			return stompMessageHandler;
 		}
 
 		@Bean

spring-integration-stomp/src/test/resources/log4j.properties

@@ -2,8 +2,8 @@ log4j.rootCategory=WARN, stdout
 
 log4j.appender.stdout=org.apache.log4j.ConsoleAppender
 log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
-log4j.appender.stdout.layout.ConversionPattern=%d %c{1} [%t] : %m%n
+log4j.appender.stdout.layout.ConversionPattern=%d{HH:mm:ss.SSS} %-5p [%t][%c] %m%n
 
-#log4j.category.org.springframework.messaging=DEBUG
+#log4j.category.org.springframework=DEBUG
 #log4j.category.org.springframework.integration=DEBUG
 log4j.category.org.springframework.integration.stomp=WARN

spring-integration-websocket/src/test/resources/log4j.properties

@@ -2,7 +2,7 @@ log4j.rootCategory=WARN, stdout
 
 log4j.appender.stdout=org.apache.log4j.ConsoleAppender
 log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
-log4j.appender.stdout.layout.ConversionPattern=%d %c{1} [%t] : %m%n
+log4j.appender.stdout.layout.ConversionPattern=%d{HH:mm:ss.SSS} %-5p [%t][%c] %m%n
 
 log4j.category.org.springframework.integration=WARN
 log4j.category.org.springframework.integration.websocket=WARN

src/reference/asciidoc/stomp.adoc

@@ -86,10 +86,10 @@ For more information, see the JavaDocs of those classes and the `mapped-headers`
 === STOMP Integration Events
 
 Many STOMP operations are asynchronous, including error handling.
-For example, STOMP has a `RECEIPT` server frame that is returned when a client frame has requested one by addding
+For example, STOMP has a `RECEIPT` server frame that is returned when a client frame has requested one by adding
 the `RECEIPT` header.
 To provide access to these asynchronous events, Spring Integration emits `StompIntegrationEvent` s which can be
-obtained by implementing an `ApplicationListener` or using an event inbound channel adapter.
+obtained by implementing an `ApplicationListener` or using an `<int-event:inbound-channel-adapter>` (see <<appevent-inbound>>).
 
 Specifically, a `StompExceptionEvent` is emitted from the `AbstractStompSessionManager`, when a
 `stompSessionListenableFuture` receives `onFailure()` in case of failure to connect to STOMP Broker.
@@ -316,7 +316,7 @@ mapped or provide your own `HeaderMapper` implementation using `header-mapper`.
 
 <7> Comma-separated list of STOMP destination names to subscribe.
 The list of destinations (and therefore subscriptions) can be modified at runtime
-through the `addDestination() and `removeDestination()` `@ManagedOperation` s.
+through the `addDestination()` and `removeDestination()` `@ManagedOperation` s.
 
 
 <8> Maximum amount of time in milliseconds to wait when sending a message to the channel if the channel may block.