diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java index 6df091b0af..dea1b508cb 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java @@ -41,6 +41,7 @@ import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper; import org.springframework.integration.gateway.MessagingGatewaySupport; import org.springframework.integration.support.ErrorMessageStrategy; import org.springframework.integration.support.ErrorMessageUtils; +import org.springframework.messaging.MessageChannel; import org.springframework.retry.RecoveryCallback; import org.springframework.retry.support.RetrySynchronizationManager; import org.springframework.retry.support.RetryTemplate; @@ -292,10 +293,11 @@ public class AmqpInboundGateway extends MessagingGatewaySupport { } } catch (RuntimeException e) { - if (getErrorChannel() != null) { + MessageChannel errorChannel = getErrorChannel(); + if (errorChannel != null) { setAttributesIfNecessary(message, null); - AmqpInboundGateway.this.messagingTemplate.send(getErrorChannel(), buildErrorMessage(null, - new ListenerExecutionFailedException("Message conversion failed", e, message))); + AmqpInboundGateway.this.messagingTemplate.send(errorChannel, buildErrorMessage(null, + new ListenerExecutionFailedException("Message conversion failed", e, message))); } else { throw e; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMethodInboundMessageMapper.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMethodInboundMessageMapper.java index c073eb3be5..dfcc4bcf78 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMethodInboundMessageMapper.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMethodInboundMessageMapper.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 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. @@ -188,6 +188,7 @@ class GatewayMethodInboundMessageMapper implements InboundMessageMapper mapArgumentsToMessage(Object[] arguments, Map headers) { try { return this.argsMapper.toMessage(new MethodArgsHolder(this.method, arguments), headers); @@ -220,6 +221,7 @@ class GatewayMethodInboundMessageMapper implements InboundMessageMapper returnType = invocation.getMethod().getReturnType(); if (this.asyncExecutor != null && !Object.class.equals(returnType)) { @@ -458,16 +459,17 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint if (Mono.class.isAssignableFrom(returnType)) { return Mono.fromSupplier(new Invoker(invocation)); } - return this.doInvoke(invocation, true); + return doInvoke(invocation, true); } + @Nullable protected Object doInvoke(MethodInvocation invocation, boolean runningOnCallerThread) throws Throwable { Method method = invocation.getMethod(); if (AopUtils.isToStringMethod(method)) { return "gateway proxy for service interface [" + this.serviceInterface + "]"; } try { - return this.invokeGatewayMethod(invocation, runningOnCallerThread); + return invokeGatewayMethod(invocation, runningOnCallerThread); } catch (Throwable e) { //NOSONAR - ok to catch, rethrown below this.rethrowExceptionCauseIfPossible(e, invocation.getMethod()); @@ -709,7 +711,10 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint gateway.setRequestTimeout(-1); } else if (requestTimeout instanceof ValueExpression) { - gateway.setRequestTimeout(requestTimeout.getValue(Long.class)); + Long timeout = requestTimeout.getValue(Long.class); + if (timeout != null) { + gateway.setRequestTimeout(timeout); + } } else { messageMapper.setSendTimeoutExpression(requestTimeout); @@ -718,7 +723,10 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint gateway.setReplyTimeout(-1); } else if (replyTimeout instanceof ValueExpression) { - gateway.setReplyTimeout(replyTimeout.getValue(Long.class)); + Long timeout = replyTimeout.getValue(Long.class); + if (timeout != null) { + gateway.setReplyTimeout(timeout); + } } else { messageMapper.setReplyTimeoutExpression(replyTimeout); @@ -751,6 +759,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint } @SuppressWarnings("unchecked") + @Nullable private T convert(Object source, Class expectedReturnType) { if (Future.class.isAssignableFrom(expectedReturnType)) { return (T) source; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java index f5c5f28f47..fc84e797aa 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java @@ -396,6 +396,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint * @return the channel or null. * @since 4.3 */ + @Nullable public MessageChannel getErrorChannel() { if (this.errorChannelName != null) { synchronized (this) { @@ -431,6 +432,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint } } + @Nullable protected Object receive() { this.initializeIfNecessary(); MessageChannel replyChannel = getReplyChannel(); @@ -439,6 +441,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint return this.messagingTemplate.receiveAndConvert(replyChannel, Object.class); } + @Nullable protected Message receiveMessage() { initializeIfNecessary(); MessageChannel replyChannel = getReplyChannel(); @@ -447,6 +450,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint return this.messagingTemplate.receive(replyChannel); } + @Nullable protected Object receive(long timeout) { this.initializeIfNecessary(); MessageChannel replyChannel = getReplyChannel(); @@ -455,6 +459,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint return this.messagingTemplate.receiveAndConvert(replyChannel, timeout); } + @Nullable protected Message receiveMessage(long timeout) { initializeIfNecessary(); MessageChannel replyChannel = getReplyChannel(); @@ -463,10 +468,12 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint return this.messagingTemplate.receive(replyChannel, timeout); } + @Nullable protected Object sendAndReceive(Object object) { return this.doSendAndReceive(object, true); } + @Nullable protected Message sendAndReceiveMessage(Object object) { return (Message) this.doSendAndReceive(object, false); } @@ -716,7 +723,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint * @return the attributes. * @since 4.3.10 */ - protected AttributeAccessor getErrorMessageAttributes(Message message) { + protected AttributeAccessor getErrorMessageAttributes(@Nullable Message message) { return ErrorMessageUtils.getAttributeAccessor(message, null); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationGraphServer.java b/spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationGraphServer.java index 41204293d8..1842d36d7e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationGraphServer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationGraphServer.java @@ -351,10 +351,10 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat } private MessageGatewayNode gatewayNode(String name, MessagingGatewaySupport gateway) { - String errorChannel = gateway.getErrorChannel() != null ? gateway.getErrorChannel().toString() : null; - String requestChannel = gateway.getRequestChannel() != null - ? gateway.getRequestChannel().toString() // NOSONAR not null - : null; + MessageChannel gwErrorChannel = gateway.getErrorChannel(); + String errorChannel = gwErrorChannel != null ? gwErrorChannel.toString() : null; + MessageChannel gwRequestChannel = gateway.getRequestChannel(); + String requestChannel = gwRequestChannel != null ? gwRequestChannel.toString() : null; return new MessageGatewayNode(this.nodeId.incrementAndGet(), name, gateway, requestChannel, errorChannel); } diff --git a/spring-integration-groovy/src/main/java/org/springframework/integration/groovy/GroovyCommandMessageProcessor.java b/spring-integration-groovy/src/main/java/org/springframework/integration/groovy/GroovyCommandMessageProcessor.java index 34faed6d5a..4fbf392119 100644 --- a/spring-integration-groovy/src/main/java/org/springframework/integration/groovy/GroovyCommandMessageProcessor.java +++ b/spring-integration-groovy/src/main/java/org/springframework/integration/groovy/GroovyCommandMessageProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 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. @@ -17,6 +17,7 @@ package org.springframework.integration.groovy; import java.util.Map; +import java.util.UUID; import org.springframework.integration.scripting.AbstractScriptExecutingMessageProcessor; import org.springframework.integration.scripting.DefaultScriptVariableGenerator; @@ -28,6 +29,7 @@ import org.springframework.scripting.groovy.GroovyScriptFactory; import org.springframework.scripting.support.StaticScriptSource; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; +import org.springframework.util.ObjectUtils; import groovy.lang.Binding; import groovy.lang.GString; @@ -38,6 +40,7 @@ import groovy.lang.GString; * @author Oleg Zhurakousky * @author Artem Bilan * @author Stefan Reuter + * @author Gary Russell * @since 2.0 */ public class GroovyCommandMessageProcessor extends AbstractScriptExecutingMessageProcessor { @@ -133,7 +136,9 @@ public class GroovyCommandMessageProcessor extends AbstractScriptExecutingMessag protected String generateScriptName(Message message) { // Don't use the same script (class) name for all invocations by default - return getClass().getSimpleName() + message.getHeaders().getId().toString().replaceAll("-", ""); + UUID id = message.getHeaders().getId(); + return getClass().getSimpleName() + + (id != null ? id.toString().replaceAll("-", "") : ObjectUtils.getIdentityHexString(message)); } }