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

Timeout Overrides in Messaging Templates

Browse Source 
Provide a mechanism to override the configured send and receive
timeouts in the GenericMessagingTemplate.

- overload `doReceive()` to take a receive timeout argument
- for `sendAndReceive()` methods examine message headers for these
  timeout values
  - remove headers to avoid propagation
  - avoid the unconditional rebuild of the reply message if the
    original headers weren't present
- also remove headers from simple `send()` operations
- change javadocs for the setters to indicate they are now defaults
- add properties to allow the user to override the header names used
- change `TemporaryReplyChannel` to use `send` arg and change to static
  - add package-protected ctor to avoid the compiler creating a
    synthetic constructor for access

Issue: SPR-15591
This commit is contained in:
Gary Russell
2017-05-26 10:58:56 -04:00
committed by Rossen Stoyanchev
parent 5dcfd84d3b
commit e06c4d5116
2 changed files with 265 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

173
spring-messaging/src/test/java/org/springframework/messaging/core/GenericMessagingTemplateTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2017 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.
@@ -23,7 +23,6 @@ import java.util.concurrent.atomic.AtomicReference;
import org.junit.Before;
import org.junit.Test;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageDeliveryException;
@@ -34,15 +33,19 @@ import org.springframework.messaging.StubMessageChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.support.ExecutorSubscribableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import static org.junit.Assert.*;
import static org.mockito.ArgumentMatchers.*;
import static org.mockito.Mockito.*;
/**
* Unit tests for {@link GenericMessagingTemplate}.
*
* @author Rossen Stoyanchev
* @author Gary Russell
*/
public class GenericMessagingTemplateTests {
@@ -63,6 +66,43 @@ public class GenericMessagingTemplateTests {
this.executor.afterPropertiesSet();
}
@Test
public void sendWithTimeout() {
SubscribableChannel channel = mock(SubscribableChannel.class);
final AtomicReference<Message<?>> sent = new AtomicReference<>();
doAnswer(invocation -> {
sent.set(invocation.getArgument(0));
return true;
}).when(channel).send(any(Message.class), eq(30_000L));
Message<?> message = MessageBuilder.withPayload("request")
.setHeader(GenericMessagingTemplate.DEFAULT_SEND_TIMEOUT_HEADER, 30_000L)
.setHeader(GenericMessagingTemplate.DEFAULT_RECEIVE_TIMEOUT_HEADER, 1L)
.build();
this.template.send(channel, message);
verify(channel).send(any(Message.class), eq(30_000L));
assertNotNull(sent.get());
assertFalse(sent.get().getHeaders().containsKey(GenericMessagingTemplate.DEFAULT_SEND_TIMEOUT_HEADER));
assertFalse(sent.get().getHeaders().containsKey(GenericMessagingTemplate.DEFAULT_RECEIVE_TIMEOUT_HEADER));
}
@Test
public void sendWithTimeoutMutable() {
SubscribableChannel channel = mock(SubscribableChannel.class);
final AtomicReference<Message<?>> sent = new AtomicReference<>();
doAnswer(invocation -> {
sent.set(invocation.getArgument(0));
return true;
}).when(channel).send(any(Message.class), eq(30_000L));
MessageHeaderAccessor accessor = new MessageHeaderAccessor();
accessor.setLeaveMutable(true);
Message<?> message = new GenericMessage<>("request", accessor.getMessageHeaders());
accessor.setHeader(GenericMessagingTemplate.DEFAULT_SEND_TIMEOUT_HEADER, 30_000L);
this.template.send(channel, message);
verify(channel).send(any(Message.class), eq(30_000L));
assertNotNull(sent.get());
assertFalse(sent.get().getHeaders().containsKey(GenericMessagingTemplate.DEFAULT_SEND_TIMEOUT_HEADER));
assertFalse(sent.get().getHeaders().containsKey(GenericMessagingTemplate.DEFAULT_RECEIVE_TIMEOUT_HEADER));
}
@Test
public void sendAndReceive() {
@@ -85,41 +125,118 @@ public class GenericMessagingTemplateTests {
final CountDownLatch latch = new CountDownLatch(1);
this.template.setReceiveTimeout(1);
this.template.setSendTimeout(30_000L);
this.template.setThrowExceptionOnLateReply(true);
SubscribableChannel channel = new ExecutorSubscribableChannel(this.executor);
channel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
try {
Thread.sleep(500);
MessageChannel replyChannel = (MessageChannel) message.getHeaders().getReplyChannel();
replyChannel.send(new GenericMessage<>("response"));
failure.set(new IllegalStateException("Expected exception"));
}
catch (InterruptedException e) {
failure.set(e);
}
catch (MessageDeliveryException ex) {
String expected = "Reply message received but the receiving thread has exited due to a timeout";
String actual = ex.getMessage();
if (!expected.equals(actual)) {
failure.set(new IllegalStateException("Unexpected error: '" + actual + "'"));
}
}
finally {
latch.countDown();
}
}
});
SubscribableChannel channel = mock(SubscribableChannel.class);
MessageHandler handler = createLateReplier(latch, failure);
doAnswer(invocation -> {
this.executor.execute(() -> {
handler.handleMessage(invocation.getArgument(0));
});
return true;
}).when(channel).send(any(Message.class), anyLong());
assertNull(this.template.convertSendAndReceive(channel, "request", String.class));
assertTrue(latch.await(1000, TimeUnit.MILLISECONDS));
assertTrue(latch.await(10_000, TimeUnit.MILLISECONDS));
Throwable ex = failure.get();
if (ex != null) {
throw new AssertionError(ex);
}
verify(channel).send(any(Message.class), eq(30_000L));
}
@Test
public void sendAndReceiveVariableTimeout() throws InterruptedException {
final AtomicReference<Throwable> failure = new AtomicReference<Throwable>();
final CountDownLatch latch = new CountDownLatch(1);
this.template.setSendTimeout(20_000);
this.template.setReceiveTimeout(10_000);
this.template.setThrowExceptionOnLateReply(true);
SubscribableChannel channel = mock(SubscribableChannel.class);
MessageHandler handler = createLateReplier(latch, failure);
doAnswer(invocation -> {
this.executor.execute(() -> {
handler.handleMessage(invocation.getArgument(0));
});
return true;
}).when(channel).send(any(Message.class), anyLong());
Message<?> message = MessageBuilder.withPayload("request")
.setHeader(GenericMessagingTemplate.DEFAULT_SEND_TIMEOUT_HEADER, 30_000L)
.setHeader(GenericMessagingTemplate.DEFAULT_RECEIVE_TIMEOUT_HEADER, 1L)
.build();
assertNull(this.template.sendAndReceive(channel, message));
assertTrue(latch.await(10_000, TimeUnit.MILLISECONDS));
Throwable ex = failure.get();
if (ex != null) {
throw new AssertionError(ex);
}
verify(channel).send(any(Message.class), eq(30_000L));
}
@Test
public void sendAndReceiveVariableTimeoutCustomHeaders() throws InterruptedException {
final AtomicReference<Throwable> failure = new AtomicReference<Throwable>();
final CountDownLatch latch = new CountDownLatch(1);
this.template.setSendTimeout(20_000);
this.template.setReceiveTimeout(10_000);
this.template.setThrowExceptionOnLateReply(true);
this.template.setSendTimeoutHeader("sto");
this.template.setReceiveTimeoutHeader("rto");
SubscribableChannel channel = mock(SubscribableChannel.class);
MessageHandler handler = createLateReplier(latch, failure);
doAnswer(invocation -> {
this.executor.execute(() -> {
handler.handleMessage(invocation.getArgument(0));
});
return true;
}).when(channel).send(any(Message.class), anyLong());
Message<?> message = MessageBuilder.withPayload("request")
.setHeader("sto", 30_000L)
.setHeader("rto", 1L)
.build();
assertNull(this.template.sendAndReceive(channel, message));
assertTrue(latch.await(10_000, TimeUnit.MILLISECONDS));
Throwable ex = failure.get();
if (ex != null) {
throw new AssertionError(ex);
}
verify(channel).send(any(Message.class), eq(30_000L));
}
private MessageHandler createLateReplier(final CountDownLatch latch, final AtomicReference<Throwable> failure) {
MessageHandler handler = message -> {
try {
Thread.sleep(500);
MessageChannel replyChannel = (MessageChannel) message.getHeaders().getReplyChannel();
replyChannel.send(new GenericMessage<>("response"));
failure.set(new IllegalStateException("Expected exception"));
}
catch (InterruptedException e) {
failure.set(e);
}
catch (MessageDeliveryException ex) {
String expected = "Reply message received but the receiving thread has exited due to a timeout";
String actual = ex.getMessage();
if (!expected.equals(actual)) {
failure.set(new IllegalStateException(
"Unexpected error: '" + actual + "'"));
}
}
finally {
latch.countDown();
}
};
return handler;
}
@Test