diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java index d516d8c8f0..635df75d2b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java @@ -24,7 +24,6 @@ import java.util.List; import java.util.concurrent.Callable; import java.util.concurrent.Executor; import java.util.concurrent.ScheduledFuture; -import java.util.function.Consumer; import java.util.stream.Collectors; import org.aopalliance.aop.Advice; @@ -71,7 +70,7 @@ import reactor.core.scheduler.Schedulers; * The standard polling logic is based on a periodic task scheduling according the provided * {@link Trigger}. * When this endpoint is treated as {@link #isReactive()}, a polling logic is turned into a - * {@link Flux#generate(Consumer)} and {@link Mono#delay(Duration)} combination based on the + * {@link Flux#generate(java.util.function.Consumer)} and {@link Mono#delay(Duration)} combination based on the * {@link SimpleTriggerContext} state. * * @author Mark Fisher diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java index bc6d774ad5..e39a2b0fb7 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java @@ -975,22 +975,25 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply boolean recursion, F file) throws IOException { String fileName = getFilename(file); - String fileSep = this.remoteFileTemplate.getRemoteFileSeparator(); - boolean isDots = ".".equals(fileName) + String fileSeparator = this.remoteFileTemplate.getRemoteFileSeparator(); + final boolean isDirectory = isDirectory(file); + boolean isDots = + ".".equals(fileName) || "..".equals(fileName) - || fileName.endsWith(fileSep + ".") - || fileName.endsWith(fileSep + ".."); - if (this.options.contains(Option.SUBDIRS) || !isDirectory(file)) { - if (recursion && StringUtils.hasText(subDirectory) && (!isDots || this.options.contains(Option.ALL))) { - lsFiles.add(enhanceNameWithSubDirectory(file, subDirectory)); - } - else if (this.options.contains(Option.ALL) || !isDots) { - lsFiles.add(file); + || fileName.endsWith(fileSeparator + ".") + || fileName.endsWith(fileSeparator + ".."); + if ((this.options.contains(Option.SUBDIRS) || !isDirectory) + && (!isDots || this.options.contains(Option.ALL))) { + + F fileToAdd = file; + if (recursion && StringUtils.hasText(subDirectory)) { + fileToAdd = enhanceNameWithSubDirectory(file, subDirectory); } + lsFiles.add(fileToAdd); } - if (recursion && isDirectory(file) && !isDots) { - lsFiles.addAll(listFilesInRemoteDir(session, directory, - subDirectory + fileName + fileSep)); + + if (recursion && isDirectory && !isDots) { + lsFiles.addAll(listFilesInRemoteDir(session, directory, subDirectory + fileName + fileSeparator)); } } @@ -1085,6 +1088,7 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply if ((this.options.contains(Option.PRESERVE_TIMESTAMP) || FileExistsMode.REPLACE_IF_MODIFIED.equals(existsMode)) && (!localFile.setLastModified(getModified(fileInfo)))) { + logger.warn(() -> "Failed to set lastModified on " + localFile); } if (this.options.contains(Option.DELETE)) { @@ -1092,8 +1096,8 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply if (!result) { logger.error("Failed to delete: " + remoteFilePath); } - else if (logger.isDebugEnabled()) { - logger.debug(remoteFilePath + " deleted"); + else { + logger.debug(() -> remoteFilePath + " deleted"); } } } diff --git a/spring-integration-jpa/src/main/java/org/springframework/integration/jpa/config/xml/RetrievingJpaOutboundGatewayParser.java b/spring-integration-jpa/src/main/java/org/springframework/integration/jpa/config/xml/RetrievingJpaOutboundGatewayParser.java index be5f61f168..484975db98 100644 --- a/spring-integration-jpa/src/main/java/org/springframework/integration/jpa/config/xml/RetrievingJpaOutboundGatewayParser.java +++ b/spring-integration-jpa/src/main/java/org/springframework/integration/jpa/config/xml/RetrievingJpaOutboundGatewayParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2021 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. @@ -34,6 +34,7 @@ import org.springframework.util.xml.DomUtils; * @author Amol Nayak * @author Gunnar Hillert * @author Artem Bilan + * * @since 2.2 */ public class RetrievingJpaOutboundGatewayParser extends AbstractJpaOutboundGatewayParser { @@ -41,9 +42,9 @@ public class RetrievingJpaOutboundGatewayParser extends AbstractJpaOutboundGatew @Override protected BeanDefinitionBuilder parseHandler(Element gatewayElement, ParserContext parserContext) { - final BeanDefinitionBuilder jpaOutboundGatewayBuilder = super.parseHandler(gatewayElement, parserContext); + BeanDefinitionBuilder jpaOutboundGatewayBuilder = super.parseHandler(gatewayElement, parserContext); - final BeanDefinitionBuilder jpaExecutorBuilder = + BeanDefinitionBuilder jpaExecutorBuilder = JpaParserUtils.getOutboundGatewayJpaExecutorBuilder(gatewayElement, parserContext); BeanDefinition firstResultExpression = IntegrationNamespaceUtils @@ -60,10 +61,32 @@ public class RetrievingJpaOutboundGatewayParser extends AbstractJpaOutboundGatew jpaExecutorBuilder.addPropertyValue("maxResultsExpression", maxResultsExpression); } + parseIdExpression(gatewayElement, parserContext, jpaExecutorBuilder); + + IntegrationNamespaceUtils.setValueIfAttributeDefined(jpaExecutorBuilder, gatewayElement, "delete-after-poll"); + IntegrationNamespaceUtils.setValueIfAttributeDefined( + jpaExecutorBuilder, gatewayElement, "flush-after-delete", "flush"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(jpaExecutorBuilder, gatewayElement, "delete-in-batch"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(jpaExecutorBuilder, gatewayElement, "expect-single-result"); + + BeanDefinition jpaExecutorBuilderBeanDefinition = jpaExecutorBuilder.getBeanDefinition(); + String gatewayId = resolveId(gatewayElement, jpaOutboundGatewayBuilder.getRawBeanDefinition(), parserContext); + String jpaExecutorBeanName = gatewayId + ".jpaExecutor"; + + parserContext.registerBeanComponent( + new BeanComponentDefinition(jpaExecutorBuilderBeanDefinition, jpaExecutorBeanName)); + + return jpaOutboundGatewayBuilder.addPropertyReference("jpaExecutor", jpaExecutorBeanName) + .addPropertyValue("gatewayType", OutboundGatewayType.RETRIEVING); + } + + private static void parseIdExpression(Element gatewayElement, ParserContext parserContext, + BeanDefinitionBuilder jpaExecutorBuilder) { + if (StringUtils.hasText(gatewayElement.getAttribute("id-expression"))) { - String[] otherAttributes = {"jpa-query", "native-query", "named-query", "first-result", + String[] otherAttributes = { "jpa-query", "native-query", "named-query", "first-result", "first-result-expression", "max-results", "max-results-expression", "delete-in-batch", - "expect-single-result", "parameter-source-factory", "use-payload-as-parameter-source"}; + "expect-single-result", "parameter-source-factory", "use-payload-as-parameter-source" }; StringBuilder others = new StringBuilder(); for (String otherAttribute : otherAttributes) { if (gatewayElement.hasAttribute(otherAttribute) && @@ -79,7 +102,7 @@ public class RetrievingJpaOutboundGatewayParser extends AbstractJpaOutboundGatew if (others.length() > 0 || childElementsExist) { parserContext.getReaderContext().error( (others.length() == 0 ? "" : "'" + others.toString() + "' " - + (childElementsExist ? "and " : "")) + + (childElementsExist ? "and " : "")) + (childElementsExist ? "child elements " : "") + "not allowed with an 'id-expression' attribute.", gatewayElement); @@ -88,22 +111,6 @@ public class RetrievingJpaOutboundGatewayParser extends AbstractJpaOutboundGatew IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined("id-expression", gatewayElement); jpaExecutorBuilder.addPropertyValue("idExpression", idExpressionDef); } - - IntegrationNamespaceUtils.setValueIfAttributeDefined(jpaExecutorBuilder, gatewayElement, "delete-after-poll"); - IntegrationNamespaceUtils.setValueIfAttributeDefined( - jpaExecutorBuilder, gatewayElement, "flush-after-delete", "flush"); - IntegrationNamespaceUtils.setValueIfAttributeDefined(jpaExecutorBuilder, gatewayElement, "delete-in-batch"); - IntegrationNamespaceUtils.setValueIfAttributeDefined(jpaExecutorBuilder, gatewayElement, "expect-single-result"); - - final BeanDefinition jpaExecutorBuilderBeanDefinition = jpaExecutorBuilder.getBeanDefinition(); - final String gatewayId = resolveId(gatewayElement, jpaOutboundGatewayBuilder.getRawBeanDefinition(), parserContext); - final String jpaExecutorBeanName = gatewayId + ".jpaExecutor"; - - parserContext.registerBeanComponent( - new BeanComponentDefinition(jpaExecutorBuilderBeanDefinition, jpaExecutorBeanName)); - - return jpaOutboundGatewayBuilder.addPropertyReference("jpaExecutor", jpaExecutorBeanName) - .addPropertyValue("gatewayType", OutboundGatewayType.RETRIEVING); } } diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index adcb56a451..1a7cdd2530 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -487,7 +487,7 @@ public class AmqpJavaApplication { ---- ==== -[[amqp-gatewway-debatching]] +[[amqp-gateway-debatching]] ==== Batched Messages See <>. @@ -531,7 +531,7 @@ public Object handle(@Payload String payload, @Header(AmqpHeaders.CHANNEL) Chann ==== [[amqp-outbound-endpoints]] -=== Outbound Channel Adapter +=== Outbound Endpoints The following outbound endpoints have many similar configuration options. Starting with version 5.2, the `confirm-timeout` has been added. @@ -1183,7 +1183,7 @@ try { Confirm Confirm = corr.getFuture().get(10, TimeUnit.SECONDS); Message returned = corr.getReturnedMessage(); if (returned !- null) { - // meessage could not be routed + // message could not be routed } } catch { ... } @@ -1424,7 +1424,7 @@ public IntegrationFlow messageDrivenInFow(ConnectionFactory connectionFactory) { public IntegrationFlow pubSubInFlow(ConnectionFactory connectionFactory) { return IntegrationFlows.from(...) ... - .channel(Amqp.publisSubscribeChannel(connectionFactory) + .channel(Amqp.publishSubscribeChannel(connectionFactory) .queueName("baz")) ... .get(); @@ -1516,7 +1516,7 @@ Inbound properties will be mapped to the `amqp_*` headers as before. It is useful to populate the `messageId` property when message consumers are using stateful retry. [[amqp-content-type]] -==== contentType Header +==== The `contentType` Header Unlike other headers, the `AmqpHeaders.CONTENT_TYPE` is not prefixed with `amqp_`; this allows transparent passing of the contentType header across different technologies. For example an inbound HTTP message sent to a RabbitMQ queue.