Optimize AbstractMessageSources

To avoid `Message` re-creation during `AbstractMessageSource.receive()`
logic, refactor `AbstractMessageSource` implementations
to return `AbstractIntegrationMessageBuilder`

* Add `AbstractIntegrationMessageBuilder<File> doReceive()` to the
`FileReadingMessageSource` to be called from the `AbstractInboundFileSynchronizingMessageSource`
to avoid message recreation in its `doReceive()`
* Some code style refactoring in the `AbstractMessageSource`

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractMessageSource.java

@@ -37,11 +37,13 @@
  * @author Mark Fisher
  * @author Oleg Zhurakousky
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 2.0
  */
 @IntegrationManagedResource
-public abstract class AbstractMessageSource<T> extends AbstractExpressionEvaluator implements MessageSource<T>,
-		MessageSourceMetrics, NamedComponent, BeanNameAware {
+public abstract class AbstractMessageSource<T> extends AbstractExpressionEvaluator
+		implements MessageSource<T>, MessageSourceMetrics, NamedComponent, BeanNameAware {
 
 	private final AtomicLong messageCount = new AtomicLong();
 
@@ -61,7 +63,7 @@ public abstract class AbstractMessageSource<T> extends AbstractExpressionEvaluat
 
 	public void setHeaderExpressions(Map<String, Expression> headerExpressions) {
 		this.headerExpressions = (headerExpressions != null)
-				? headerExpressions : Collections.<String, Expression>emptyMap();
+				? headerExpressions : Collections.emptyMap();
 	}
 
 	@Override
@@ -160,24 +162,24 @@ else if (result instanceof Message<?>) {
 			}
 			if (!CollectionUtils.isEmpty(headers)) {
 				// create a new Message from this one in order to apply headers
-				AbstractIntegrationMessageBuilder<T> 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<T> 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 @@ else if (result != null) {
 	}
 
 	private Map<String, Object> evaluateHeaders() {
-		Map<String, Object> results = new HashMap<String, Object>();
+		Map<String, Object> results = new HashMap<>();
 		for (Map.Entry<String, Expression> entry : this.headerExpressions.entrySet()) {
 			Object headerValue = this.evaluateExpression(entry.getValue());
 			if (headerValue != null) {
@@ -197,9 +199,9 @@ private Map<String, Object> evaluateHeaders() {
 	}
 
 	/**
-	 * 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();

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.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 FileReadingMessageSource(int internalQueueCapacity) {
 	 *            queue
 	 */
 	public FileReadingMessageSource(Comparator<File> receptionOrderComparator) {
-		this.toBeReceived = new PriorityBlockingQueue<File>(
-				DEFAULT_INTERNAL_QUEUE_CAPACITY, receptionOrderComparator);
+		this.toBeReceived = new PriorityBlockingQueue<>(DEFAULT_INTERNAL_QUEUE_CAPACITY, receptionOrderComparator);
 	}
 
 
@@ -350,8 +350,21 @@ protected void onInit() {
 
 	@Override
 	public Message<File> receive() throws MessagingException {
+		AbstractIntegrationMessageBuilder<File> messageBuilder = doReceive();
+
 		Message<File> message = null;
 
+		if (messageBuilder != null) {
+			message = messageBuilder.build();
+			if (logger.isInfoEnabled()) {
+				logger.info("Created message: [" + message + "]");
+			}
+		}
+
+		return message;
+	}
+
+	protected AbstractIntegrationMessageBuilder<File> doReceive() {
 		// rescan only if needed or explicitly configured
 		if (this.scanEachPoll || this.toBeReceived.isEmpty()) {
 			scanInputDirectory();
@@ -366,23 +379,22 @@ public Message<File> receive() throws MessagingException {
 		}
 
 		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<File> filteredFiles = this.scanner.listFiles(this.directory);
-		Set<File> freshFiles = new LinkedHashSet<File>(filteredFiles);
+		Set<File> freshFiles = new LinkedHashSet<>(filteredFiles);
 		if (!freshFiles.isEmpty()) {
 			this.toBeReceived.addAll(freshFiles);
 			if (logger.isDebugEnabled()) {
@@ -421,7 +433,7 @@ public enum WatchEventType {
 
 	private class WatchServiceDirectoryScanner extends DefaultDirectoryScanner implements Lifecycle {
 
-		private final ConcurrentMap<Path, WatchKey> pathKeys = new ConcurrentHashMap<Path, WatchKey>();
+		private final ConcurrentMap<Path, WatchKey> pathKeys = new ConcurrentHashMap<>();
 
 		private WatchService watcher;
 
@@ -547,7 +559,7 @@ else if (event.kind() == StandardWatchEventKinds.OVERFLOW) {
 		}
 
 		private Set<File> walkDirectory(Path directory, final WatchEvent.Kind<?> kind) {
-			final Set<File> walkedFiles = new LinkedHashSet<File>();
+			final Set<File> walkedFiles = new LinkedHashSet<>();
 			try {
 				registerWatch(directory);
 				Files.walkFileTree(directory, new SimpleFileVisitor<Path>() {

spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java

@@ -161,8 +161,7 @@ protected Object doReceive() {
 						.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);

spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java

@@ -35,7 +35,7 @@
 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,14 +63,8 @@
  * @author Venil Noronha
  */
 public abstract class AbstractInboundFileSynchronizingMessageSource<F>
-		extends AbstractFetchLimitingMessageSource<File> implements Lifecycle {
-
-	private volatile boolean running;
-
-	/**
-	 * Should the endpoint attempt to create the local directory? True by default.
-	 */
-	private volatile boolean autoCreateLocalDirectory = true;
+		extends AbstractFetchLimitingMessageSource<File>
+		implements Lifecycle {
 
 	/**
 	 * An implementation that will handle the chores of actually connecting to and synchronizing
@@ -79,14 +73,21 @@ public abstract class AbstractInboundFileSynchronizingMessageSource<F>
 	private final AbstractInboundFileSynchronizer<F> synchronizer;
 
 	/**
-	 * Directory to which things should be synchronized locally.
+	 * The actual {@link LocalFileReadingMessageSource} that monitors the local file system once files are synchronized.
 	 */
-	private volatile File localDirectory;
+	private final LocalFileReadingMessageSource fileSource;
+
+	private volatile boolean running;
+
+	/**
+	 * Should the endpoint attempt to create the local directory? True by default.
+	 */
+	private volatile boolean autoCreateLocalDirectory = true;
 
 	/**
-	 * The actual {@link FileReadingMessageSource} that monitors the local file system once files are synchronized.
+	 * Directory to which things should be synchronized locally.
 	 */
-	private final FileReadingMessageSource fileSource;
+	private volatile File localDirectory;
 
 	private volatile FileListFilter<File> localFileListFilter;
 
@@ -101,13 +102,14 @@ public AbstractInboundFileSynchronizingMessageSource(AbstractInboundFileSynchron
 
 	public AbstractInboundFileSynchronizingMessageSource(AbstractInboundFileSynchronizer<F> synchronizer,
 			Comparator<File> 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 boolean isRunning() {
 	 * @param maxFetchSize the maximum files to fetch.
 	 */
 	@Override
-	public final Message<File> doReceive(int maxFetchSize) {
-		Message<File> message = this.fileSource.receive();
-		if (message == null) {
+	public final AbstractIntegrationMessageBuilder<File> doReceive(int maxFetchSize) {
+		AbstractIntegrationMessageBuilder<File> 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<File> buildFilter() {
 		Pattern completePattern = Pattern.compile("^.*(?<!" + this.synchronizer.getTemporaryFileSuffix() + ")$");
-		return new CompositeFileListFilter<File>(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<File> receptionOrderComparator) {
+			super(receptionOrderComparator);
+		}
+
+		@Override
+		protected AbstractIntegrationMessageBuilder<File> doReceive() {
+			return super.doReceive();
+		}
+
 	}
 
 }