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

INT-3895: Fix MonitorTests Race Condition

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3895

* `@Ignore` `DelayerUsageTests.testDelayWithCustomScheduler()` as very weak test. (We can consider it to remove at all: doesn't test anything from our side)
* Add `LogAdjustingTestSupport` diagnostic to the `OutboundGatewayFunctionTests`

Tentative commit to see more logs from Travis

Additional diagnostics

More STOMP Diagnostics
This commit is contained in:
Artem Bilan
2015-11-20 12:25:32 -05:00
committed by Gary Russell
parent fc0af2fed6
commit 1cc1b2ef79
5 changed files with 87 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -20,7 +20,10 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -39,6 +42,7 @@ import org.springframework.messaging.simp.stomp.StompSession;
import org.springframework.messaging.simp.stomp.StompSessionHandler;
import org.springframework.messaging.simp.stomp.StompSessionHandlerAdapter;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
@@ -72,6 +76,8 @@ public abstract class AbstractStompSessionManager implements StompSessionManager
protected final StompClientSupport stompClient;
private final AtomicInteger epoch = new AtomicInteger();
private boolean autoStartup = false;
private boolean running = false;
@@ -164,18 +170,43 @@ public abstract class AbstractStompSessionManager implements StompSessionManager
return this.phase;
}
private void connect() {
private synchronized void connect() {
if (this.connecting || this.connected) {
if (logger.isDebugEnabled()) {
logger.debug("Aborting connect; another thread is connecting.");
}
return;
}
final int epoch = this.epoch.get();
this.connecting = true;
this.stompSessionListenableFuture = doConnect(this.compositeStompSessionHandler);
if (logger.isDebugEnabled()) {
logger.debug("Connecting " + this);
}
try {
this.stompSessionListenableFuture = doConnect(this.compositeStompSessionHandler);
}
catch (Exception e) {
logger.error("doConnect() error for " + this, e);
}
final CountDownLatch latch = new CountDownLatch(1);
this.stompSessionListenableFuture.addCallback(new ListenableFutureCallback<StompSession>() {
@Override
public void onFailure(Throwable e) {
scheduleReconnect(e);
if (logger.isDebugEnabled()) {
logger.debug("onFailure", e);
}
latch.countDown();
if (epoch == AbstractStompSessionManager.this.epoch.get()) {
scheduleReconnect(e);
}
}
@Override
public void onSuccess(StompSession stompSession) {
if (logger.isDebugEnabled()) {
logger.debug("onSuccess");
}
AbstractStompSessionManager.this.connected = true;
AbstractStompSessionManager.this.connecting = false;
stompSession.setAutoReceipt(isAutoReceiptEnabled());
@@ -184,14 +215,28 @@ public abstract class AbstractStompSessionManager implements StompSessionManager
new StompSessionConnectedEvent(this));
}
AbstractStompSessionManager.this.reconnectFuture = null;
latch.countDown();
}
});
try {
if (!latch.await(10, TimeUnit.SECONDS)) {
logger.error("No response to connection attempt");
if (epoch == this.epoch.get()) {
scheduleReconnect(null);
}
}
}
catch (InterruptedException e1) {
logger.error("Interrupted while waiting for connection attempt");
Thread.currentThread().interrupt();
}
}
private void scheduleReconnect(Throwable e) {
this.epoch.incrementAndGet();
this.connecting = this.connected = false;
logger.error("STOMP connect error.", e);
logger.error("STOMP connect error for " + this, e);
if (this.applicationEventPublisher != null) {
this.applicationEventPublisher.publishEvent(
new StompConnectionFailedEvent(this, e));
@@ -297,8 +342,9 @@ public abstract class AbstractStompSessionManager implements StompSessionManager
@Override
public String toString() {
return "StompSessionManager{" +
"connected=" + connected +
return ObjectUtils.identityToString(this) +
" {connecting=" + connecting +
", connected=" + connected +
", name='" + name + '\'' +
'}';
}