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

Add BackOffExecution to isolate state

Browse Source 
This commit separates the BackOff configuration from an actual
 execution. BackOffExecution now contains all the state of a
 particular execution and BackOff is only meant to start (i.e.
 create) a new execution.

 The method "reset" has been removed as its no longer necessary:
 when an execution does not need to be used for a given operation
 anymore it can be simply discarded.

 Issue: SPR-11746
This commit is contained in:
Stephane Nicoll
2014-05-09 16:39:01 +02:00
parent 49040a2925
commit 89fc3c0257
10 changed files with 270 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
spring-jms/src/main/java/org/springframework/jms/listener/DefaultMessageListenerContainer.java
View File

@@ -35,6 +35,7 @@ import org.springframework.scheduling.SchedulingAwareRunnable;
import org.springframework.scheduling.SchedulingTaskExecutor;
import org.springframework.util.Assert;
import org.springframework.util.BackOff;
import org.springframework.util.BackOffExecution;
import org.springframework.util.ClassUtils;
import org.springframework.util.FixedBackOff;
@@ -221,8 +222,8 @@ public class DefaultMessageListenerContainer extends AbstractPollingMessageListe
/**
* Specify the {@link BackOff} instance to use to compute the interval
* between recovery attempts. If the {@link BackOff} implementation
* returns {@link BackOff#STOP}, this listener container will not further
* between recovery attempts. If the {@link BackOffExecution} implementation
* returns {@link BackOffExecution#STOP}, this listener container will not further
* attempt to recover.
* <p>The {@link #setRecoveryInterval(long) recovery interval} is ignored
* when this property is set.
@@ -897,6 +898,7 @@ public class DefaultMessageListenerContainer extends AbstractPollingMessageListe
* @see #stop()
*/
protected void refreshConnectionUntilSuccessful() {
BackOffExecution execution = backOff.start();
while (isRunning()) {
try {
if (sharedConnectionEnabled()) {
@@ -907,7 +909,6 @@ public class DefaultMessageListenerContainer extends AbstractPollingMessageListe
JmsUtils.closeConnection(con);
}
logger.info("Successfully refreshed JMS Connection");
backOff.reset();
break;
}
catch (Exception ex) {
@@ -917,7 +918,7 @@ public class DefaultMessageListenerContainer extends AbstractPollingMessageListe
StringBuilder msg = new StringBuilder();
msg.append("Could not refresh JMS Connection for destination '");
msg.append(getDestinationDescription()).append("' - retrying using ");
msg.append(this.backOff).append(". Cause: ");
msg.append(execution).append(". Cause: ");
msg.append(ex instanceof JMSException ? JmsUtils.buildExceptionMessage((JMSException) ex) : ex.getMessage());
if (logger.isDebugEnabled()) {
logger.error(msg, ex);
@@ -926,7 +927,7 @@ public class DefaultMessageListenerContainer extends AbstractPollingMessageListe
logger.error(msg);
}
}
if (!applyBackOffTime()) {
if (!applyBackOffTime(execution)) {
stop();
}
}
@@ -952,13 +953,14 @@ public class DefaultMessageListenerContainer extends AbstractPollingMessageListe
}
/**
* Apply the next back off time. Return {@code true} if the back off period has
* been applied and a new attempt to recover should be made, {@code false} if no
* further attempt should be made.
* Apply the next back off time using the specified {@link BackOffExecution}.
* <p>Return {@code true} if the back off period has been applied and a new
* attempt to recover should be made, {@code false} if no further attempt
* should be made.
*/
protected boolean applyBackOffTime() {
long interval = backOff.nextBackOff();
if (interval == BackOff.STOP) {
protected boolean applyBackOffTime(BackOffExecution execution) {
long interval = execution.nextBackOff();
if (interval == BackOffExecution.STOP) {
return false;
}
else {

2
spring-jms/src/main/resources/org/springframework/jms/config/spring-jms-4.1.xsd
View File

@@ -322,7 +322,7 @@
<xsd:annotation>
<xsd:documentation><![CDATA[
Specify the BackOff instance to use to compute the interval between recovery
attempts. If the BackOff implementation returns "BackOff#STOP", the listener
attempts. If the BackOff implementation returns "BackOffExecution#STOP", the listener
container will not further attempt to recover. The recovery-interval value is
ignored when this property is set. The default is a FixedBackOff with an
interval of 5000 ms, that is 5 seconds.

23
spring-jms/src/test/java/org/springframework/jms/listener/DefaultMessageListenerContainerTests.java
View File

@@ -29,6 +29,7 @@ import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.util.BackOff;
import org.springframework.util.BackOffExecution;
/**
*
@@ -39,7 +40,9 @@ public class DefaultMessageListenerContainerTests {
@Test
public void applyBackOff() {
BackOff mock = mock(BackOff.class);
given(mock.nextBackOff()).willReturn(BackOff.STOP);
BackOffExecution execution = mock(BackOffExecution.class);
given(execution.nextBackOff()).willReturn(BackOffExecution.STOP);
given(mock.start()).willReturn(execution);
DefaultMessageListenerContainer container = createContainer(mock, createFailingContainerFactory());
container.start();
@@ -48,34 +51,40 @@ public class DefaultMessageListenerContainerTests {
container.refreshConnectionUntilSuccessful();
assertEquals(false, container.isRunning());
verify(mock).nextBackOff();
verify(mock).start();
verify(execution).nextBackOff();
}
@Test
public void applyBackOffRetry() {
BackOff mock = mock(BackOff.class);
given(mock.nextBackOff()).willReturn(50L, BackOff.STOP);
BackOffExecution execution = mock(BackOffExecution.class);
given(execution.nextBackOff()).willReturn(50L, BackOffExecution.STOP);
given(mock.start()).willReturn(execution);
DefaultMessageListenerContainer container = createContainer(mock, createFailingContainerFactory());
container.start();
container.refreshConnectionUntilSuccessful();
assertEquals(false, container.isRunning());
verify(mock, times(2)).nextBackOff();
verify(mock).start();
verify(execution, times(2)).nextBackOff();
}
@Test
public void recoverResetBackOff() {
BackOff mock = mock(BackOff.class);
given(mock.nextBackOff()).willReturn(50L, 50L, 50L); // 3 attempts max
BackOffExecution execution = mock(BackOffExecution.class);
given(execution.nextBackOff()).willReturn(50L, 50L, 50L); // 3 attempts max
given(mock.start()).willReturn(execution);
DefaultMessageListenerContainer container = createContainer(mock, createRecoverableContainerFactory(1));
container.start();
container.refreshConnectionUntilSuccessful();
assertEquals(true, container.isRunning());
verify(mock, times(1)).nextBackOff(); // only on attempt as the second one lead to a recovery
verify(mock, times(1)).reset(); // reset should have been called
verify(mock).start();
verify(execution, times(1)).nextBackOff(); // only on attempt as the second one lead to a recovery
}
@SuppressWarnings("unchecked")