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

Fix some Sonar smells

Browse Source
This commit is contained in:
Artem Bilan
2020-08-20 16:08:18 -04:00
parent 8398d9c7e7
commit 750d721437
16 changed files with 572 additions and 547 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-core/src/main/java/org/springframework/integration/support/channel/BeanFactoryChannelResolver.java
View File

@@ -89,7 +89,7 @@ public class BeanFactoryChannelResolver implements DestinationResolver<MessageCh
return this.beanFactory.getBean(name, MessageChannel.class);
}
catch (BeansException e) {
if (!(e instanceof NoSuchBeanDefinitionException)) {
if (!(e instanceof NoSuchBeanDefinitionException)) { // NOSONAR
throw new DestinationResolutionException("A bean definition with name '"
+ name + "' exists, but failed to be created", e);
}

63
spring-integration-core/src/main/java/org/springframework/integration/util/BeanFactoryTypeConverter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2020 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.
@@ -28,6 +28,7 @@ import org.springframework.core.convert.TypeDescriptor;
import org.springframework.core.convert.support.DefaultConversionService;
import org.springframework.expression.TypeConverter;
import org.springframework.integration.history.MessageHistory;
import org.springframework.lang.Nullable;
import org.springframework.messaging.MessageHeaders;
import org.springframework.util.ClassUtils;
@@ -36,15 +37,16 @@ import org.springframework.util.ClassUtils;
* @author Oleg Zhurakousky
* @author Gary Russell
* @author Soby Chacko
* @author Artem Bilan
*/
public class BeanFactoryTypeConverter implements TypeConverter, BeanFactoryAware {
private volatile SimpleTypeConverter delegate = new SimpleTypeConverter();
private SimpleTypeConverter delegate = new SimpleTypeConverter();
private ConversionService conversionService;
private volatile boolean haveCalledDelegateGetDefaultEditor;
private volatile ConversionService conversionService;
public BeanFactoryTypeConverter() {
this.conversionService = DefaultConversionService.getSharedInstance();
@@ -109,33 +111,24 @@ public class BeanFactoryTypeConverter implements TypeConverter, BeanFactoryAware
Class<?> sourceClass = sourceType.getType();
Class<?> targetClass = targetType.getType();
if ((sourceClass == MessageHeaders.class && targetClass == MessageHeaders.class) || // NOSONAR
(sourceClass == MessageHistory.class && targetClass == MessageHistory.class) ||
(sourceType.isAssignableTo(targetType) && ClassUtils.isPrimitiveArray(sourceClass))) {
(sourceClass == MessageHistory.class && targetClass == MessageHistory.class) ||
(sourceType.isAssignableTo(targetType) && ClassUtils.isPrimitiveArray(sourceClass))) {
return value;
}
}
if (this.conversionService.canConvert(sourceType, targetType)) {
return this.conversionService.convert(value, sourceType, targetType);
}
if (!String.class.isAssignableFrom(sourceType.getType())) {
PropertyEditor editor = this.delegate.findCustomEditor(sourceType.getType(), null);
if (editor == null) {
editor = this.getDefaultEditor(sourceType.getType());
}
if (editor != null) { // INT-1441
String text = null;
synchronized (editor) {
editor.setValue(value);
text = editor.getAsText();
}
if (String.class.isAssignableFrom(targetType.getType())) {
return text;
}
return convertValue(text, TypeDescriptor.valueOf(String.class), targetType);
Object editorResult = valueFromEditorIfAny(value, sourceType.getType(), targetType);
if (editorResult == null) {
synchronized (this.delegate) {
return this.delegate.convertIfNecessary(value, targetType.getType());
}
}
synchronized (this.delegate) {
return this.delegate.convertIfNecessary(value, targetType.getType());
else {
return editorResult;
}
}
@@ -154,4 +147,28 @@ public class BeanFactoryTypeConverter implements TypeConverter, BeanFactoryAware
return defaultEditor;
}
@Nullable
private Object valueFromEditorIfAny(Object value, Class<?> sourceClass, TypeDescriptor targetType) {
if (!String.class.isAssignableFrom(sourceClass)) {
PropertyEditor editor = this.delegate.findCustomEditor(sourceClass, null);
if (editor == null) {
editor = getDefaultEditor(sourceClass);
}
if (editor != null) { // INT-1441
String text;
synchronized (editor) {
editor.setValue(value);
text = editor.getAsText();
}
if (String.class.isAssignableFrom(targetType.getType())) {
return text;
}
return convertValue(text, TypeDescriptor.valueOf(String.class), targetType);
}
}
return null;
}
}

162
spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpInboundEndpointParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2020 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.
@@ -84,18 +84,7 @@ public class HttpInboundEndpointParser extends AbstractSingleBeanDefinitionParse
@Override
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
builder.addConstructorArgValue(this.expectReply);
String inputChannelAttributeName = this.getInputChannelAttributeName();
String inputChannelRef = element.getAttribute(inputChannelAttributeName);
if (!StringUtils.hasText(inputChannelRef)) {
if (this.expectReply) {
parserContext.getReaderContext().error(
"a '" + inputChannelAttributeName + "' reference is required", element);
}
else {
inputChannelRef = IntegrationNamespaceUtils.createDirectChannel(element, parserContext);
}
}
builder.addPropertyReference("requestChannel", inputChannelRef);
parseInputChannel(element, parserContext, builder);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel");
BeanDefinition payloadExpressionDef =
@@ -104,22 +93,7 @@ public class HttpInboundEndpointParser extends AbstractSingleBeanDefinitionParse
builder.addPropertyValue("payloadExpression", payloadExpressionDef);
}
List<Element> headerElements = DomUtils.getChildElementsByTagName(element, "header");
if (!CollectionUtils.isEmpty(headerElements)) {
ManagedMap<String, Object> headerElementsMap = new ManagedMap<>();
for (Element headerElement : headerElements) {
String name = headerElement.getAttribute(NAME_ATTRIBUTE);
BeanDefinition headerExpressionDef =
IntegrationNamespaceUtils
.createExpressionDefIfAttributeDefined(IntegrationNamespaceUtils.EXPRESSION_ATTRIBUTE,
headerElement);
if (headerExpressionDef != null) {
headerElementsMap.put(name, headerExpressionDef);
}
}
builder.addPropertyValue("headerExpressions", headerElementsMap);
}
parseHeaders(element, builder);
if (this.expectReply) {
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel");
@@ -145,55 +119,14 @@ public class HttpInboundEndpointParser extends AbstractSingleBeanDefinitionParse
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converters");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "merge-with-default-converters");
String headerMapper = element.getAttribute("header-mapper");
String mappedRequestHeaders = element.getAttribute("mapped-request-headers");
String mappedResponseHeaders = element.getAttribute("mapped-response-headers");
boolean hasMappedRequestHeaders = StringUtils.hasText(mappedRequestHeaders);
boolean hasMappedResponseHeaders = StringUtils.hasText(mappedResponseHeaders);
if (StringUtils.hasText(headerMapper)) {
if (hasMappedRequestHeaders || hasMappedResponseHeaders) {
parserContext.getReaderContext()
.error("Neither 'mapped-request-headers' or 'mapped-response-headers' " +
"attributes are allowed when a 'header-mapper' has been specified.",
parserContext.extractSource(element));
}
builder.addPropertyReference("headerMapper", headerMapper);
}
else {
BeanDefinitionBuilder headerMapperBuilder =
BeanDefinitionBuilder.genericBeanDefinition(DefaultHttpHeaderMapper.class);
headerMapperBuilder.setFactoryMethod("inboundMapper");
if (hasMappedRequestHeaders) {
headerMapperBuilder.addPropertyValue("inboundHeaderNames", mappedRequestHeaders);
}
if (hasMappedResponseHeaders) {
headerMapperBuilder.addPropertyValue("outboundHeaderNames", mappedResponseHeaders);
}
builder.addPropertyValue("headerMapper", headerMapperBuilder.getBeanDefinition());
}
parseHeaderMapper(element, parserContext, builder);
BeanDefinition requestMappingDef = createRequestMapping(element);
builder.addPropertyValue("requestMapping", requestMappingDef);
Element crossOriginElement = DomUtils.getChildElementByTagName(element, "cross-origin");
if (crossOriginElement != null) {
BeanDefinitionBuilder crossOriginBuilder =
BeanDefinitionBuilder.genericBeanDefinition(CrossOrigin.class);
String[] attributes = { "origin", "allowed-headers", "exposed-headers", "max-age", "method" };
for (String crossOriginAttribute : attributes) {
IntegrationNamespaceUtils.setValueIfAttributeDefined(crossOriginBuilder, crossOriginElement,
crossOriginAttribute);
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(crossOriginBuilder, crossOriginElement,
"allow-credentials", true);
builder.addPropertyValue("crossOrigin", crossOriginBuilder.getBeanDefinition());
}
parseCrossOrigin(element, builder);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
"request-payload-type", "requestPayloadTypeClass");
@@ -213,10 +146,93 @@ public class HttpInboundEndpointParser extends AbstractSingleBeanDefinitionParse
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "validator");
}
private void parseInputChannel(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
String inputChannelAttributeName = this.getInputChannelAttributeName();
String inputChannelRef = element.getAttribute(inputChannelAttributeName);
if (!StringUtils.hasText(inputChannelRef)) {
if (this.expectReply) {
parserContext.getReaderContext().error(
"a '" + inputChannelAttributeName + "' reference is required", element);
}
else {
inputChannelRef = IntegrationNamespaceUtils.createDirectChannel(element, parserContext);
}
}
builder.addPropertyReference("requestChannel", inputChannelRef);
}
private String getInputChannelAttributeName() {
return this.expectReply ? "request-channel" : "channel";
}
private void parseHeaders(Element element, BeanDefinitionBuilder builder) {
List<Element> headerElements = DomUtils.getChildElementsByTagName(element, "header");
if (!CollectionUtils.isEmpty(headerElements)) {
ManagedMap<String, Object> headerElementsMap = new ManagedMap<>();
for (Element headerElement : headerElements) {
String name = headerElement.getAttribute(NAME_ATTRIBUTE);
BeanDefinition headerExpressionDef =
IntegrationNamespaceUtils
.createExpressionDefIfAttributeDefined(IntegrationNamespaceUtils.EXPRESSION_ATTRIBUTE,
headerElement);
if (headerExpressionDef != null) {
headerElementsMap.put(name, headerExpressionDef);
}
}
builder.addPropertyValue("headerExpressions", headerElementsMap);
}
}
private void parseHeaderMapper(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
String mappedRequestHeaders = element.getAttribute("mapped-request-headers");
String mappedResponseHeaders = element.getAttribute("mapped-response-headers");
boolean hasMappedRequestHeaders = StringUtils.hasText(mappedRequestHeaders);
boolean hasMappedResponseHeaders = StringUtils.hasText(mappedResponseHeaders);
String headerMapper = element.getAttribute("header-mapper");
if (StringUtils.hasText(headerMapper)) {
if (hasMappedRequestHeaders || hasMappedResponseHeaders) {
parserContext.getReaderContext()
.error("Neither 'mapped-request-headers' or 'mapped-response-headers' " +
"attributes are allowed when a 'header-mapper' has been specified.",
parserContext.extractSource(element));
}
builder.addPropertyReference("headerMapper", headerMapper);
}
else {
BeanDefinitionBuilder headerMapperBuilder =
BeanDefinitionBuilder.genericBeanDefinition(DefaultHttpHeaderMapper.class);
headerMapperBuilder.setFactoryMethod("inboundMapper");
if (hasMappedRequestHeaders) {
headerMapperBuilder.addPropertyValue("inboundHeaderNames", mappedRequestHeaders);
}
if (hasMappedResponseHeaders) {
headerMapperBuilder.addPropertyValue("outboundHeaderNames", mappedResponseHeaders);
}
builder.addPropertyValue("headerMapper", headerMapperBuilder.getBeanDefinition());
}
}
private void parseCrossOrigin(Element element, BeanDefinitionBuilder builder) {
Element crossOriginElement = DomUtils.getChildElementByTagName(element, "cross-origin");
if (crossOriginElement != null) {
BeanDefinitionBuilder crossOriginBuilder =
BeanDefinitionBuilder.genericBeanDefinition(CrossOrigin.class);
String[] attributes = { "origin", "allowed-headers", "exposed-headers", "max-age", "method" };
for (String crossOriginAttribute : attributes) {
IntegrationNamespaceUtils.setValueIfAttributeDefined(crossOriginBuilder, crossOriginElement,
crossOriginAttribute);
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(crossOriginBuilder, crossOriginElement,
"allow-credentials", true);
builder.addPropertyValue("crossOrigin", crossOriginBuilder.getBeanDefinition());
}
}
private BeanDefinition createRequestMapping(Element element) {
BeanDefinitionBuilder requestMappingDefBuilder =
BeanDefinitionBuilder.genericBeanDefinition(RequestMapping.class);
@@ -231,7 +247,7 @@ public class HttpInboundEndpointParser extends AbstractSingleBeanDefinitionParse
Element requestMappingElement = DomUtils.getChildElementByTagName(element, "request-mapping");
if (requestMappingElement != null) {
for (String requestMappingAttribute : new String[] { "params", "headers", "consumes", "produces" }) {
for (String requestMappingAttribute : new String[]{ "params", "headers", "consumes", "produces" }) {
IntegrationNamespaceUtils.setValueIfAttributeDefined(requestMappingDefBuilder, requestMappingElement,
requestMappingAttribute);
}

81
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java
View File

@@ -42,6 +42,7 @@ import org.springframework.util.Assert;
* A TcpConnection that uses and underlying {@link Socket}.
*
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.0
*
@@ -57,7 +58,7 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
private volatile long lastSend;
/**
* Constructs a TcpNetConnection for the socket.
* Construct a TcpNetConnection for the socket.
* @param socket the socket
* @param server if true this connection was created as
* a result of an incoming request.
@@ -188,48 +189,54 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
*/
@Override
public void run() {
boolean okToRun = true;
if (logger.isDebugEnabled()) {
logger.debug(getConnectionId() + " Reading...");
}
while (okToRun) {
Message<?> message = null;
try {
message = getMapper().toMessage(this);
this.lastRead = System.currentTimeMillis();
}
catch (Exception e) {
publishConnectionExceptionEvent(e);
if (handleReadException(e)) {
okToRun = false;
}
}
if (okToRun && message != null) {
if (logger.isDebugEnabled()) {
logger.debug("Message received " + message);
}
try {
TcpListener listener = getListener();
if (listener == null) {
throw new NoListenerException("No listener");
}
listener.onMessage(message);
}
catch (@SuppressWarnings("unused") NoListenerException nle) { // could also be thrown by an interceptor
if (logger.isWarnEnabled()) {
logger.warn("Unexpected message - no endpoint registered with connection interceptor: "
+ getConnectionId()
+ " - "
+ message);
}
}
catch (Exception e2) {
logger.error("Exception sending message: " + message, e2);
}
while (true) {
if (!receiveAndProcessMessage()) {
break;
}
}
}
private boolean receiveAndProcessMessage() {
Message<?> message = null;
try {
message = getMapper().toMessage(this);
this.lastRead = System.currentTimeMillis();
}
catch (Exception e) {
publishConnectionExceptionEvent(e);
if (handleReadException(e)) {
return false;
}
}
if (message != null) {
if (logger.isDebugEnabled()) {
logger.debug("Message received " + message);
}
try {
TcpListener listener = getListener();
if (listener == null) {
throw new NoListenerException("No listener");
}
listener.onMessage(message);
}
catch (NoListenerException nle) { // could also be thrown by an interceptor
if (logger.isWarnEnabled()) {
logger.warn("Unexpected message - no endpoint registered with connection interceptor: "
+ getConnectionId()
+ " - "
+ message);
}
}
catch (Exception e2) {
logger.error("Exception sending message: " + message, e2);
}
}
return true;
}
protected boolean handleReadException(Exception exception) {
Exception e = exception instanceof UncheckedIOException ? (Exception) exception.getCause() : exception;
if (checkTimeout(e)) {
@@ -238,7 +245,7 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
if (!(e instanceof SoftEndOfStreamException)) {
if (e instanceof SocketTimeoutException) {
if (logger.isDebugEnabled()) {
logger.debug("Closed socket after timeout:" + getConnectionId());
logger.debug("Closed socket after timeout: " + getConnectionId());
}
}
else {

14
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayCrLfSerializer.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2020 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.
@@ -26,6 +26,8 @@ import java.io.OutputStream;
* Writes a byte[] to an OutputStream and adds \r\n.
*
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.0
*/
public class ByteArrayCrLfSerializer extends AbstractPooledBufferByteArraySerializer {
@@ -45,8 +47,8 @@ public class ByteArrayCrLfSerializer extends AbstractPooledBufferByteArraySerial
*/
@Override
public byte[] doDeserialize(InputStream inputStream, byte[] buffer) throws IOException {
int n = this.fillToCrLf(inputStream, buffer);
return this.copyToSizedArray(buffer, n);
int n = fillToCrLf(inputStream, buffer);
return copyToSizedArray(buffer, n);
}
public int fillToCrLf(InputStream inputStream, byte[] buffer) throws IOException {
@@ -75,11 +77,7 @@ public class ByteArrayCrLfSerializer extends AbstractPooledBufferByteArraySerial
catch (SoftEndOfStreamException e) { // NOSONAR catch and throw
throw e; // it's an IO exception and we don't want an event for this
}
catch (IOException e) {
publishEvent(e, buffer, n);
throw e;
}
catch (RuntimeException e) {
catch (IOException | RuntimeException e) {
publishEvent(e, buffer, n);
throw e;
}

22
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayRawSerializer.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2020 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.
@@ -38,6 +38,8 @@ import java.net.SocketTimeoutException;
* behavior, set the {@code treatTimeoutAsEndOfMessage} constructor argument to true.
*
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.0.3
*
*/
@@ -65,20 +67,19 @@ public class ByteArrayRawSerializer extends AbstractPooledBufferByteArraySeriali
}
@Override
public void serialize(byte[] bytes, OutputStream outputStream)
throws IOException {
public void serialize(byte[] bytes, OutputStream outputStream) throws IOException {
outputStream.write(bytes);
}
@Override
protected byte[] doDeserialize(InputStream inputStream, byte[] buffer) throws IOException {
int n = 0;
int bite = 0;
int bite;
if (logger.isDebugEnabled()) {
logger.debug("Available to read:" + inputStream.available());
logger.debug("Available to read: " + inputStream.available());
}
try {
while (bite >= 0) {
while (true) {
try {
bite = inputStream.read();
}
@@ -95,8 +96,7 @@ public class ByteArrayRawSerializer extends AbstractPooledBufferByteArraySeriali
break;
}
if (n >= getMaxMessageSize()) {
throw new IOException("Socket was not closed before max message length: "
+ getMaxMessageSize());
throw new IOException("Socket was not closed before max message length: " + getMaxMessageSize());
}
buffer[n++] = (byte) bite;
}
@@ -105,11 +105,7 @@ public class ByteArrayRawSerializer extends AbstractPooledBufferByteArraySeriali
catch (SoftEndOfStreamException e) { // NOSONAR catch and throw
throw e; // it's an IO exception and we don't want an event for this
}
catch (IOException e) {
publishEvent(e, buffer, n);
throw e;
}
catch (RuntimeException e) {
catch (IOException | RuntimeException e) {
publishEvent(e, buffer, n);
throw e;
}

215
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/StoredProcParserUtils.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2020 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.
@@ -48,13 +48,14 @@ import org.springframework.util.xml.DomUtils;
/**
* @author Gunnar Hillert
* @author Artem Bilan
*
* @since 2.1
*/
public final class StoredProcParserUtils {
private static final Log LOGGER = LogFactory.getLog(StoredProcParserUtils.class);
/** Prevent instantiation. */
private StoredProcParserUtils() {
throw new AssertionError();
}
@@ -66,88 +67,94 @@ public final class StoredProcParserUtils {
*/
public static ManagedList<BeanDefinition> getSqlParameterDefinitionBeanDefinitions(
Element storedProcComponent, ParserContext parserContext) {
List<Element> sqlParameterDefinitionChildElements =
DomUtils.getChildElementsByTagName(storedProcComponent, "sql-parameter-definition");
ManagedList<BeanDefinition> sqlParameterList = new ManagedList<BeanDefinition>();
ManagedList<BeanDefinition> sqlParameterList = new ManagedList<>();
for (Element childElement : sqlParameterDefinitionChildElements) {
String name = childElement.getAttribute(AbstractBeanDefinitionParser.NAME_ATTRIBUTE);
String sqlType = childElement.getAttribute("type");
String direction = childElement.getAttribute("direction");
String scale = childElement.getAttribute("scale");
String typeName = childElement.getAttribute("type-name");
String returnType = childElement.getAttribute("return-type");
if (StringUtils.hasText(typeName) && StringUtils.hasText(scale)) {
parserContext.getReaderContext().error("'type-name' and 'scale' attributes are mutually exclusive " +
"for 'sql-parameter-definition' element.", storedProcComponent);
}
if (StringUtils.hasText(returnType) && StringUtils.hasText(scale)) {
parserContext.getReaderContext().error("'returnType' and 'scale' attributes are mutually exclusive " +
"for 'sql-parameter-definition' element.", storedProcComponent);
}
final BeanDefinitionBuilder parameterBuilder;
if ("OUT".equalsIgnoreCase(direction)) {
parameterBuilder = BeanDefinitionBuilder.genericBeanDefinition(SqlOutParameter.class);
}
else if ("INOUT".equalsIgnoreCase(direction)) {
parameterBuilder = BeanDefinitionBuilder.genericBeanDefinition(SqlInOutParameter.class);
}
else {
parameterBuilder = BeanDefinitionBuilder.genericBeanDefinition(SqlParameter.class);
if (StringUtils.hasText(returnType)) {
parserContext.getReaderContext().error("'return-type' attribute can't be provided " +
"for IN 'sql-parameter-definition' element.", storedProcComponent);
}
}
if (StringUtils.hasText(name)) {
parameterBuilder.addConstructorArgValue(name);
}
else {
parserContext.getReaderContext().error(
"The 'name' attribute must be set for the Sql parameter element.", storedProcComponent);
}
if (StringUtils.hasText(sqlType)) {
JdbcTypesEnum jdbcTypeEnum = JdbcTypesEnum.convertToJdbcTypesEnum(sqlType);
if (jdbcTypeEnum != null) {
parameterBuilder.addConstructorArgValue(jdbcTypeEnum.getCode());
}
else {
parameterBuilder.addConstructorArgValue(sqlType);
}
}
else {
parameterBuilder.addConstructorArgValue(Types.VARCHAR);
}
if (StringUtils.hasText(typeName)) {
parameterBuilder.addConstructorArgValue(typeName);
}
else if (StringUtils.hasText(scale)) {
parameterBuilder.addConstructorArgValue(new TypedStringValue(scale, Integer.class));
}
else {
parameterBuilder.addConstructorArgValue(null);
}
if (StringUtils.hasText(returnType)) {
parameterBuilder.addConstructorArgReference(returnType);
}
final BeanDefinitionBuilder parameterBuilder =
parseSqlParameter(storedProcComponent, parserContext, childElement);
sqlParameterList.add(parameterBuilder.getBeanDefinition());
}
return sqlParameterList;
}
private static BeanDefinitionBuilder parseSqlParameter(Element storedProcComponent, ParserContext parserContext,
Element childElement) {
String name = childElement.getAttribute(AbstractBeanDefinitionParser.NAME_ATTRIBUTE);
String sqlType = childElement.getAttribute("type");
String direction = childElement.getAttribute("direction");
String scale = childElement.getAttribute("scale");
String typeName = childElement.getAttribute("type-name");
String returnType = childElement.getAttribute("return-type");
if (StringUtils.hasText(typeName) && StringUtils.hasText(scale)) {
parserContext.getReaderContext().error("'type-name' and 'scale' attributes are mutually exclusive " +
"for 'sql-parameter-definition' element.", storedProcComponent);
}
if (StringUtils.hasText(returnType) && StringUtils.hasText(scale)) {
parserContext.getReaderContext().error("'returnType' and 'scale' attributes are mutually exclusive " +
"for 'sql-parameter-definition' element.", storedProcComponent);
}
final BeanDefinitionBuilder parameterBuilder;
if ("OUT".equalsIgnoreCase(direction)) {
parameterBuilder = BeanDefinitionBuilder.genericBeanDefinition(SqlOutParameter.class);
}
else if ("INOUT".equalsIgnoreCase(direction)) {
parameterBuilder = BeanDefinitionBuilder.genericBeanDefinition(SqlInOutParameter.class);
}
else {
parameterBuilder = BeanDefinitionBuilder.genericBeanDefinition(SqlParameter.class);
if (StringUtils.hasText(returnType)) {
parserContext.getReaderContext().error("'return-type' attribute can't be provided " +
"for IN 'sql-parameter-definition' element.", storedProcComponent);
}
}
if (StringUtils.hasText(name)) {
parameterBuilder.addConstructorArgValue(name);
}
else {
parserContext.getReaderContext()
.error("The 'name' attribute must be set for the Sql parameter element.", storedProcComponent);
}
if (StringUtils.hasText(sqlType)) {
JdbcTypesEnum jdbcTypeEnum = JdbcTypesEnum.convertToJdbcTypesEnum(sqlType);
if (jdbcTypeEnum != null) {
parameterBuilder.addConstructorArgValue(jdbcTypeEnum.getCode());
}
else {
parameterBuilder.addConstructorArgValue(sqlType);
}
}
else {
parameterBuilder.addConstructorArgValue(Types.VARCHAR);
}
if (StringUtils.hasText(typeName)) {
parameterBuilder.addConstructorArgValue(typeName);
}
else if (StringUtils.hasText(scale)) {
parameterBuilder.addConstructorArgValue(new TypedStringValue(scale, Integer.class));
}
else {
parameterBuilder.addConstructorArgValue(null);
}
if (StringUtils.hasText(returnType)) {
parameterBuilder.addConstructorArgReference(returnType);
}
return parameterBuilder;
}
/**
* @param storedProcComponent The element.
* @param parserContext The parser context.
@@ -156,14 +163,15 @@ public final class StoredProcParserUtils {
public static ManagedList<BeanDefinition> getProcedureParameterBeanDefinitions(
Element storedProcComponent, ParserContext parserContext) {
ManagedList<BeanDefinition> procedureParameterList = new ManagedList<BeanDefinition>();
ManagedList<BeanDefinition> procedureParameterList = new ManagedList<>();
List<Element> parameterChildElements = DomUtils
.getChildElementsByTagName(storedProcComponent, "parameter");
for (Element childElement : parameterChildElements) {
BeanDefinitionBuilder parameterBuilder = BeanDefinitionBuilder.genericBeanDefinition(ProcedureParameter.class);
BeanDefinitionBuilder parameterBuilder =
BeanDefinitionBuilder.genericBeanDefinition(ProcedureParameter.class);
String name = childElement.getAttribute(AbstractBeanDefinitionParser.NAME_ATTRIBUTE);
String expression = childElement.getAttribute("expression");
@@ -179,27 +187,18 @@ public final class StoredProcParserUtils {
}
if (StringUtils.hasText(value)) {
if (!StringUtils.hasText(type)) {
if (LOGGER.isInfoEnabled()) {
LOGGER.info(String
.format("Type attribute not set for Store "
+ "Procedure parameter '%s'. Defaulting to "
+ "'java.lang.String'.", value));
LOGGER.info(String.format("Type attribute not set for Store "
+ "Procedure parameter '%s'. Defaulting to "
+ "'java.lang.String'.", value));
}
parameterBuilder.addPropertyValue("value",
new TypedStringValue(value, String.class));
parameterBuilder.addPropertyValue("value", new TypedStringValue(value, String.class));
}
else {
parameterBuilder.addPropertyValue("value",
new TypedStringValue(value, type));
parameterBuilder.addPropertyValue("value", new TypedStringValue(value, type));
}
}
procedureParameterList.add(parameterBuilder.getBeanDefinition());
}
@@ -215,21 +214,23 @@ public final class StoredProcParserUtils {
public static ManagedMap<String, BeanMetadataElement> getReturningResultsetBeanDefinitions(
Element storedProcComponent, ParserContext parserContext) {
List<Element> returningResultsetChildElements = DomUtils.getChildElementsByTagName(storedProcComponent, "returning-resultset");
List<Element> returningResultsetChildElements =
DomUtils.getChildElementsByTagName(storedProcComponent, "returning-resultset");
ManagedMap<String, BeanMetadataElement> returningResultsetMap = new ManagedMap<String, BeanMetadataElement>();
ManagedMap<String, BeanMetadataElement> returningResultsetMap = new ManagedMap<>();
for (Element childElement : returningResultsetChildElements) {
String name = childElement.getAttribute(AbstractBeanDefinitionParser.NAME_ATTRIBUTE);
String rowMapperAsString = childElement.getAttribute("row-mapper");
BeanMetadataElement rowMapperBeanDefinition = null;
BeanMetadataElement rowMapperBeanDefinition;
try {
// Backward compatibility
ClassUtils.forName(rowMapperAsString, parserContext.getReaderContext().getBeanClassLoader());
rowMapperBeanDefinition = BeanDefinitionBuilder.genericBeanDefinition(rowMapperAsString).getBeanDefinition();
rowMapperBeanDefinition =
BeanDefinitionBuilder.genericBeanDefinition(rowMapperAsString).getBeanDefinition();
}
catch (@SuppressWarnings("unused") ClassNotFoundException e) {
//Ignore it and fallback to bean reference
@@ -246,20 +247,20 @@ public final class StoredProcParserUtils {
/**
* Create a new {@link BeanDefinitionBuilder} for the class {@link StoredProcExecutor}.
* Initialize the wrapped {@link StoredProcExecutor} with common properties.
*
* @param element Must not be Null
* @param parserContext Must not be Null
* @return The {@link BeanDefinitionBuilder} for the {@link StoredProcExecutor}
*/
public static BeanDefinitionBuilder getStoredProcExecutorBuilder(final Element element,
final ParserContext parserContext) {
final ParserContext parserContext) {
Assert.notNull(element, "The provided element must not be Null.");
Assert.notNull(element, "The provided element must not be Null.");
Assert.notNull(parserContext, "The provided parserContext must not be Null.");
final String dataSourceRef = element.getAttribute("data-source");
final BeanDefinitionBuilder storedProcExecutorBuilder = BeanDefinitionBuilder.genericBeanDefinition(StoredProcExecutor.class);
final BeanDefinitionBuilder storedProcExecutorBuilder =
BeanDefinitionBuilder.genericBeanDefinition(StoredProcExecutor.class);
storedProcExecutorBuilder.addConstructorArgReference(dataSourceRef);
final String storedProcedureName = element.getAttribute("stored-procedure-name");
@@ -267,7 +268,7 @@ public final class StoredProcParserUtils {
boolean hasStoredProcedureName = StringUtils.hasText(storedProcedureName);
boolean hasStoredProcedureNameExpression = StringUtils.hasText(storedProcedureNameExpression);
if (!(hasStoredProcedureName ^ hasStoredProcedureNameExpression)) {
if (hasStoredProcedureName == hasStoredProcedureNameExpression) {
parserContext.getReaderContext()
.error("Exactly one of 'stored-procedure-name' or 'stored-procedure-name-expression' is required",
element);
@@ -282,13 +283,18 @@ public final class StoredProcParserUtils {
expressionBuilder = BeanDefinitionBuilder.genericBeanDefinition(LiteralExpression.class);
expressionBuilder.addConstructorArgValue(storedProcedureName);
}
storedProcExecutorBuilder.addPropertyValue("storedProcedureNameExpression", expressionBuilder.getBeanDefinition());
storedProcExecutorBuilder.addPropertyValue("storedProcedureNameExpression",
expressionBuilder.getBeanDefinition());
IntegrationNamespaceUtils.setValueIfAttributeDefined(storedProcExecutorBuilder, element, "ignore-column-meta-data");
IntegrationNamespaceUtils.setValueIfAttributeDefined(storedProcExecutorBuilder, element, "jdbc-call-operations-cache-size");
IntegrationNamespaceUtils.setValueIfAttributeDefined(storedProcExecutorBuilder, element,
"ignore-column-meta-data");
IntegrationNamespaceUtils.setValueIfAttributeDefined(storedProcExecutorBuilder, element,
"jdbc-call-operations-cache-size");
final ManagedList<BeanDefinition> procedureParameterList = StoredProcParserUtils.getProcedureParameterBeanDefinitions(element, parserContext);
final ManagedList<BeanDefinition> sqlParameterDefinitionList = StoredProcParserUtils.getSqlParameterDefinitionBeanDefinitions(element, parserContext);
final ManagedList<BeanDefinition> procedureParameterList =
StoredProcParserUtils.getProcedureParameterBeanDefinitions(element, parserContext);
final ManagedList<BeanDefinition> sqlParameterDefinitionList =
StoredProcParserUtils.getSqlParameterDefinitionBeanDefinitions(element, parserContext);
if (!procedureParameterList.isEmpty()) {
storedProcExecutorBuilder.addPropertyValue("procedureParameters", procedureParameterList);
@@ -298,7 +304,6 @@ public final class StoredProcParserUtils {
}
return storedProcExecutorBuilder;
}
}

136
spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2020 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.
@@ -68,35 +68,35 @@ public class ChannelPublishingJmsMessageListener
protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR final
private volatile boolean expectReply;
private volatile MessageConverter messageConverter = new SimpleMessageConverter();
private volatile boolean extractRequestPayload = true;
private volatile boolean extractReplyPayload = true;
private volatile Object defaultReplyDestination;
private volatile String correlationKey;
private volatile long replyTimeToLive = javax.jms.Message.DEFAULT_TIME_TO_LIVE;
private volatile int replyPriority = javax.jms.Message.DEFAULT_PRIORITY;
private volatile int replyDeliveryMode = javax.jms.Message.DEFAULT_DELIVERY_MODE;
private volatile boolean explicitQosEnabledForReplies;
private volatile DestinationResolver destinationResolver = new DynamicDestinationResolver();
private volatile JmsHeaderMapper headerMapper = new DefaultJmsHeaderMapper();
private final GatewayDelegate gatewayDelegate = new GatewayDelegate();
private volatile BeanFactory beanFactory;
private boolean expectReply;
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private MessageConverter messageConverter = new SimpleMessageConverter();
private boolean extractRequestPayload = true;
private boolean extractReplyPayload = true;
private Object defaultReplyDestination;
private String correlationKey;
private long replyTimeToLive = javax.jms.Message.DEFAULT_TIME_TO_LIVE;
private int replyPriority = javax.jms.Message.DEFAULT_PRIORITY;
private int replyDeliveryMode = javax.jms.Message.DEFAULT_DELIVERY_MODE;
private boolean explicitQosEnabledForReplies;
private DestinationResolver destinationResolver = new DynamicDestinationResolver();
private JmsHeaderMapper headerMapper = new DefaultJmsHeaderMapper();
private BeanFactory beanFactory;
private MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
/**
* Specify whether a JMS reply Message is expected.
@@ -313,10 +313,9 @@ public class ChannelPublishingJmsMessageListener
@Override
public void onMessage(javax.jms.Message jmsMessage, Session session) throws JMSException {
Object result = jmsMessage;
Message<?> requestMessage = null;
boolean errors = false;
Message<?> requestMessage;
try {
Object result = jmsMessage;
if (this.extractRequestPayload) {
result = this.messageConverter.fromMessage(jmsMessage);
if (this.logger.isDebugEnabled()) {
@@ -326,53 +325,52 @@ public class ChannelPublishingJmsMessageListener
}
Map<String, Object> headers = this.headerMapper.toHeaders(jmsMessage);
requestMessage = (result instanceof Message<?>) ?
this.messageBuilderFactory.fromMessage((Message<?>) result).copyHeaders(headers).build() :
this.messageBuilderFactory.withPayload(result).copyHeaders(headers).build();
requestMessage =
(result instanceof Message<?>) ?
this.messageBuilderFactory.fromMessage((Message<?>) result).copyHeaders(headers).build() :
this.messageBuilderFactory.withPayload(result).copyHeaders(headers).build();
}
catch (RuntimeException e) {
MessageChannel errorChannel = this.gatewayDelegate.getErrorChannel();
if (errorChannel == null) {
throw e;
}
this.gatewayDelegate.getMessagingTemplate().send(errorChannel,
this.gatewayDelegate.buildErrorMessage(
new MessagingException("Inbound conversion failed for: " + jmsMessage, e)));
errors = true;
this.gatewayDelegate.getMessagingTemplate()
.send(errorChannel,
this.gatewayDelegate.buildErrorMessage(
new MessagingException("Inbound conversion failed for: " + jmsMessage, e)));
return;
}
if (!errors) {
if (!this.expectReply) {
this.gatewayDelegate.send(requestMessage);
if (!this.expectReply) {
this.gatewayDelegate.send(requestMessage);
}
else {
Message<?> replyMessage = this.gatewayDelegate.sendAndReceiveMessage(requestMessage);
if (replyMessage != null) {
Destination destination = getReplyDestination(jmsMessage, session);
if (this.logger.isDebugEnabled()) {
this.logger.debug("Reply destination: " + destination);
}
// convert SI Message to JMS Message
Object replyResult = replyMessage;
if (this.extractReplyPayload) {
replyResult = replyMessage.getPayload();
}
try {
javax.jms.Message jmsReply = this.messageConverter.toMessage(replyResult, session);
// map SI Message Headers to JMS Message Properties/Headers
this.headerMapper.fromHeaders(replyMessage.getHeaders(), jmsReply);
copyCorrelationIdFromRequestToReply(jmsMessage, jmsReply);
sendReply(jmsReply, destination, session);
}
catch (RuntimeException e) {
this.logger.error("Failed to generate JMS Reply Message from: " + replyResult, e);
throw e;
}
}
else {
Message<?> replyMessage = this.gatewayDelegate.sendAndReceiveMessage(requestMessage);
if (replyMessage != null) {
Destination destination = this.getReplyDestination(jmsMessage, session);
if (this.logger.isDebugEnabled()) {
this.logger.debug("Reply destination: " + destination);
}
if (destination != null) {
// convert SI Message to JMS Message
Object replyResult = replyMessage;
if (this.extractReplyPayload) {
replyResult = replyMessage.getPayload();
}
try {
javax.jms.Message jmsReply = this.messageConverter.toMessage(replyResult, session);
// map SI Message Headers to JMS Message Properties/Headers
this.headerMapper.fromHeaders(replyMessage.getHeaders(), jmsReply);
this.copyCorrelationIdFromRequestToReply(jmsMessage, jmsReply);
this.sendReply(jmsReply, destination, session);
}
catch (RuntimeException e) {
this.logger.error("Failed to generate JMS Reply Message from: " + replyResult, e);
throw e;
}
}
}
else if (this.logger.isDebugEnabled()) {
this.logger.debug("expected a reply but none was received");
}
this.logger.debug("expected a reply but none was received");
}
}
}
@@ -396,6 +394,7 @@ public class ChannelPublishingJmsMessageListener
private void copyCorrelationIdFromRequestToReply(javax.jms.Message requestMessage, javax.jms.Message replyMessage)
throws JMSException {
if (this.correlationKey != null) {
if (this.correlationKey.equals("JMSCorrelationID")) {
replyMessage.setJMSCorrelationID(requestMessage.getJMSCorrelationID());
@@ -468,6 +467,7 @@ public class ChannelPublishingJmsMessageListener
private void sendReply(javax.jms.Message replyMessage, Destination destination, Session session)
throws JMSException {
MessageProducer producer = session.createProducer(destination);
try {
if (this.explicitQosEnabledForReplies) {

71
spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2020 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.
@@ -58,20 +58,21 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition
private static final String EXPLICIT_QOS_ENABLED_FOR_REPLIES = "explicit-qos-enabled-for-replies";
private static String[] containerAttributes = new String[] {
JmsParserUtils.CONNECTION_FACTORY_PROPERTY,
JmsParserUtils.DESTINATION_ATTRIBUTE,
JmsParserUtils.DESTINATION_NAME_ATTRIBUTE,
"destination-resolver", "transaction-manager",
"concurrent-consumers", "max-concurrent-consumers",
"acknowledge",
"max-messages-per-task", "selector",
"receive-timeout", "recovery-interval",
"idle-consumer-limit", "idle-task-execution-limit",
"cache-level", "subscription-durable",
"subscription-shared", "subscription-name",
"client-id", "task-executor"
};
private static final String[] CONTAINER_ATTRIBUTES =
{
JmsParserUtils.CONNECTION_FACTORY_PROPERTY,
JmsParserUtils.DESTINATION_ATTRIBUTE,
JmsParserUtils.DESTINATION_NAME_ATTRIBUTE,
"destination-resolver", "transaction-manager",
"concurrent-consumers", "max-concurrent-consumers",
"acknowledge",
"max-messages-per-task", "selector",
"receive-timeout", "recovery-interval",
"idle-consumer-limit", "idle-task-execution-limit",
"cache-level", "subscription-durable",
"subscription-shared", "subscription-name",
"client-id", "task-executor"
};
private final boolean expectReply;
@@ -90,6 +91,7 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition
@Override
protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext)
throws BeanDefinitionStoreException {
String id = super.resolveId(element, definition, parserContext);
if (!this.expectReply && !element.hasAttribute("channel")) {
@@ -124,18 +126,10 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition
private String parseMessageListenerContainer(Element element, ParserContext parserContext,
BeanDefinition adapterBeanDefinition) {
String containerClass = element.getAttribute("container-class");
if (hasExternalContainer(element)) {
if (StringUtils.hasText(containerClass)) {
parserContext.getReaderContext().error("Cannot have both 'container' and 'container-class'", element);
}
for (String containerAttribute : containerAttributes) {
if (element.hasAttribute(containerAttribute)) {
parserContext.getReaderContext().error("The '" + containerAttribute +
"' attribute should not be provided when specifying a 'container' reference.", element);
}
}
return element.getAttribute("container");
return parseExternalContainer(element, parserContext, containerClass);
}
// otherwise, we build a DefaultMessageListenerContainer instance
BeanDefinitionBuilder builder;
@@ -153,9 +147,9 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition
boolean hasDestination = StringUtils.hasText(destination);
boolean hasDestinationName = StringUtils.hasText(destinationName);
if (hasDestination == hasDestinationName) {
parserContext.getReaderContext().error(
"Exactly one of '" + destinationAttribute +
"' or '" + destinationNameAttribute + "' is required.", element);
parserContext.getReaderContext()
.error("Exactly one of '" + destinationAttribute +
"' or '" + destinationNameAttribute + "' is required.", element);
}
builder.addPropertyReference(JmsParserUtils.CONNECTION_FACTORY_PROPERTY,
JmsParserUtils.determineConnectionFactoryBeanName(element, parserContext));
@@ -164,7 +158,8 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition
}
else {
builder.addPropertyValue("destinationName", destinationName);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, pubSubDomainAttribute, "pubSubDomain");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, pubSubDomainAttribute,
"pubSubDomain");
}
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "destination-resolver");
@@ -189,6 +184,19 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition
return beanName;
}
private String parseExternalContainer(Element element, ParserContext parserContext, String containerClass) {
if (StringUtils.hasText(containerClass)) {
parserContext.getReaderContext().error("Cannot have both 'container' and 'container-class'", element);
}
for (String containerAttribute : CONTAINER_ATTRIBUTES) {
if (element.hasAttribute(containerAttribute)) {
parserContext.getReaderContext().error("The '" + containerAttribute +
"' attribute should not be provided when specifying a 'container' reference.", element);
}
}
return element.getAttribute("container");
}
private boolean hasExternalContainer(Element element) {
return element.hasAttribute("container");
@@ -238,7 +246,8 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition
}
builder.addPropertyReference("requestChannel", channelName);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout", "requestTimeout");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-payload", "extractRequestPayload");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-payload",
"extractRequestPayload");
}
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter");
@@ -247,7 +256,7 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition
+ ".listener";
BeanDefinition beanDefinition = builder.getBeanDefinition();
String beanName = BeanDefinitionReaderUtils.generateBeanName(beanDefinition, parserContext.getRegistry());
BeanComponentDefinition component = new BeanComponentDefinition(beanDefinition, beanName, new String[] { alias });
BeanComponentDefinition component = new BeanComponentDefinition(beanDefinition, beanName, new String[]{ alias });
parserContext.registerBeanComponent(component);
return beanName;
}

100
spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2019 the original author or authors.
* Copyright 2014-2020 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.
@@ -57,8 +57,6 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport
private static final String QUEUE_NAME_SUFFIX = ".reply";
private static final RedisSerializer<String> stringSerializer = new StringRedisSerializer();
public static final long DEFAULT_RECEIVE_TIMEOUT = 1000;
public static final long DEFAULT_RECOVERY_INTERVAL = 5000;
@@ -184,7 +182,6 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport
}
}
@SuppressWarnings("unchecked")
private void receiveAndReply() {
byte[] value;
try {
@@ -194,13 +191,13 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport
handlePopException(e);
return;
}
String uuid = null;
String uuid;
if (value != null) {
if (!this.active) {
this.boundListOperations.rightPush(value);
return;
}
uuid = stringSerializer.deserialize(value);
uuid = StringRedisSerializer.UTF_8.deserialize(value);
if (uuid == null) {
return;
}
@@ -211,53 +208,58 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport
handlePopException(e);
return;
}
Message<Object> requestMessage = null;
if (value != null) {
if (!this.active) {
this.template.boundListOps(uuid).rightPush(value);
byte[] serialized = stringSerializer.serialize(uuid);
if (serialized != null) {
this.boundListOperations.rightPush(serialized);
}
getRequestSendAndProduceReply(value, uuid);
}
}
}
@SuppressWarnings("unchecked")
private void getRequestSendAndProduceReply(byte[] value, String uuid) {
Message<Object> requestMessage;
if (!this.active) {
this.template.boundListOps(uuid).rightPush(value);
byte[] serialized = StringRedisSerializer.UTF_8.serialize(uuid);
if (serialized != null) {
this.boundListOperations.rightPush(serialized);
}
return;
}
if (this.extractPayload) {
Object payload = value;
if (this.serializer != null) {
payload = this.serializer.deserialize(value);
if (payload == null) {
return;
}
if (this.extractPayload) {
Object payload = value;
if (this.serializer != null) {
payload = this.serializer.deserialize(value);
if (payload == null) {
return;
}
}
requestMessage = getMessageBuilderFactory().withPayload(payload).build();
}
else {
try {
requestMessage = (Message<Object>) this.serializer.deserialize(value);
if (requestMessage == null) {
return;
}
}
catch (Exception e) {
throw new MessagingException("Deserialization of Message failed.", e);
}
}
Message<?> replyMessage = sendAndReceiveMessage(requestMessage);
if (replyMessage != null) {
if (this.extractPayload) {
value = extractReplyPayload(replyMessage);
}
else {
if (this.serializer != null) {
value = ((RedisSerializer<Object>) this.serializer).serialize(replyMessage);
if (value == null) {
return;
}
}
}
this.template.boundListOps(uuid + QUEUE_NAME_SUFFIX).leftPush(value);
}
requestMessage = getMessageBuilderFactory().withPayload(payload).build();
}
else {
try {
requestMessage = (Message<Object>) this.serializer.deserialize(value);
if (requestMessage == null) {
return;
}
}
catch (Exception e) {
throw new MessagingException("Deserialization of Message failed.", e);
}
}
Message<?> replyMessage = sendAndReceiveMessage(requestMessage);
if (replyMessage != null) {
if (this.extractPayload) {
value = extractReplyPayload(replyMessage);
}
else {
if (this.serializer != null) {
value = ((RedisSerializer<Object>) this.serializer).serialize(replyMessage);
if (value == null) {
return;
}
}
}
this.template.boundListOps(uuid + QUEUE_NAME_SUFFIX).leftPush(value);
}
}
@@ -266,7 +268,7 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport
byte[] value;
if (!(replyMessage.getPayload() instanceof byte[])) {
if (replyMessage.getPayload() instanceof String && !this.serializerExplicitlySet) {
value = stringSerializer.serialize((String) replyMessage.getPayload());
value = StringRedisSerializer.UTF_8.serialize((String) replyMessage.getPayload());
}
else {
value = ((RedisSerializer<Object>) this.serializer).serialize(replyMessage.getPayload());

35
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java
View File

@@ -27,6 +27,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.DisposableBean;
@@ -198,7 +199,25 @@ public abstract class AbstractStompSessionManager implements StompSessionManager
}
return;
}
final CountDownLatch connectLatch = new CountDownLatch(1);
CountDownLatch connectLatch = addStompSessionCallback(currentEpoch);
try {
if (!connectLatch.await(30, TimeUnit.SECONDS)) {
this.logger.error("No response to connection attempt");
if (currentEpoch == this.epoch.get()) {
scheduleReconnect(null);
}
}
}
catch (InterruptedException e1) {
this.logger.error("Interrupted while waiting for connection attempt");
Thread.currentThread().interrupt();
}
}
@NotNull
private CountDownLatch addStompSessionCallback(int currentEpoch) {
CountDownLatch connectLatch = new CountDownLatch(1);
this.stompSessionListenableFuture.addCallback(
stompSession -> {
AbstractStompSessionManager.this.logger.debug("onSuccess");
@@ -222,19 +241,7 @@ public abstract class AbstractStompSessionManager implements StompSessionManager
scheduleReconnect(e);
}
});
try {
if (!connectLatch.await(30, TimeUnit.SECONDS)) {
this.logger.error("No response to connection attempt");
if (currentEpoch == this.epoch.get()) {
scheduleReconnect(null);
}
}
}
catch (InterruptedException e1) {
this.logger.error("Interrupted while waiting for connection attempt");
Thread.currentThread().interrupt();
}
return connectLatch;
}
private void scheduleReconnect(Throwable e) {

62
spring-integration-syslog/src/main/java/org/springframework/integration/syslog/RFC5424SyslogParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2019 the original author or authors.
* Copyright 2014-2020 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.
@@ -21,6 +21,7 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import org.springframework.integration.util.JavaUtils;
import org.springframework.util.Assert;
/**
@@ -29,7 +30,9 @@ import org.springframework.util.Assert;
*
* @author Duncan McIntyre
* @author Gary Russell
* @since 1.4.1
* @author Artem Bilan
*
* @since 4.1.1
*
*/
public class RFC5424SyslogParser {
@@ -57,10 +60,8 @@ public class RFC5424SyslogParser {
this.retainOriginal = retainOriginal;
}
public Map<String, ?> parse(String lineArg, int octetCount, boolean shortRead) { // NOSONAR NCSS line count
Map<String, Object> map = new LinkedHashMap<String, Object>();
public Map<String, ?> parse(String lineArg, int octetCount, boolean shortRead) {
Map<String, Object> map = new LinkedHashMap<>();
String line = lineArg;
Reader r = new Reader(line);
@@ -101,47 +102,26 @@ public class RFC5424SyslogParser {
int severity = pri & 0x7;
int facility = pri >> 3;
map.put(SyslogHeaders.FACILITY, facility);
map.put(SyslogHeaders.SEVERITY, severity);
map.put(SyslogHeaders.SEVERITY_TEXT, Severity.parseInt(severity).label());
if (timestamp != null) {
map.put(SyslogHeaders.TIMESTAMP, timestamp);
}
if (host != null) {
map.put(SyslogHeaders.HOST, host);
}
if (app != null) {
map.put(SyslogHeaders.APP_NAME, app);
}
if (procId != null) {
map.put(SyslogHeaders.PROCID, procId);
}
if (msgId != null) {
map.put(SyslogHeaders.MSGID, msgId);
}
map.put(SyslogHeaders.VERSION, version);
if (structuredData != null) {
map.put(SyslogHeaders.STRUCTURED_DATA, structuredData);
}
map.put(SyslogHeaders.MESSAGE, message);
map.put(SyslogHeaders.DECODE_ERRORS, "false");
if (this.retainOriginal) {
map.put(SyslogHeaders.UNDECODED, line);
}
JavaUtils.INSTANCE
.acceptIfNotNull(timestamp, (value) -> map.put(SyslogHeaders.TIMESTAMP, value))
.acceptIfNotNull(host, (value) -> map.put(SyslogHeaders.HOST, value))
.acceptIfNotNull(app, (value) -> map.put(SyslogHeaders.APP_NAME, value))
.acceptIfNotNull(procId, (value) -> map.put(SyslogHeaders.PROCID, value))
.acceptIfNotNull(msgId, (value) -> map.put(SyslogHeaders.MSGID, value))
.acceptIfNotNull(structuredData, (value) -> map.put(SyslogHeaders.STRUCTURED_DATA, value))
.acceptIfCondition(this.retainOriginal, line, (value) -> map.put(SyslogHeaders.UNDECODED, value));
}
catch (IllegalStateException e) {
catch (IllegalStateException | StringIndexOutOfBoundsException ex) {
map.put(SyslogHeaders.DECODE_ERRORS, "true");
map.put(SyslogHeaders.ERRORS, e.getMessage());
map.put(SyslogHeaders.UNDECODED, line);
}
catch (StringIndexOutOfBoundsException sob) {
map.put(SyslogHeaders.DECODE_ERRORS, "true");
map.put(SyslogHeaders.ERRORS, "Unexpected end of message: " + sob.getMessage());
map.put(SyslogHeaders.ERRORS,
(ex instanceof StringIndexOutOfBoundsException ? "Unexpected end of message: " : "")
+ ex.getMessage());
map.put(SyslogHeaders.UNDECODED, line);
}
return map;
@@ -188,7 +168,7 @@ public class RFC5424SyslogParser {
* @return the structured data.
*/
protected Object parseStructuredDataElements(Reader r) {
List<String> fragments = new ArrayList<String>();
List<String> fragments = new ArrayList<>();
while (r.is('[')) {
r.mark();
r.skipTo(']');
@@ -201,7 +181,7 @@ public class RFC5424SyslogParser {
private final String line;
public int idx;
private int idx;
private int mark;

37
spring-integration-syslog/src/main/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterFactoryBean.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2020 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.
@@ -16,7 +16,6 @@
package org.springframework.integration.syslog.config;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.config.AbstractFactoryBean;
import org.springframework.context.ApplicationEventPublisher;
@@ -28,6 +27,7 @@ import org.springframework.integration.syslog.MessageConverter;
import org.springframework.integration.syslog.inbound.SyslogReceivingChannelAdapterSupport;
import org.springframework.integration.syslog.inbound.TcpSyslogReceivingChannelAdapter;
import org.springframework.integration.syslog.inbound.UdpSyslogReceivingChannelAdapter;
import org.springframework.integration.util.JavaUtils;
import org.springframework.messaging.MessageChannel;
import org.springframework.util.Assert;
@@ -205,30 +205,19 @@ public class SyslogReceivingChannelAdapterFactoryBean extends AbstractFactoryBea
else {
throw new IllegalStateException("Unsupported protocol: " + this.protocol.toString());
}
if (this.port != null) {
adapter.setPort(this.port);
}
if (this.outputChannel != null) {
adapter.setOutputChannel(this.outputChannel);
}
adapter.setAutoStartup(this.autoStartup);
adapter.setPhase(this.phase);
if (this.errorChannel != null) {
adapter.setErrorChannel(this.errorChannel);
}
if (this.sendTimeout != null) {
adapter.setSendTimeout(this.sendTimeout);
}
if (this.converter != null) {
adapter.setConverter(this.converter);
}
if (this.beanName != null) {
adapter.setBeanName(this.beanName);
}
BeanFactory beanFactory = getBeanFactory();
if (beanFactory != null) {
adapter.setBeanFactory(beanFactory);
}
JavaUtils.INSTANCE
.acceptIfNotNull(this.port, adapter::setPort)
.acceptIfNotNull(this.outputChannel, adapter::setOutputChannel)
.acceptIfNotNull(this.errorChannel, adapter::setErrorChannel)
.acceptIfNotNull(this.sendTimeout, adapter::setSendTimeout)
.acceptIfNotNull(this.converter, adapter::setConverter)
.acceptIfNotNull(this.beanName, adapter::setBeanName)
.acceptIfNotNull(getBeanFactory(), adapter::setBeanFactory);
adapter.afterPropertiesSet();
this.syslogAdapter = adapter;
return this.syslogAdapter;

39
spring-integration-ws/src/main/java/org/springframework/integration/ws/config/WebServiceOutboundGatewayParser.java
View File

@@ -98,23 +98,7 @@ public class WebServiceOutboundGatewayParser extends AbstractOutboundGatewayPars
@Override
protected void postProcessGateway(BeanDefinitionBuilder builder, Element element, ParserContext parserContext) {
String marshallerRef = element.getAttribute("marshaller");
String unmarshallerRef = element.getAttribute("unmarshaller");
if (StringUtils.hasText(marshallerRef)) {
builder.addConstructorArgReference(marshallerRef);
if (StringUtils.hasText(unmarshallerRef)) {
builder.addConstructorArgReference(unmarshallerRef);
}
}
else {
String sourceExtractorRef = element.getAttribute("source-extractor");
if (StringUtils.hasText(sourceExtractorRef)) {
builder.addConstructorArgReference(sourceExtractorRef);
}
else {
builder.addConstructorArgValue(null);
}
}
parseMarshallerAttribute(builder, element, parserContext);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "request-callback");
@@ -156,6 +140,26 @@ public class WebServiceOutboundGatewayParser extends AbstractOutboundGatewayPars
if (StringUtils.hasText(interceptorListRef)) {
builder.addPropertyReference("interceptors", interceptorListRef);
}
}
private void parseMarshallerAttribute(BeanDefinitionBuilder builder, Element element, ParserContext parserContext) {
String marshallerRef = element.getAttribute("marshaller");
String unmarshallerRef = element.getAttribute("unmarshaller");
if (StringUtils.hasText(marshallerRef)) {
builder.addConstructorArgReference(marshallerRef);
if (StringUtils.hasText(unmarshallerRef)) {
builder.addConstructorArgReference(unmarshallerRef);
}
}
else {
String sourceExtractorRef = element.getAttribute("source-extractor");
if (StringUtils.hasText(sourceExtractorRef)) {
builder.addConstructorArgReference(sourceExtractorRef);
}
else {
builder.addConstructorArgValue(null);
}
}
if (StringUtils.hasText(marshallerRef) || StringUtils.hasText(unmarshallerRef)) {
String extractPayload = element.getAttribute("extract-payload");
@@ -168,7 +172,6 @@ public class WebServiceOutboundGatewayParser extends AbstractOutboundGatewayPars
else {
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-payload");
}
}
}

47
spring-integration-xmpp/src/main/java/org/springframework/integration/xmpp/support/DefaultXmppHeaderMapper.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2020 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.
@@ -104,30 +104,15 @@ public class DefaultXmppHeaderMapper extends AbstractHeaderMapper<Message> imple
if (StringUtils.hasText(threadId)) {
target.setThread(threadId);
}
String to = getHeaderIfAvailable(headers, XmppHeaders.TO, String.class);
if (StringUtils.hasText(to)) {
try {
target.setTo(JidCreate.from(to));
}
catch (XmppStringprepException e) {
throw new IllegalStateException("Cannot parse 'xmpp_to' header value", e);
}
}
populateToHeader(headers, target);
String from = getHeaderIfAvailable(headers, XmppHeaders.FROM, String.class);
if (StringUtils.hasText(from)) {
try {
target.setFrom(JidCreate.from(from));
}
catch (XmppStringprepException e) {
throw new IllegalStateException("Cannot parse 'xmpp_from' header value", e);
}
}
populateFromHeader(headers, target);
String subject = getHeaderIfAvailable(headers, XmppHeaders.SUBJECT, String.class);
if (StringUtils.hasText(subject)) {
target.setSubject(subject);
}
Object typeHeader = getHeaderIfAvailable(headers, XmppHeaders.TYPE, Object.class);
if (typeHeader instanceof String) {
try {
@@ -145,6 +130,30 @@ public class DefaultXmppHeaderMapper extends AbstractHeaderMapper<Message> imple
}
}
private void populateToHeader(Map<String, Object> headers, Message target) {
String to = getHeaderIfAvailable(headers, XmppHeaders.TO, String.class);
if (StringUtils.hasText(to)) {
try {
target.setTo(JidCreate.from(to));
}
catch (XmppStringprepException e) {
throw new IllegalStateException("Cannot parse 'xmpp_to' header value", e);
}
}
}
private void populateFromHeader(Map<String, Object> headers, Message target) {
String from = getHeaderIfAvailable(headers, XmppHeaders.FROM, String.class);
if (StringUtils.hasText(from)) {
try {
target.setFrom(JidCreate.from(from));
}
catch (XmppStringprepException e) {
throw new IllegalStateException("Cannot parse 'xmpp_from' header value", e);
}
}
}
@Override
protected void populateUserDefinedHeader(String headerName, Object headerValue, Message target) {
JivePropertiesManager.addProperty(target, headerName, headerValue);

33
spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStore.java
View File

@@ -51,8 +51,6 @@ public class ZookeeperMetadataStore implements ListenableMetadataStore, SmartLif
private static final String KEY_MUST_NOT_BE_NULL = "'key' must not be null.";
private static final String UNUSED = "unused";
private final Object lifecycleMonitor = new Object();
private final CuratorFramework client;
@@ -123,7 +121,7 @@ public class ZookeeperMetadataStore implements ListenableMetadataStore, SmartLif
createNode(key, value);
return null;
}
catch (@SuppressWarnings(UNUSED) KeeperException.NodeExistsException e) {
catch (KeeperException.NodeExistsException e) {
// so the data actually exists, we can read it
return get(key);
}
@@ -147,14 +145,11 @@ public class ZookeeperMetadataStore implements ListenableMetadataStore, SmartLif
}
return true;
}
catch (@SuppressWarnings(UNUSED) KeeperException.NoNodeException e) {
catch (KeeperException.NoNodeException | KeeperException.BadVersionException e) {
// ignore, the node doesn't exist there's nothing to replace
return false;
}
catch (@SuppressWarnings(UNUSED) KeeperException.BadVersionException e) {
// ignore
return false;
}
// ignore
catch (Exception e) {
throw new ZookeeperMetadataStoreException("Cannot replace value", e);
}
@@ -183,7 +178,7 @@ public class ZookeeperMetadataStore implements ListenableMetadataStore, SmartLif
try {
createNode(key, value);
}
catch (@SuppressWarnings(UNUSED) KeeperException.NodeExistsException e) {
catch (KeeperException.NodeExistsException e) {
updateNode(key, value, -1);
}
}
@@ -237,7 +232,7 @@ public class ZookeeperMetadataStore implements ListenableMetadataStore, SmartLif
this.updateMap.put(key, new LocalChildData(null, Integer.MAX_VALUE));
return IntegrationUtils.bytesToString(bytes, this.encoding);
}
catch (@SuppressWarnings(UNUSED) KeeperException.NoNodeException e) {
catch (KeeperException.NoNodeException e) {
// ignore - the node doesn't exist
return null;
}
@@ -353,7 +348,8 @@ public class ZookeeperMetadataStore implements ListenableMetadataStore, SmartLif
synchronized (ZookeeperMetadataStore.this.updateMap) {
String eventPath = event.getData().getPath();
String eventKey = getKey(eventPath);
byte[] eventData = event.getData().getData();
String value =
IntegrationUtils.bytesToString(event.getData().getData(), ZookeeperMetadataStore.this.encoding);
switch (event.getType()) {
case CHILD_ADDED:
if (ZookeeperMetadataStore.this.updateMap.containsKey(eventKey)) {
@@ -362,10 +358,7 @@ public class ZookeeperMetadataStore implements ListenableMetadataStore, SmartLif
ZookeeperMetadataStore.this.updateMap.remove(eventPath);
}
}
for (MetadataStoreListener listener : ZookeeperMetadataStore.this.listeners) {
listener.onAdd(eventKey, IntegrationUtils.bytesToString(eventData,
ZookeeperMetadataStore.this.encoding));
}
ZookeeperMetadataStore.this.listeners.forEach((listener) -> listener.onAdd(eventKey, value));
break;
case CHILD_UPDATED:
if (ZookeeperMetadataStore.this.updateMap.containsKey(eventKey)) {
@@ -374,17 +367,11 @@ public class ZookeeperMetadataStore implements ListenableMetadataStore, SmartLif
ZookeeperMetadataStore.this.updateMap.remove(eventPath);
}
}
for (MetadataStoreListener listener : ZookeeperMetadataStore.this.listeners) {
listener.onUpdate(eventKey, IntegrationUtils.bytesToString(eventData,
ZookeeperMetadataStore.this.encoding));
}
ZookeeperMetadataStore.this.listeners.forEach((listener) -> listener.onUpdate(eventKey, value));
break;
case CHILD_REMOVED:
ZookeeperMetadataStore.this.updateMap.remove(eventKey);
for (MetadataStoreListener listener : ZookeeperMetadataStore.this.listeners) {
listener.onRemove(eventKey, IntegrationUtils.bytesToString(eventData,
ZookeeperMetadataStore.this.encoding));
}
ZookeeperMetadataStore.this.listeners.forEach((listener) -> listener.onRemove(eventKey, value));
break;
default:
// ignore all other events