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

Optimize AbstractMessageSources

Browse Source 
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`
This commit is contained in:
Artem Bilan
2018-01-17 18:14:29 -05:00
committed by Gary Russell
parent e6ec86c505
commit 86c76999f6
4 changed files with 93 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<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 @@ public abstract class AbstractMessageSource<T> extends AbstractExpressionEvaluat
}
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 @@ public abstract class AbstractMessageSource<T> extends AbstractExpressionEvaluat
}
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 @@ public abstract class AbstractMessageSource<T> 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();

44
spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java
View File

@@ -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<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 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
@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 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<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 class FileReadingMessageSource extends IntegrationObjectSupport implement
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 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
}
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>() {

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

@@ -161,8 +161,7 @@ public abstract class AbstractRemoteFileStreamingMessageSource<F>
.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);

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

@@ -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<F>
extends AbstractFetchLimitingMessageSource<File> implements Lifecycle {
extends AbstractFetchLimitingMessageSource<File>
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<F> 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<F>
*/
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<F> 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<File> localFileListFilter;
/**
@@ -101,13 +102,14 @@ public abstract class AbstractInboundFileSynchronizingMessageSource<F>
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 abstract class AbstractInboundFileSynchronizingMessageSource<F>
* @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();
}
}
}