Reputation: 245
I have the following initial spring integration flow:
final var inboundStreamingAdapter = Sftp.inboundStreamingAdapter(new SftpRemoteFileTemplate(sftpSessionFactory))
.patternFilter("*.csv")
.remoteDirectory(sftpRemoteDirectoryDownload);
IntegrationFlows.from(inboundStreamingAdapter,
c -> c.poller(Pollers.trigger(new PeriodicTrigger(10)).maxMessagesPerPoll(1)))
.channel("data")
.split(mySplitter)
.get();
and I have the following files on my sftp server:
.DS_Store
PL_12002_1815_1.csv1
From my understanding adapter should filter out everything what does not end with csv postfix.
What is actually happening is that some of the files (.DS_Store) are sometimes making it through to mySplitter depending what is currently stored on FTP.
As far I can tell problem is in AbstractRemoteFileStreamingMessageSource (used as base class for SftpStreamingMessageSource class):
if (this.filter != null && this.filter.supportsSingleFileFiltering()
&& !this.filter.accept(file.getFileInfo())) { //HERE APPLY A FILTER AND FIND OUT PL_12002_1815_1.csv1 SHOULD NOT BE PROCESSED
if (this.toBeReceived.size() > 0) { // don't re-fetch already filtered files
//HERE WE LOAD ANOTHER FILE FROM THE QUEUE (.DS_Store)
file = poll();
}
else {
file = null;
}
}
//FILTER IS NOT REAPPLIED FURTHER DOWN AND WE CREATE A MESSAGE FOR .DS_Store
if (file != null) {
try {
String remotePath = remotePath(file);
Session<?> session = this.remoteFileTemplate.getSession();
try {
return getMessageBuilderFactory()
.withPayload(session.readRaw(remotePath))
.setHeader(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE, session)
.setHeader(FileHeaders.REMOTE_DIRECTORY, file.getRemoteDirectory())
.setHeader(FileHeaders.REMOTE_FILE, file.getFilename())
.setHeader(FileHeaders.REMOTE_HOST_PORT, session.getHostPort())
.setHeader(FileHeaders.REMOTE_FILE_INFO,
this.fileInfoJson ? file.toJson() : file);
}
I would say this is a bug but maybe I just misconfigured something, can someone please help?
Upvotes: 1
Views: 367
Reputation: 174484
Looks like a bug; the if
should be a while
(with a null check).
Please open an issue on GitHub.
Upvotes: 1