From b9730cd183416b0bf4eaf2a9ac9c59229abb4d90 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Sat, 7 Nov 2015 23:31:56 -0500 Subject: [PATCH] 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` --- .../main/resources/META-INF/spring.tooling | 8 +- ...dapter.gif => spring-integration-mqtt.gif} | Bin .../stomp/AbstractStompSessionManager.java | 189 +++++++++++++++--- .../event/StompConnectionFailedEvent.java | 32 +++ .../event/StompSessionConnectedEvent.java | 32 +++ .../inbound/StompInboundChannelAdapter.java | 21 +- .../stomp/outbound/StompMessageHandler.java | 87 +++++++- .../client/StompServerIntegrationTests.java | 66 +++++- .../config/StompAdaptersParserTests.java | 1 - ...annelAdapterWebSocketIntegrationTests.java | 33 ++- ...ssageHandlerWebSocketIntegrationTests.java | 22 +- .../src/test/resources/log4j.properties | 4 +- .../src/test/resources/log4j.properties | 2 +- src/reference/asciidoc/stomp.adoc | 6 +- 14 files changed, 426 insertions(+), 77 deletions(-) rename spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/xml/{spring-integration-mqttadapter.gif => spring-integration-mqtt.gif} (100%) create mode 100644 spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/StompConnectionFailedEvent.java create mode 100644 spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/StompSessionConnectedEvent.java diff --git a/spring-integration-mqtt/src/main/resources/META-INF/spring.tooling b/spring-integration-mqtt/src/main/resources/META-INF/spring.tooling index a062077f7e..a6ee10e101 100644 --- a/spring-integration-mqtt/src/main/resources/META-INF/spring.tooling +++ b/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 diff --git a/spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/xml/spring-integration-mqttadapter.gif b/spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/xml/spring-integration-mqtt.gif similarity index 100% rename from spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/xml/spring-integration-mqttadapter.gif rename to spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/xml/spring-integration-mqtt.gif 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 4ec268bff1..f1050b2220 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 @@ -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. *

- * The connection to the {@link StompSession} is made during {@link #afterPropertiesSet()}. + * The connection to the {@link StompSession} is made during {@link #start()}. *

- * The {@link #destroy()} lifecycle method manages {@link StompSession#disconnect()}. + * The {@link #stop()} lifecycle method manages {@link StompSession#disconnect()}. *

* 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() { @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() { - - @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() { - @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 doConnect(StompSessionHandler handler); - private static class CompositeStompSessionHandler extends StompSessionHandlerAdapter { + private class CompositeStompSessionHandler extends StompSessionHandlerAdapter { private final List delegates = Collections.synchronizedList(new ArrayList()); 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); diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/StompConnectionFailedEvent.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/StompConnectionFailedEvent.java new file mode 100644 index 0000000000..ed3ab21707 --- /dev/null +++ b/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); + } + +} diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/StompSessionConnectedEvent.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/StompSessionConnectedEvent.java new file mode 100644 index 0000000000..d75e484646 --- /dev/null +++ b/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); + } + +} diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java index 9f3b0b0aa6..8a92b0c60c 100644 --- a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java +++ b/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); } diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/outbound/StompMessageHandler.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/outbound/StompMessageHandler.java index a2a8537c0e..ec0871d730 100644 --- a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/outbound/StompMessageHandler.java +++ b/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 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 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); + } } } diff --git a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java index 21d7129088..13450ddcb1 100644 --- a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java +++ b/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("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("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; } diff --git a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/config/StompAdaptersParserTests.java b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/config/StompAdaptersParserTests.java index 238fabcc8b..8e944b3a82 100644 --- a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/config/StompAdaptersParserTests.java +++ b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/config/StompAdaptersParserTests.java @@ -164,7 +164,6 @@ public class StompAdaptersParserTests { List bars = lifecycles.get("bar"); bars.contains(this.customInboundAdapter); assertTrue(lifecycles.containsKey("foo")); - List foos = lifecycles.get("bar"); bars.contains(this.customOutboundAdapter); } diff --git a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapterWebSocketIntegrationTests.java b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapterWebSocketIntegrationTests.java index 31eb19ec0d..5e8b050de7 100644 --- a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapterWebSocketIntegrationTests.java +++ b/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); diff --git a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/outbound/StompMessageHandlerWebSocketIntegrationTests.java b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/outbound/StompMessageHandlerWebSocketIntegrationTests.java index fd6f59385c..1ac1cf2161 100644 --- a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/outbound/StompMessageHandlerWebSocketIntegrationTests.java +++ b/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 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 diff --git a/spring-integration-stomp/src/test/resources/log4j.properties b/spring-integration-stomp/src/test/resources/log4j.properties index 63821d90a0..87e6cad772 100644 --- a/spring-integration-stomp/src/test/resources/log4j.properties +++ b/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 diff --git a/spring-integration-websocket/src/test/resources/log4j.properties b/spring-integration-websocket/src/test/resources/log4j.properties index 7fcd10a721..7ab743934d 100644 --- a/spring-integration-websocket/src/test/resources/log4j.properties +++ b/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 diff --git a/src/reference/asciidoc/stomp.adoc b/src/reference/asciidoc/stomp.adoc index a1f0594ac0..003a74523f 100644 --- a/src/reference/asciidoc/stomp.adoc +++ b/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 `` (see <>). 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.