From 86c76999f6694d0fe35439d62e5fab6f203f642b Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 17 Jan 2018 18:14:29 -0500 Subject: [PATCH] Optimize AbstractMessageSources To avoid `Message` re-creation during `AbstractMessageSource.receive()` logic, refactor `AbstractMessageSource` implementations to return `AbstractIntegrationMessageBuilder` * Add `AbstractIntegrationMessageBuilder doReceive()` to the `FileReadingMessageSource` to be called from the `AbstractInboundFileSynchronizingMessageSource` to avoid message recreation in its `doReceive()` * Some code style refactoring in the `AbstractMessageSource` --- .../endpoint/AbstractMessageSource.java | 34 ++++----- .../file/FileReadingMessageSource.java | 44 +++++++----- ...tractRemoteFileStreamingMessageSource.java | 3 +- ...InboundFileSynchronizingMessageSource.java | 69 ++++++++++++------- 4 files changed, 93 insertions(+), 57 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractMessageSource.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractMessageSource.java index 62c09b2db6..5571a985b4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractMessageSource.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractMessageSource.java @@ -37,11 +37,13 @@ import org.springframework.util.CollectionUtils; * @author Mark Fisher * @author Oleg Zhurakousky * @author Gary Russell + * @author Artem Bilan + * * @since 2.0 */ @IntegrationManagedResource -public abstract class AbstractMessageSource extends AbstractExpressionEvaluator implements MessageSource, - MessageSourceMetrics, NamedComponent, BeanNameAware { +public abstract class AbstractMessageSource extends AbstractExpressionEvaluator + implements MessageSource, MessageSourceMetrics, NamedComponent, BeanNameAware { private final AtomicLong messageCount = new AtomicLong(); @@ -61,7 +63,7 @@ public abstract class AbstractMessageSource extends AbstractExpressionEvaluat public void setHeaderExpressions(Map headerExpressions) { this.headerExpressions = (headerExpressions != null) - ? headerExpressions : Collections.emptyMap(); + ? headerExpressions : Collections.emptyMap(); } @Override @@ -160,24 +162,24 @@ public abstract class AbstractMessageSource extends AbstractExpressionEvaluat } if (!CollectionUtils.isEmpty(headers)) { // create a new Message from this one in order to apply headers - AbstractIntegrationMessageBuilder builder = getMessageBuilderFactory().fromMessage(message); - builder.copyHeaders(headers); - message = builder.build(); + message = getMessageBuilderFactory() + .fromMessage(message) + .copyHeaders(headers) + .build(); } } else if (result != null) { - T payload = null; + T payload; try { payload = (T) result; } catch (Exception e) { throw new MessagingException("MessageSource returned unexpected type.", e); } - AbstractIntegrationMessageBuilder builder = getMessageBuilderFactory().withPayload(payload); - if (!CollectionUtils.isEmpty(headers)) { - builder.copyHeaders(headers); - } - message = builder.build(); + message = getMessageBuilderFactory() + .withPayload(payload) + .copyHeaders(headers) + .build(); } if (this.countsEnabled && message != null) { this.messageCount.incrementAndGet(); @@ -186,7 +188,7 @@ public abstract class AbstractMessageSource extends AbstractExpressionEvaluat } private Map evaluateHeaders() { - Map results = new HashMap(); + Map results = new HashMap<>(); for (Map.Entry entry : this.headerExpressions.entrySet()) { Object headerValue = this.evaluateExpression(entry.getValue()); if (headerValue != null) { @@ -197,9 +199,9 @@ public abstract class AbstractMessageSource extends AbstractExpressionEvaluat } /** - * Subclasses must implement this method. Typically the returned value will be the payload of - * type T, but the returned value may also be a Message instance whose payload is of type T. - * + * Subclasses must implement this method. Typically the returned value will be the {@code payload} of + * type T, but the returned value may also be a {@link Message} instance whose payload is of type T; + * also can be {@link AbstractIntegrationMessageBuilder} which is used for additional headers population. * @return The value returned. */ protected abstract Object doReceive(); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java index 0c15fd8a1a..a4f8f66c65 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -51,6 +51,7 @@ import org.springframework.integration.core.MessageSource; import org.springframework.integration.file.filters.AcceptOnceFileListFilter; import org.springframework.integration.file.filters.FileListFilter; import org.springframework.integration.file.filters.ResettableFileListFilter; +import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessagingException; import org.springframework.util.Assert; @@ -159,8 +160,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement * queue */ public FileReadingMessageSource(Comparator receptionOrderComparator) { - this.toBeReceived = new PriorityBlockingQueue( - DEFAULT_INTERNAL_QUEUE_CAPACITY, receptionOrderComparator); + this.toBeReceived = new PriorityBlockingQueue<>(DEFAULT_INTERNAL_QUEUE_CAPACITY, receptionOrderComparator); } @@ -350,8 +350,21 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement @Override public Message receive() throws MessagingException { + AbstractIntegrationMessageBuilder messageBuilder = doReceive(); + Message message = null; + if (messageBuilder != null) { + message = messageBuilder.build(); + if (logger.isInfoEnabled()) { + logger.info("Created message: [" + message + "]"); + } + } + + return message; + } + + protected AbstractIntegrationMessageBuilder doReceive() { // rescan only if needed or explicitly configured if (this.scanEachPoll || this.toBeReceived.isEmpty()) { scanInputDirectory(); @@ -366,23 +379,22 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement } if (file != null) { - message = getMessageBuilderFactory().withPayload(file) - .setHeader(FileHeaders.RELATIVE_PATH, file.getAbsolutePath() - .replaceFirst(Matcher.quoteReplacement(this.directory.getAbsolutePath() + File.separator), - "")) + return getMessageBuilderFactory() + .withPayload(file) + .setHeader(FileHeaders.RELATIVE_PATH, + file.getAbsolutePath() + .replaceFirst(Matcher.quoteReplacement( + this.directory.getAbsolutePath() + File.separator), "")) .setHeader(FileHeaders.FILENAME, file.getName()) - .setHeader(FileHeaders.ORIGINAL_FILE, file) - .build(); - if (logger.isInfoEnabled()) { - logger.info("Created message: [" + message + "]"); - } + .setHeader(FileHeaders.ORIGINAL_FILE, file); } - return message; + + return null; } private void scanInputDirectory() { List filteredFiles = this.scanner.listFiles(this.directory); - Set freshFiles = new LinkedHashSet(filteredFiles); + Set freshFiles = new LinkedHashSet<>(filteredFiles); if (!freshFiles.isEmpty()) { this.toBeReceived.addAll(freshFiles); if (logger.isDebugEnabled()) { @@ -421,7 +433,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement private class WatchServiceDirectoryScanner extends DefaultDirectoryScanner implements Lifecycle { - private final ConcurrentMap pathKeys = new ConcurrentHashMap(); + private final ConcurrentMap pathKeys = new ConcurrentHashMap<>(); private WatchService watcher; @@ -547,7 +559,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement } private Set walkDirectory(Path directory, final WatchEvent.Kind kind) { - final Set walkedFiles = new LinkedHashSet(); + final Set walkedFiles = new LinkedHashSet<>(); try { registerWatch(directory); Files.walkFileTree(directory, new SimpleFileVisitor() { diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java index 4214b518c6..196ebf16be 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java @@ -161,8 +161,7 @@ public abstract class AbstractRemoteFileStreamingMessageSource .setHeader(FileHeaders.REMOTE_DIRECTORY, file.getRemoteDirectory()) .setHeader(FileHeaders.REMOTE_FILE, file.getFilename()) .setHeader(FileHeaders.REMOTE_FILE_INFO, - this.fileInfoJson ? file.toJson() : file) - .build(); + this.fileInfoJson ? file.toJson() : file); } catch (IOException e) { throw new MessagingException("IOException when retrieving " + remotePath, e); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java index 4ff5069ae4..ed9121fb94 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java @@ -35,7 +35,7 @@ import org.springframework.integration.file.filters.FileListFilter; import org.springframework.integration.file.filters.FileSystemPersistentAcceptOnceFileListFilter; import org.springframework.integration.file.filters.RegexPatternFileListFilter; import org.springframework.integration.metadata.SimpleMetadataStore; -import org.springframework.messaging.Message; +import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.util.Assert; /** @@ -63,7 +63,19 @@ import org.springframework.util.Assert; * @author Venil Noronha */ public abstract class AbstractInboundFileSynchronizingMessageSource - extends AbstractFetchLimitingMessageSource implements Lifecycle { + extends AbstractFetchLimitingMessageSource + implements Lifecycle { + + /** + * An implementation that will handle the chores of actually connecting to and synchronizing + * the remote file system with the local one, in an inbound direction. + */ + private final AbstractInboundFileSynchronizer synchronizer; + + /** + * The actual {@link LocalFileReadingMessageSource} that monitors the local file system once files are synchronized. + */ + private final LocalFileReadingMessageSource fileSource; private volatile boolean running; @@ -72,22 +84,11 @@ public abstract class AbstractInboundFileSynchronizingMessageSource */ private volatile boolean autoCreateLocalDirectory = true; - /** - * An implementation that will handle the chores of actually connecting to and synchronizing - * the remote file system with the local one, in an inbound direction. - */ - private final AbstractInboundFileSynchronizer synchronizer; - /** * Directory to which things should be synchronized locally. */ private volatile File localDirectory; - /** - * The actual {@link FileReadingMessageSource} that monitors the local file system once files are synchronized. - */ - private final FileReadingMessageSource fileSource; - private volatile FileListFilter localFileListFilter; /** @@ -101,13 +102,14 @@ public abstract class AbstractInboundFileSynchronizingMessageSource public AbstractInboundFileSynchronizingMessageSource(AbstractInboundFileSynchronizer synchronizer, Comparator comparator) { + Assert.notNull(synchronizer, "synchronizer must not be null"); this.synchronizer = synchronizer; if (comparator == null) { - this.fileSource = new FileReadingMessageSource(); + this.fileSource = new LocalFileReadingMessageSource(); } else { - this.fileSource = new FileReadingMessageSource(comparator); + this.fileSource = new LocalFileReadingMessageSource(comparator); } } @@ -243,20 +245,41 @@ public abstract class AbstractInboundFileSynchronizingMessageSource * @param maxFetchSize the maximum files to fetch. */ @Override - public final Message doReceive(int maxFetchSize) { - Message message = this.fileSource.receive(); - if (message == null) { + public final AbstractIntegrationMessageBuilder doReceive(int maxFetchSize) { + AbstractIntegrationMessageBuilder messageBuilder = this.fileSource.doReceive(); + if (messageBuilder == null) { this.synchronizer.synchronizeToLocalDirectory(this.localDirectory, maxFetchSize); - message = this.fileSource.receive(); + messageBuilder = this.fileSource.doReceive(); } - return message; + + return messageBuilder; } private FileListFilter buildFilter() { Pattern completePattern = Pattern.compile("^.*(?(Arrays.asList( - this.localFileListFilter, - new RegexPatternFileListFilter(completePattern))); + return new CompositeFileListFilter<>( + Arrays.asList(this.localFileListFilter, new RegexPatternFileListFilter(completePattern))); + } + + + /** + * The {@link FileReadingMessageSource} extension to increase visibility + * for the {@link FileReadingMessageSource#doReceive()} + */ + private static final class LocalFileReadingMessageSource extends FileReadingMessageSource { + + LocalFileReadingMessageSource() { + } + + LocalFileReadingMessageSource(Comparator receptionOrderComparator) { + super(receptionOrderComparator); + } + + @Override + protected AbstractIntegrationMessageBuilder doReceive() { + return super.doReceive(); + } + } }