diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/channel/BeanFactoryChannelResolver.java b/spring-integration-core/src/main/java/org/springframework/integration/support/channel/BeanFactoryChannelResolver.java index cb83ff7836..192900553e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/channel/BeanFactoryChannelResolver.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/channel/BeanFactoryChannelResolver.java @@ -89,7 +89,7 @@ public class BeanFactoryChannelResolver implements DestinationResolver 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; + } + } diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpInboundEndpointParser.java b/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpInboundEndpointParser.java index 4956f8a7d1..f89d6c16ac 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpInboundEndpointParser.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpInboundEndpointParser.java @@ -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 headerElements = DomUtils.getChildElementsByTagName(element, "header"); - - if (!CollectionUtils.isEmpty(headerElements)) { - ManagedMap 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 headerElements = DomUtils.getChildElementsByTagName(element, "header"); + + if (!CollectionUtils.isEmpty(headerElements)) { + ManagedMap 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); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java index 396e57f00e..dffca3229b 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java @@ -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 { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayCrLfSerializer.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayCrLfSerializer.java index fdfbd71425..bcf89c562a 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayCrLfSerializer.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayCrLfSerializer.java @@ -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; } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayRawSerializer.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayRawSerializer.java index 049d1b8c16..a6542cf4ff 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayRawSerializer.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayRawSerializer.java @@ -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; } diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/StoredProcParserUtils.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/StoredProcParserUtils.java index 85a8079896..d7be7d44b4 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/StoredProcParserUtils.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/StoredProcParserUtils.java @@ -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 getSqlParameterDefinitionBeanDefinitions( Element storedProcComponent, ParserContext parserContext) { + List sqlParameterDefinitionChildElements = DomUtils.getChildElementsByTagName(storedProcComponent, "sql-parameter-definition"); - ManagedList sqlParameterList = new ManagedList(); + ManagedList 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 getProcedureParameterBeanDefinitions( Element storedProcComponent, ParserContext parserContext) { - ManagedList procedureParameterList = new ManagedList(); + ManagedList procedureParameterList = new ManagedList<>(); List 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 getReturningResultsetBeanDefinitions( Element storedProcComponent, ParserContext parserContext) { - List returningResultsetChildElements = DomUtils.getChildElementsByTagName(storedProcComponent, "returning-resultset"); + List returningResultsetChildElements = + DomUtils.getChildElementsByTagName(storedProcComponent, "returning-resultset"); - ManagedMap returningResultsetMap = new ManagedMap(); + ManagedMap 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 procedureParameterList = StoredProcParserUtils.getProcedureParameterBeanDefinitions(element, parserContext); - final ManagedList sqlParameterDefinitionList = StoredProcParserUtils.getSqlParameterDefinitionBeanDefinitions(element, parserContext); + final ManagedList procedureParameterList = + StoredProcParserUtils.getProcedureParameterBeanDefinitions(element, parserContext); + final ManagedList sqlParameterDefinitionList = + StoredProcParserUtils.getSqlParameterDefinitionBeanDefinitions(element, parserContext); if (!procedureParameterList.isEmpty()) { storedProcExecutorBuilder.addPropertyValue("procedureParameters", procedureParameterList); @@ -298,7 +304,6 @@ public final class StoredProcParserUtils { } return storedProcExecutorBuilder; - } } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java index 4c35858a74..49ff1b0988 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java @@ -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 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) { diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java index 809db029d1..b425524213 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java @@ -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; } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java index 450354e3ec..a7c4c6f6d3 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java @@ -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 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 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 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) 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) 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) 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) 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) this.serializer).serialize(replyMessage.getPayload()); diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java index 286b3e3d78..e9ff9c3702 100644 --- a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java @@ -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) { diff --git a/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/RFC5424SyslogParser.java b/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/RFC5424SyslogParser.java index 0874717912..8fb8f9ab7f 100644 --- a/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/RFC5424SyslogParser.java +++ b/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/RFC5424SyslogParser.java @@ -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 parse(String lineArg, int octetCount, boolean shortRead) { // NOSONAR NCSS line count - - Map map = new LinkedHashMap(); - + public Map parse(String lineArg, int octetCount, boolean shortRead) { + Map 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 fragments = new ArrayList(); + List 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; diff --git a/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterFactoryBean.java b/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterFactoryBean.java index fffd9cfa43..503b55cd57 100644 --- a/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterFactoryBean.java +++ b/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterFactoryBean.java @@ -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; diff --git a/spring-integration-ws/src/main/java/org/springframework/integration/ws/config/WebServiceOutboundGatewayParser.java b/spring-integration-ws/src/main/java/org/springframework/integration/ws/config/WebServiceOutboundGatewayParser.java index 2b00c68f30..4ad21afcdf 100644 --- a/spring-integration-ws/src/main/java/org/springframework/integration/ws/config/WebServiceOutboundGatewayParser.java +++ b/spring-integration-ws/src/main/java/org/springframework/integration/ws/config/WebServiceOutboundGatewayParser.java @@ -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"); } - } } diff --git a/spring-integration-xmpp/src/main/java/org/springframework/integration/xmpp/support/DefaultXmppHeaderMapper.java b/spring-integration-xmpp/src/main/java/org/springframework/integration/xmpp/support/DefaultXmppHeaderMapper.java index 098c54329b..94672e77e8 100644 --- a/spring-integration-xmpp/src/main/java/org/springframework/integration/xmpp/support/DefaultXmppHeaderMapper.java +++ b/spring-integration-xmpp/src/main/java/org/springframework/integration/xmpp/support/DefaultXmppHeaderMapper.java @@ -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 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 imple } } + private void populateToHeader(Map 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 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); diff --git a/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStore.java b/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStore.java index 0a24f594fc..31b561558d 100644 --- a/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStore.java +++ b/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStore.java @@ -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