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

INT-4244: Optimize JpaOutboundGateway for Message

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

The `JpaOutboundGateway` returns a whole `Message` for reply copying request headers.
Building the reply a `AbstractMessageProducingHandler` uses `MessageBuilder` to create a new `Message` with all request headers because of `shouldCopyRequestHeaders() == true`

* To avoid some overhead with new `Message` and `MessageBuilder` for nothing, just return a JPA result as is and let super class to build the Message

* Optimize all reply producers to return `MessageBuilder` to the super `handleMessage()` process
* Return the whole `gatherResult` `Message<?>` from the `ScatterGatherHandler` to have ability to transfer reply headers as well
* Mark `DelayHandler`  as `shouldCopyRequestHeaders() return false` since it doesn't modify Message and returns it as is after delay
This commit is contained in:
Artem Bilan
2017-03-16 10:10:01 -04:00
committed by Gary Russell
parent 752fe0356f
commit df9593a528
13 changed files with 120 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2016-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.
@@ -42,6 +42,8 @@ import org.springframework.util.StringUtils;
/**
* @author Gary Russell
* @author Artem Bilan
*
* @since 4.3
*
*/
@@ -421,7 +423,7 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin
}
}
protected Message<?> buildReplyMessage(MessageConverter converter,
protected AbstractIntegrationMessageBuilder<?> buildReply(MessageConverter converter,
org.springframework.amqp.core.Message amqpReplyMessage) {
Object replyObject = converter.fromMessage(amqpReplyMessage);
AbstractIntegrationMessageBuilder<?> builder = (replyObject instanceof Message)
@@ -429,7 +431,7 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin
: this.getMessageBuilderFactory().withPayload(replyObject);
Map<String, ?> headers = getHeaderMapper().toHeadersFromReply(amqpReplyMessage.getMessageProperties());
builder.copyHeadersIfAbsent(headers);
return builder.build();
return builder;
}
protected Message<?> buildReturnedMessage(org.springframework.amqp.core.Message message,

10
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.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,6 +23,7 @@ import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.Lifecycle;
import org.springframework.integration.amqp.support.MappingUtils;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
@@ -33,6 +34,7 @@ import org.springframework.util.Assert;
* @author Oleg Zhurakousky
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.1
*/
public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint
@@ -114,8 +116,8 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint
}
}
private Message<?> sendAndReceive(String exchangeName, String routingKey, Message<?> requestMessage,
CorrelationData correlationData) {
private AbstractIntegrationMessageBuilder<?> sendAndReceive(String exchangeName, String routingKey,
Message<?> requestMessage, CorrelationData correlationData) {
Assert.isInstanceOf(RabbitTemplate.class, this.amqpTemplate,
"RabbitTemplate implementation is required for publisher confirms");
MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter();
@@ -129,7 +131,7 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint
if (amqpReplyMessage == null) {
return null;
}
return buildReplyMessage(converter, amqpReplyMessage);
return buildReply(converter, amqpReplyMessage);
}
@Override

14
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java
View File

@@ -24,6 +24,7 @@ import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.integration.amqp.support.MappingUtils;
import org.springframework.integration.handler.ReplyRequiredException;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessagingException;
@@ -35,6 +36,8 @@ import org.springframework.util.concurrent.ListenableFutureCallback;
* is sent on the async template's listener container thread.
*
* @author Gary Russell
* @author Artem Bilan
*
* @since 4.3
*
*/
@@ -95,17 +98,18 @@ public class AsyncAmqpOutboundGateway extends AbstractAmqpOutboundEndpoint {
@Override
public void onSuccess(org.springframework.amqp.core.Message result) {
Message<?> replyMessage = null;
AbstractIntegrationMessageBuilder<?> replyMessageBuilder = null;
try {
replyMessage = buildReplyMessage(AsyncAmqpOutboundGateway.this.messageConverter, result);
sendOutputs(replyMessage, this.requestMessage);
replyMessageBuilder = buildReply(AsyncAmqpOutboundGateway.this.messageConverter, result);
sendOutputs(replyMessageBuilder, this.requestMessage);
}
catch (Exception e) {
Exception exceptionToLogAndSend = e;
if (!(e instanceof MessagingException)) {
exceptionToLogAndSend = new MessageHandlingException(this.requestMessage, e);
if (replyMessage != null) {
exceptionToLogAndSend = new MessagingException(replyMessage, exceptionToLogAndSend);
if (replyMessageBuilder != null) {
exceptionToLogAndSend =
new MessagingException(replyMessageBuilder.build(), exceptionToLogAndSend);
}
}
logger.error("Failed to send async reply: " + result.toString(), exceptionToLogAndSend);

11
spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java
View File

@@ -75,6 +75,7 @@ import org.springframework.util.CollectionUtils;
* @author Mark Fisher
* @author Artem Bilan
* @author Gary Russell
*
* @since 1.0.3
*/
@@ -234,6 +235,11 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
return releaseHandler;
}
@Override
protected boolean shouldCopyRequestHeaders() {
return false;
}
/**
* Checks if 'requestMessage' wasn't delayed before
* ({@link #releaseMessageAfterDelay} and {@link DelayHandler.DelayedMessageWrapper}).
@@ -243,10 +249,8 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
* @param requestMessage - the Message which may be delayed.
* @return - {@code null} if 'requestMessage' is delayed,
* otherwise - 'payload' from 'requestMessage'.
*
* @see #releaseMessage
*/
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
boolean delayed = requestMessage.getPayload() instanceof DelayedMessageWrapper;
@@ -260,8 +264,7 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
}
// no delay
Object payload = requestMessage.getPayload();
return delayed ? ((DelayedMessageWrapper) payload).getOriginal().getPayload() : payload;
return delayed ? ((DelayedMessageWrapper) requestMessage.getPayload()).getOriginal() : requestMessage;
}
private long determineDelayForMessage(Message<?> message) {

2
spring-integration-core/src/main/java/org/springframework/integration/scattergather/ScatterGatherHandler.java
View File

@@ -145,7 +145,7 @@ public class ScatterGatherHandler extends AbstractReplyProducingMessageHandler i
Message<?> gatherResult = gatherResultChannel.receive(this.gatherTimeout);
if (gatherResult != null) {
return gatherResult.getPayload();
return gatherResult;
}
return null;

6
spring-integration-core/src/main/java/org/springframework/integration/transformer/ContentEnricher.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.
@@ -417,7 +417,9 @@ public class ContentEnricher extends AbstractReplyProducingMessageHandler
targetHeaders.put(header, value);
}
}
return this.getMessageBuilderFactory().withPayload(targetPayload).copyHeaders(targetHeaders).build();
return getMessageBuilderFactory()
.withPayload(targetPayload)
.copyHeaders(targetHeaders);
}
}

