From df9593a528b0317e46c8c9081f4719c38e1437ba Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 16 Mar 2017 10:10:01 -0400 Subject: [PATCH] INT-4244: Optimize JpaOutboundGateway for Message 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 --- .../AbstractAmqpOutboundEndpoint.java | 8 +- .../amqp/outbound/AmqpOutboundEndpoint.java | 10 ++- .../outbound/AsyncAmqpOutboundGateway.java | 14 ++-- .../integration/handler/DelayHandler.java | 11 ++- .../scattergather/ScatterGatherHandler.java | 2 +- .../transformer/ContentEnricher.java | 6 +- .../AbstractRemoteFileOutboundGateway.java | 32 ++++---- .../RemoteFileOutboundGatewayTests.java | 81 ++++++++++--------- .../integration/jdbc/JdbcOutboundGateway.java | 7 +- .../jpa/outbound/JpaOutboundGateway.java | 21 +++-- .../JpaOutboundGatewayIntegrationTests.java | 5 +- .../outbound/RedisQueueOutboundGateway.java | 6 +- .../integration/rmi/RmiOutboundGateway.java | 26 +++--- 13 files changed, 120 insertions(+), 109 deletions(-) diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java index faea969cb7..73e04c6bc5 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java @@ -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 headers = getHeaderMapper().toHeadersFromReply(amqpReplyMessage.getMessageProperties()); builder.copyHeadersIfAbsent(headers); - return builder.build(); + return builder; } protected Message buildReturnedMessage(org.springframework.amqp.core.Message message, diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java index 7567ca8153..696d3997ba 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java @@ -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 diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java index 7b54531a91..b31c2cf305 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java @@ -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); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java index 51c0b0f177..d6622ee493 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java @@ -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) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scattergather/ScatterGatherHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/scattergather/ScatterGatherHandler.java index 1af078eb66..810a3910f1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/scattergather/ScatterGatherHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/scattergather/ScatterGatherHandler.java @@ -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; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/ContentEnricher.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/ContentEnricher.java index b428488726..8c67e23a1a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transformer/ContentEnricher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/ContentEnricher.java @@ -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); } } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java index c69a0b18aa..006d1cff7d 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java @@ -566,9 +566,9 @@ public abstract class AbstractRemoteFileOutboundGateway 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 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 extends AbstractReply final String remoteDir = getRemoteDirectory(remoteFilePath, remoteFilename); List 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 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 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) { diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java index 4dc606d4b3..c02994becc 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java @@ -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> out = (Message>) gw - .handleRequestMessage(new GenericMessage("testremote/x")); + MessageBuilder> out = (MessageBuilder>) 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> out = (Message>) gw - .handleRequestMessage(new GenericMessage("testremote/*")); + MessageBuilder> out = (MessageBuilder>) 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> out = (Message>) gw - .handleRequestMessage(new GenericMessage("testremote/f1")); + MessageBuilder> out = (MessageBuilder>) 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 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 args = new AtomicReference(); 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("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 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> out = (Message>) gw - .handleRequestMessage(new GenericMessage("testremote/x")); + MessageBuilder> out = (MessageBuilder>) 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> out = (Message>) gw - .handleRequestMessage(new GenericMessage("testremote/x")); + MessageBuilder> out = (MessageBuilder>) 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> out = (Message>) gw - .handleRequestMessage(new GenericMessage("testremote/x")); + MessageBuilder> out = (MessageBuilder>) 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> out = (Message>) gw - .handleRequestMessage(new GenericMessage("testremote")); + MessageBuilder> out = (MessageBuilder>) 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> out = (Message>) gw - .handleRequestMessage(new GenericMessage("testremote")); + MessageBuilder> out = (MessageBuilder>) 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> out = (Message>) gw - .handleRequestMessage(new GenericMessage("testremote")); + MessageBuilder> out = (MessageBuilder>) 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> out = (Message>) gw - .handleRequestMessage(new GenericMessage("testremote")); + MessageBuilder> out = (MessageBuilder>) 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> out = (Message>) gw - .handleRequestMessage(new GenericMessage("testremote")); + MessageBuilder> out = (MessageBuilder>) 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> out = (Message>) gw - .handleRequestMessage(new GenericMessage("testremote")); + MessageBuilder> out = (MessageBuilder>) 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> out = (Message>) gw - .handleRequestMessage(new GenericMessage("testremote")); + MessageBuilder> out = (MessageBuilder>) 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 out = (Message) gw.handleRequestMessage(new GenericMessage("f1")); + MessageBuilder out = (MessageBuilder) 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 out; + MessageBuilder out; try { - out = (Message) gw.handleRequestMessage(new GenericMessage("f1")); + out = (MessageBuilder) 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) gw.handleRequestMessage(new GenericMessage("f1")); + out = (MessageBuilder) gw.handleRequestMessage(new GenericMessage<>("f1")); fail("Exception expected"); } catch (MessageHandlingException e) { @@ -642,17 +643,17 @@ public class RemoteFileOutboundGatewayTests { } gw.setFileExistsMode(FileExistsMode.IGNORE); - out = (Message) gw.handleRequestMessage(new GenericMessage("f1")); + out = (MessageBuilder) gw.handleRequestMessage(new GenericMessage<>("f1")); assertEquals(outFile, out.getPayload()); assertContents("foo", outFile); gw.setFileExistsMode(FileExistsMode.APPEND); - out = (Message) gw.handleRequestMessage(new GenericMessage("f1")); + out = (MessageBuilder) gw.handleRequestMessage(new GenericMessage<>("f1")); assertEquals(outFile, out.getPayload()); assertContents("footestfile", outFile); gw.setFileExistsMode(FileExistsMode.REPLACE); - out = (Message) gw.handleRequestMessage(new GenericMessage("f1")); + out = (MessageBuilder) gw.handleRequestMessage(new GenericMessage<>("f1")); assertEquals(outFile, out.getPayload()); assertContents("testfile", outFile); @@ -730,7 +731,7 @@ public class RemoteFileOutboundGatewayTests { }); @SuppressWarnings("unchecked") - Message out = (Message) gw.handleRequestMessage(new GenericMessage("x/f1")); + MessageBuilder out = (MessageBuilder) 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 out = (Message) gw - .handleRequestMessage(new GenericMessage("testremote/x/f1")); + MessageBuilder out = (MessageBuilder) gw + .handleRequestMessage(new GenericMessage<>("testremote/x/f1")); assertEquals(Boolean.TRUE, out.getPayload()); verify(session).remove("testremote/x/f1"); assertEquals("testremote/x/", diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcOutboundGateway.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcOutboundGateway.java index f833f16954..f2b422bd76 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcOutboundGateway.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcOutboundGateway.java @@ -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; } /** diff --git a/spring-integration-jpa/src/main/java/org/springframework/integration/jpa/outbound/JpaOutboundGateway.java b/spring-integration-jpa/src/main/java/org/springframework/integration/jpa/outbound/JpaOutboundGateway.java index a004282b9a..6de879fdf1 100644 --- a/spring-integration-jpa/src/main/java/org/springframework/integration/jpa/outbound/JpaOutboundGateway.java +++ b/spring-integration-jpa/src/main/java/org/springframework/integration/jpa/outbound/JpaOutboundGateway.java @@ -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; } + } diff --git a/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/outbound/JpaOutboundGatewayIntegrationTests.java b/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/outbound/JpaOutboundGatewayIntegrationTests.java index fb8a8e0bd7..9830de727c 100644 --- a/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/outbound/JpaOutboundGatewayIntegrationTests.java +++ b/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/outbound/JpaOutboundGatewayIntegrationTests.java @@ -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); diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundGateway.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundGateway.java index ff6ada413a..78346991a5 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundGateway.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundGateway.java @@ -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; diff --git a/spring-integration-rmi/src/main/java/org/springframework/integration/rmi/RmiOutboundGateway.java b/spring-integration-rmi/src/main/java/org/springframework/integration/rmi/RmiOutboundGateway.java index 0f53a91058..2fb697621a 100644 --- a/spring-integration-rmi/src/main/java/org/springframework/integration/rmi/RmiOutboundGateway.java +++ b/spring-integration-rmi/src/main/java/org/springframework/integration/rmi/RmiOutboundGateway.java @@ -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); } }