32
spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java
View File

@@ -566,9 +566,9 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
final String fullDir = dir;
List<?> payload = this.remoteFileTemplate.execute(session ->
AbstractRemoteFileOutboundGateway.this.ls(session, fullDir));
return this.getMessageBuilderFactory().withPayload(payload)
.setHeader(FileHeaders.REMOTE_DIRECTORY, dir)
.build();
return getMessageBuilderFactory()
.withPayload(payload)
.setHeader(FileHeaders.REMOTE_DIRECTORY, dir);
}
private Object doGet(final Message<?> requestMessage) {
@@ -592,11 +592,11 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
payload = this.remoteFileTemplate.execute(session1 ->
get(requestMessage, session1, remoteDir, remoteFilePath, remoteFilename, null));
}
return getMessageBuilderFactory().withPayload(payload)
return getMessageBuilderFactory()
.withPayload(payload)
.setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir)
.setHeader(FileHeaders.REMOTE_FILE, remoteFilename)
.setHeader(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE, session)
.build();
.setHeader(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE, session);
}
private Object doMget(final Message<?> requestMessage) {
@@ -605,10 +605,10 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
final String remoteDir = getRemoteDirectory(remoteFilePath, remoteFilename);
List<File> payload = this.remoteFileTemplate.execute(session ->
mGet(requestMessage, session, remoteDir, remoteFilename));
return this.getMessageBuilderFactory().withPayload(payload)
return getMessageBuilderFactory()
.withPayload(payload)
.setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir)
.setHeader(FileHeaders.REMOTE_FILE, remoteFilename)
.build();
.setHeader(FileHeaders.REMOTE_FILE, remoteFilename);
}
private Object doRm(Message<?> requestMessage) {
@@ -618,10 +618,10 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
boolean payload = this.remoteFileTemplate.remove(remoteFilePath);
return this.getMessageBuilderFactory().withPayload(payload)
return getMessageBuilderFactory()
.withPayload(payload)
.setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir)
.setHeader(FileHeaders.REMOTE_FILE, remoteFilename)
.build();
.setHeader(FileHeaders.REMOTE_FILE, remoteFilename);
}
private Object doMv(Message<?> requestMessage) {
@@ -632,15 +632,15 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
Assert.hasLength(remoteFileNewPath, "New filename cannot be empty");
this.remoteFileTemplate.rename(remoteFilePath, remoteFileNewPath);
return this.getMessageBuilderFactory().withPayload(Boolean.TRUE)
return getMessageBuilderFactory()
.withPayload(Boolean.TRUE)
.setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir)
.setHeader(FileHeaders.REMOTE_FILE, remoteFilename)
.setHeader(FileHeaders.RENAME_TO, remoteFileNewPath)
.build();
.setHeader(FileHeaders.RENAME_TO, remoteFileNewPath);
}
private String doPut(Message<?> requestMessage) {
return this.doPut(requestMessage, null);
return doPut(requestMessage, null);
}
private String doPut(Message<?> requestMessage, String subDirectory) {

81
spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java
View File

@@ -77,6 +77,7 @@ import org.springframework.messaging.support.GenericMessage;
* @author Gary Russell
* @author Liu Jiong
* @author Artem Bilan
*
* @since 2.1
*/
@SuppressWarnings("rawtypes")
@@ -135,8 +136,8 @@ public class RemoteFileOutboundGatewayTests {
TestLsEntry[] files = fileList();
when(session.list("testremote/x/")).thenReturn(files);
@SuppressWarnings("unchecked")
Message<List<TestLsEntry>> out = (Message<List<TestLsEntry>>) gw
.handleRequestMessage(new GenericMessage<String>("testremote/x"));
MessageBuilder<List<TestLsEntry>> out = (MessageBuilder<List<TestLsEntry>>) gw
.handleRequestMessage(new GenericMessage<>("testremote/x"));
assertEquals(2, out.getPayload().size());
assertSame(files[1], out.getPayload().get(0)); // sort by default
assertSame(files[0], out.getPayload().get(1));
@@ -191,8 +192,8 @@ public class RemoteFileOutboundGatewayTests {
});
@SuppressWarnings("unchecked")
Message<List<File>> out = (Message<List<File>>) gw
.handleRequestMessage(new GenericMessage<String>("testremote/*"));
MessageBuilder<List<File>> out = (MessageBuilder<List<File>>) gw
.handleRequestMessage(new GenericMessage<>("testremote/*"));
assertEquals(2, out.getPayload().size());
assertEquals("f1", out.getPayload().get(0).getName());
assertEquals("f2", out.getPayload().get(1).getName());
@@ -222,8 +223,8 @@ public class RemoteFileOutboundGatewayTests {
});
@SuppressWarnings("unchecked")
Message<List<File>> out = (Message<List<File>>) gw
.handleRequestMessage(new GenericMessage<String>("testremote/f1"));
MessageBuilder<List<File>> out = (MessageBuilder<List<File>>) gw
.handleRequestMessage(new GenericMessage<>("testremote/f1"));
assertEquals(1, out.getPayload().size());
assertEquals("f1", out.getPayload().get(0).getName());
assertEquals("testremote/",
@@ -267,7 +268,7 @@ public class RemoteFileOutboundGatewayTests {
Message<String> requestMessage = MessageBuilder.withPayload("foo")
.setHeader(FileHeaders.RENAME_TO, "bar")
.build();
Message<?> out = (Message<?>) gw.handleRequestMessage(requestMessage);
MessageBuilder<?> out = (MessageBuilder<?>) gw.handleRequestMessage(requestMessage);
assertEquals("foo", out.getHeaders().get(FileHeaders.REMOTE_FILE));
assertEquals("foobar", args.get());
assertEquals(Boolean.TRUE, out.getPayload());
@@ -283,11 +284,11 @@ public class RemoteFileOutboundGatewayTests {
final AtomicReference<String> args = new AtomicReference<String>();
doAnswer(invocation -> {
Object[] arguments = invocation.getArguments();
args.set((String) arguments[0] + (String) arguments[1]);
args.set((String) arguments[0] + arguments[1]);
return null;
}).when(session).rename(anyString(), anyString());
when(sessionFactory.getSession()).thenReturn(session);
Message<?> out = (Message<?>) gw.handleRequestMessage(new GenericMessage<String>("foo"));
MessageBuilder<?> out = (MessageBuilder<?>) gw.handleRequestMessage(new GenericMessage<>("foo"));
assertEquals("oo", out.getHeaders().get(FileHeaders.RENAME_TO));
assertEquals("foo", out.getHeaders().get(FileHeaders.REMOTE_FILE));
assertEquals("foooo", args.get());
@@ -316,7 +317,7 @@ public class RemoteFileOutboundGatewayTests {
Message<String> requestMessage = MessageBuilder.withPayload("foo")
.setHeader(FileHeaders.RENAME_TO, "bar")
.build();
Message<?> out = (Message<?>) gw.handleRequestMessage(requestMessage);
MessageBuilder<?> out = (MessageBuilder<?>) gw.handleRequestMessage(requestMessage);
assertEquals("foo", out.getHeaders().get(FileHeaders.REMOTE_FILE));
assertEquals("foofoo/bar/baz", args.get());
assertEquals(Boolean.TRUE, out.getPayload());
@@ -347,8 +348,8 @@ public class RemoteFileOutboundGatewayTests {
TestLsEntry[] files = fileList();
when(session.list("testremote/x/")).thenReturn(files);
@SuppressWarnings("unchecked")
Message<List<TestLsEntry>> out = (Message<List<TestLsEntry>>) gw
.handleRequestMessage(new GenericMessage<String>("testremote/x"));
MessageBuilder<List<TestLsEntry>> out = (MessageBuilder<List<TestLsEntry>>) gw
.handleRequestMessage(new GenericMessage<>("testremote/x"));
assertEquals(2, out.getPayload().size());
assertSame(files[0], out.getPayload().get(0));
assertSame(files[1], out.getPayload().get(1));
@@ -392,8 +393,8 @@ public class RemoteFileOutboundGatewayTests {
when(session.list("testremote/x/d1/")).thenReturn(level2);
when(session.list("testremote/x/d1/d2/")).thenReturn(level3);
@SuppressWarnings("unchecked")
Message<List<TestLsEntry>> out = (Message<List<TestLsEntry>>) gw
.handleRequestMessage(new GenericMessage<String>("testremote/x"));
MessageBuilder<List<TestLsEntry>> out = (MessageBuilder<List<TestLsEntry>>) gw
.handleRequestMessage(new GenericMessage<>("testremote/x"));
assertEquals(4, out.getPayload().size());
assertEquals("f1", out.getPayload().get(0).getFilename());
assertEquals("d1/d2/f4", out.getPayload().get(1).getFilename());
@@ -418,8 +419,8 @@ public class RemoteFileOutboundGatewayTests {
when(session.list("testremote/x/d1/")).thenReturn(level2);
when(session.list("testremote/x/d1/d2/")).thenReturn(level3);
@SuppressWarnings("unchecked")
Message<List<TestLsEntry>> out = (Message<List<TestLsEntry>>) gw
.handleRequestMessage(new GenericMessage<String>("testremote/x"));
MessageBuilder<List<TestLsEntry>> out = (MessageBuilder<List<TestLsEntry>>) gw
.handleRequestMessage(new GenericMessage<>("testremote/x"));
assertEquals(6, out.getPayload().size());
assertEquals("f1", out.getPayload().get(0).getFilename());
assertEquals("d1", out.getPayload().get(1).getFilename());
@@ -441,8 +442,8 @@ public class RemoteFileOutboundGatewayTests {
TestLsEntry[] files = new TestLsEntry[0];
when(session.list("testremote/")).thenReturn(files);
@SuppressWarnings("unchecked")
Message<List<TestLsEntry>> out = (Message<List<TestLsEntry>>) gw
.handleRequestMessage(new GenericMessage<String>("testremote"));
MessageBuilder<List<TestLsEntry>> out = (MessageBuilder<List<TestLsEntry>>) gw
.handleRequestMessage(new GenericMessage<>("testremote"));
assertEquals(0, out.getPayload().size());
}
@@ -457,8 +458,8 @@ public class RemoteFileOutboundGatewayTests {
TestLsEntry[] files = fileList();
when(session.list("testremote/")).thenReturn(files);
@SuppressWarnings("unchecked")
Message<List<String>> out = (Message<List<String>>) gw
.handleRequestMessage(new GenericMessage<String>("testremote"));
MessageBuilder<List<String>> out = (MessageBuilder<List<String>>) gw
.handleRequestMessage(new GenericMessage<>("testremote"));
assertEquals(2, out.getPayload().size());
assertEquals("f1", out.getPayload().get(0));
assertEquals("f2", out.getPayload().get(1));
@@ -475,8 +476,8 @@ public class RemoteFileOutboundGatewayTests {
TestLsEntry[] files = fileList();
when(session.list("testremote/")).thenReturn(files);
@SuppressWarnings("unchecked")
Message<List<String>> out = (Message<List<String>>) gw
.handleRequestMessage(new GenericMessage<String>("testremote"));
MessageBuilder<List<String>> out = (MessageBuilder<List<String>>) gw
.handleRequestMessage(new GenericMessage<>("testremote"));
assertEquals(2, out.getPayload().size());
assertEquals("f2", out.getPayload().get(0));
assertEquals("f1", out.getPayload().get(1));
@@ -493,8 +494,8 @@ public class RemoteFileOutboundGatewayTests {
TestLsEntry[] files = fileList();
when(session.list("testremote/")).thenReturn(files);
@SuppressWarnings("unchecked")
Message<List<String>> out = (Message<List<String>>) gw
.handleRequestMessage(new GenericMessage<String>("testremote"));
MessageBuilder<List<String>> out = (MessageBuilder<List<String>>) gw
.handleRequestMessage(new GenericMessage<>("testremote"));
assertEquals(3, out.getPayload().size());
assertEquals("f1", out.getPayload().get(0));
assertEquals("f2", out.getPayload().get(1));
@@ -512,8 +513,8 @@ public class RemoteFileOutboundGatewayTests {
TestLsEntry[] files = fileList();
when(session.list("testremote/")).thenReturn(files);
@SuppressWarnings("unchecked")
Message<List<String>> out = (Message<List<String>>) gw
.handleRequestMessage(new GenericMessage<String>("testremote"));
MessageBuilder<List<String>> out = (MessageBuilder<List<String>>) gw
.handleRequestMessage(new GenericMessage<>("testremote"));
assertEquals(4, out.getPayload().size());
assertEquals("f1", out.getPayload().get(0));
assertEquals("f2", out.getPayload().get(1));
@@ -532,8 +533,8 @@ public class RemoteFileOutboundGatewayTests {
TestLsEntry[] files = fileList();
when(session.list("testremote/")).thenReturn(files);
@SuppressWarnings("unchecked")
Message<List<String>> out = (Message<List<String>>) gw
.handleRequestMessage(new GenericMessage<String>("testremote"));
MessageBuilder<List<String>> out = (MessageBuilder<List<String>>) gw
.handleRequestMessage(new GenericMessage<>("testremote"));
assertEquals(6, out.getPayload().size());
assertEquals("f2", out.getPayload().get(0));
assertEquals("f1", out.getPayload().get(1));
@@ -555,8 +556,8 @@ public class RemoteFileOutboundGatewayTests {
TestLsEntry[] files = fileList();
when(session.list("testremote/")).thenReturn(files);
@SuppressWarnings("unchecked")
Message<List<String>> out = (Message<List<String>>) gw
.handleRequestMessage(new GenericMessage<String>("testremote"));
MessageBuilder<List<String>> out = (MessageBuilder<List<String>>) gw
.handleRequestMessage(new GenericMessage<>("testremote"));
assertEquals(1, out.getPayload().size());
assertEquals("f4", out.getPayload().get(0));
}
@@ -585,7 +586,7 @@ public class RemoteFileOutboundGatewayTests {
});
@SuppressWarnings("unchecked")
Message<File> out = (Message<File>) gw.handleRequestMessage(new GenericMessage<String>("f1"));
MessageBuilder<File> out = (MessageBuilder<File>) gw.handleRequestMessage(new GenericMessage<>("f1"));
File outFile = new File(this.tmpDir + "/f1");
assertEquals(outFile, out.getPayload());
assertTrue(outFile.exists());
@@ -623,9 +624,9 @@ public class RemoteFileOutboundGatewayTests {
});
// default (null)
Message<File> out;
MessageBuilder<File> out;
try {
out = (Message<File>) gw.handleRequestMessage(new GenericMessage<String>("f1"));
out = (MessageBuilder<File>) gw.handleRequestMessage(new GenericMessage<>("f1"));
fail("Exception expected");
}
catch (MessageHandlingException e) {
@@ -634,7 +635,7 @@ public class RemoteFileOutboundGatewayTests {
gw.setFileExistsMode(FileExistsMode.FAIL);
try {
out = (Message<File>) gw.handleRequestMessage(new GenericMessage<String>("f1"));
out = (MessageBuilder<File>) gw.handleRequestMessage(new GenericMessage<>("f1"));
fail("Exception expected");
}
catch (MessageHandlingException e) {
@@ -642,17 +643,17 @@ public class RemoteFileOutboundGatewayTests {
}
gw.setFileExistsMode(FileExistsMode.IGNORE);
out = (Message<File>) gw.handleRequestMessage(new GenericMessage<String>("f1"));
out = (MessageBuilder<File>) gw.handleRequestMessage(new GenericMessage<>("f1"));
assertEquals(outFile, out.getPayload());
assertContents("foo", outFile);
gw.setFileExistsMode(FileExistsMode.APPEND);
out = (Message<File>) gw.handleRequestMessage(new GenericMessage<String>("f1"));
out = (MessageBuilder<File>) gw.handleRequestMessage(new GenericMessage<>("f1"));
assertEquals(outFile, out.getPayload());
assertContents("footestfile", outFile);
gw.setFileExistsMode(FileExistsMode.REPLACE);
out = (Message<File>) gw.handleRequestMessage(new GenericMessage<String>("f1"));
out = (MessageBuilder<File>) gw.handleRequestMessage(new GenericMessage<>("f1"));
assertEquals(outFile, out.getPayload());
assertContents("testfile", outFile);
@@ -730,7 +731,7 @@ public class RemoteFileOutboundGatewayTests {
});
@SuppressWarnings("unchecked")
Message<File> out = (Message<File>) gw.handleRequestMessage(new GenericMessage<String>("x/f1"));
MessageBuilder<File> out = (MessageBuilder<File>) gw.handleRequestMessage(new GenericMessage<>("x/f1"));
File outFile = new File(this.tmpDir + "/f1");
assertEquals(outFile, out.getPayload());
assertTrue(outFile.exists());
@@ -781,8 +782,8 @@ public class RemoteFileOutboundGatewayTests {
when(sessionFactory.getSession()).thenReturn(session);
when(session.remove("testremote/x/f1")).thenReturn(Boolean.TRUE);
@SuppressWarnings("unchecked")
Message<Boolean> out = (Message<Boolean>) gw
.handleRequestMessage(new GenericMessage<String>("testremote/x/f1"));
MessageBuilder<Boolean> out = (MessageBuilder<Boolean>) gw
.handleRequestMessage(new GenericMessage<>("testremote/x/f1"));
assertEquals(Boolean.TRUE, out.getPayload());
verify(session).remove("testremote/x/f1");
assertEquals("testremote/x/",

7
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcOutboundGateway.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.
@@ -44,7 +44,8 @@ public class JdbcOutboundGateway extends AbstractReplyProducingMessageHandler im
private final JdbcPollingChannelAdapter poller;
private volatile SqlParameterSourceFactory sqlParameterSourceFactory = new ExpressionEvaluatingSqlParameterSourceFactory();
private volatile SqlParameterSourceFactory sqlParameterSourceFactory =
new ExpressionEvaluatingSqlParameterSourceFactory();
private volatile boolean sqlParameterSourceFactorySet;
@@ -167,7 +168,7 @@ public class JdbcOutboundGateway extends AbstractReplyProducingMessageHandler im
if (list.size() == 1) {
payload = list.get(0);
}
return this.getMessageBuilderFactory().withPayload(payload).copyHeaders(requestMessage.getHeaders()).build();
return payload;
}
/**

21
spring-integration-jpa/src/main/java/org/springframework/integration/jpa/outbound/JpaOutboundGateway.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.
@@ -39,21 +39,22 @@ import org.springframework.util.Assert;
*
* @author Gunnar Hillert
* @author Amol Nayak
* @author Artem Bilan
*
* @since 2.2
*
*/
public class JpaOutboundGateway extends AbstractReplyProducingMessageHandler {
private final JpaExecutor jpaExecutor;
private final JpaExecutor jpaExecutor;
private OutboundGatewayType gatewayType = OutboundGatewayType.UPDATING;
private boolean producesReply = true; //false for outbound-channel-adapter, true for outbound-gateway
private boolean producesReply = true; //false for outbound-channel-adapter, true for outbound-gateway
/**
* Constructor taking an {@link JpaExecutor} that wraps all JPA Operations.
*
* @param jpaExecutor Must not be null
*
*/
public JpaOutboundGateway(JpaExecutor jpaExecutor) {
Assert.notNull(jpaExecutor, "jpaExecutor must not be null.");
@@ -81,19 +82,18 @@ public class JpaOutboundGateway extends AbstractReplyProducingMessageHandler {
result = this.jpaExecutor.executeOutboundJpaOperation(requestMessage);
}
else {
throw new IllegalArgumentException(String.format("GatewayType '%s' is not supported.", this.gatewayType));
throw new IllegalArgumentException(String.format("GatewayType '%s' is not supported.", this.gatewayType));
}
if (result == null || !this.producesReply) {
return null;
}
return this.getMessageBuilderFactory().withPayload(result).copyHeaders(requestMessage.getHeaders()).build();
return result;
}
/**
*
* Specify the {@link JpaOutboundGateway} mode.
* @param gatewayType The gateway type.
*/
public void setGatewayType(OutboundGatewayType gatewayType) {
@@ -104,11 +104,10 @@ public class JpaOutboundGateway extends AbstractReplyProducingMessageHandler {
/**
* If set to 'false', this component will act as an Outbound Channel Adapter.
* If not explicitly set this property will default to 'true'.
*
* @param producesReply Defaults to 'true'.
*
*/
public void setProducesReply(boolean producesReply) {
this.producesReply = producesReply;
}
}

5
spring-integration-jpa/src/test/java/org/springframework/integration/jpa/outbound/JpaOutboundGatewayIntegrationTests.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.
@@ -36,6 +36,7 @@ import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.handler.ReplyRequiredException;
import org.springframework.integration.jpa.test.entity.StudentDomain;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.matcher.HeaderMatcher;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHandlingException;
@@ -52,6 +53,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
* @author Amol Nayak
* @author Artem Bilan
* @author Gary Russell
*
* @since 3.0
*
*/
@@ -110,6 +112,7 @@ public class JpaOutboundGatewayIntegrationTests {
this.handler = message -> {
assertEquals(2, ((List<?>) message.getPayload()).size());
assertEquals(1, entityManager.createQuery("from Student").getResultList().size());
assertThat(message, HeaderMatcher.hasHeader("maxResults", "10"));
};
this.responseChannel.subscribe(this.handler);

6
spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundGateway.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2016 the original author or authors.
* Copyright 2014-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.
@@ -33,6 +33,7 @@ import org.springframework.util.IdGenerator;
/**
* @author David Liu
* @author Artem Bilan
*
* @since 4.1
*/
public class RedisQueueOutboundGateway extends AbstractReplyProducingMessageHandler {
@@ -117,7 +118,8 @@ public class RedisQueueOutboundGateway extends AbstractReplyProducingMessageHand
return null;
}
if (this.extractPayload) {
return this.getMessageBuilderFactory().withPayload(replyMessage).build();
return getMessageBuilderFactory()
.withPayload(replyMessage);
}
else {
return replyMessage;

26
spring-integration-rmi/src/main/java/org/springframework/integration/rmi/RmiOutboundGateway.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.
@@ -32,6 +32,7 @@ import org.springframework.remoting.rmi.RmiProxyFactoryBean;
*
* @author Mark Fisher
* @author Gary Russell
* @author Artem Bilan
*/
public class RmiOutboundGateway extends AbstractReplyProducingMessageHandler {
@@ -72,29 +73,20 @@ public class RmiOutboundGateway extends AbstractReplyProducingMessageHandler {
}
@Override
public final Object handleRequestMessage(Message<?> message) {
if (!(message.getPayload() instanceof Serializable)) {
throw new MessageHandlingException(message,
public final Object handleRequestMessage(Message<?> requestMessage) {
if (!(requestMessage.getPayload() instanceof Serializable)) {
throw new MessageHandlingException(requestMessage,
this.getComponentName() + " expects a Serializable payload type " +
"but encountered [" + message.getPayload().getClass().getName() + "]");
"but encountered [" + requestMessage.getPayload().getClass().getName() + "]");
}
Message<?> requestMessage = this.getMessageBuilderFactory().withPayload(message.getPayload())
.copyHeaders(message.getHeaders()).build();
try {
Message<?> reply = this.proxy.exchange(requestMessage);
if (reply != null) {
reply = getMessageBuilderFactory()
.fromMessage(reply)
.copyHeadersIfAbsent(message.getHeaders())
.build();
}
return reply;
return this.proxy.exchange(requestMessage);
}
catch (MessagingException e) {
throw new MessageHandlingException(message, e);
throw new MessageHandlingException(requestMessage, e);
}
catch (RemoteAccessException e) {
throw new MessageHandlingException(message, "Remote failure in RmiOutboundGateway: " +
throw new MessageHandlingException(requestMessage, "Remote failure in RmiOutboundGateway: " +
this.getComponentName(), e);
}
}