我需要动态地将不同目录中的特定文件从不同的sftp服务器提取到spring引导应用程序的本地服务器目录 .
我将路径和文件模式存储在postgres数据库中 . 我得到了一切工作,但我不知道如何根据spring集成sftp入站适配器的远程目录动态定义文件过滤器模式,以便不会拉出该特定目录中的所有xml文件 .
我使用RotatingServerAdvice和DelegatingSessionFactory来访问动态目录和服务器 .
对于动态文件模式过滤器,我尝试过使用
.filterFunction(f -> do_some_filtering_based_on_filename(f)
我想读取该文件的远程目录,但f来自ChannelSftp.LsEntry类型,并且没有远程目录的字段 . 否则我会从数据库加载配置数据,搜索路径并应用文件模式 .
我的方案有更好的方法吗?
SFTP example:
127.0.0.1:22
目录:root / partner1
... test_p1_2343545.xml ... test_p1_453453.xml ... don_t_pull_this_file_453453.xml
127.0.0.2:22
目录:root / partner2
... companyname_2343545.xml ... companyname_453453.xml ... don_t_pull_this_file_3434.xml
Database configuration example:
| URL | PATH | FILE_PATTERN |
|-----------|-----------|-------------------|
| 127.0.0.1 | /partner1 | test_p1_*.xml |
| 127.0.0.2 | /partner2 | companyname_*.xml |
Adapter Class of my spring boot application with working code but pulling all xml files because of .patternFilter("*.xml"):
import com.jcraft.jsch.ChannelSftp;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.Pollers;
import org.springframework.integration.dsl.SourcePollingChannelAdapterSpec;
import org.springframework.integration.expression.FunctionExpression;
import org.springframework.integration.file.remote.aop.RotatingServerAdvice;
import org.springframework.integration.file.remote.session.DelegatingSessionFactory;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.sftp.dsl.Sftp;
import org.springframework.integration.sftp.dsl.SftpInboundChannelAdapterSpec;
import org.springframework.integration.sftp.session.DefaultSftpSessionFactory;
import org.springframework.messaging.MessageChannel;
import org.springframework.stereotype.Component;
import java.io.File;
import java.time.Instant;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
/**
* flow.
*/
@Configuration
@Component
public class SFTIntegration {
public static final String TIMEZONE_UTC = "UTC";
public static final String TIMESTAMP_FORMAT_OF_FILES = "yyyyMMddHHmmssSSS";
public static final String TEMPORARY_FILE_SUFFIX = ".part";
public static final int POLLER_FIXED_PERIOD_DELAY = 5000;
public static final int MAX_MESSAGES_PER_POLL = 100;
private static final Logger LOG = LoggerFactory.getLogger(SFTIntegration.class);
private static final String CHANNEL_INTERMEDIATE_STAGE = "intermediateChannel";
/** database access repository */
private final PartnerConfigRepo partnerConfigRepo;
@Value("${app.tmp-dir}")
private String localTemporaryPath;
public SFTIntegration(final PartnerConfigRepo partnerConfigRepo) {
this.partnerConfigRepo = partnerConfigRepo;
}
/**
* The default poller with 5s, 100 messages, RotatingServerAdvice and transaction.
*
* @return default poller.
*/
@Bean(name = PollerMetadata.DEFAULT_POLLER)
public PollerMetadata poller() {
return Pollers
.fixedDelay(POLLER_FIXED_PERIOD_DELAY)
.advice(advice())
.maxMessagesPerPoll(MAX_MESSAGES_PER_POLL)
.transactional()
.get();
}
/**
* The direct channel for the flow.
*
* @return MessageChannel
*/
@Bean
public MessageChannel stockIntermediateChannel() {
return new DirectChannel();
}
/**
* Get the files from a remote directory. Add a timestamp to the filename
* and write them to a local temporary folder.
*
* @return IntegrationFlow
*/
@Bean
public IntegrationFlow stockInboundFlowFromSFTPServer() {
// Source definition
final SftpInboundChannelAdapterSpec sourceSpec = Sftp.inboundAdapter(delegatingSFtpSessionFactory())
.preserveTimestamp(true)
.patternFilter("*.xml")
//.filterFunction(f -> do_some_filtering_based_on_filename(f, delegatingSFtpSessionFactory().getSession()))
//.filter(new ModifiedFilter())
//.filterExpression("#remoteDirectory")
.deleteRemoteFiles(true)
.maxFetchSize(MAX_MESSAGES_PER_POLL)
.remoteDirectory("/")
.localDirectory(new File(localTemporaryPath))
.temporaryFileSuffix(TEMPORARY_FILE_SUFFIX)
.localFilenameExpression(new FunctionExpression<String>(s -> {
final int fileTypeSepPos = s.lastIndexOf('.');
return
DateTimeFormatter
.ofPattern(TIMESTAMP_FORMAT_OF_FILES)
.withZone(ZoneId.of(TIMEZONE_UTC))
.format(Instant.now())
+ "_"
+ s.substring(0, fileTypeSepPos)
+ s.substring(fileTypeSepPos);
}));
// Poller definition
final Consumer<SourcePollingChannelAdapterSpec> stockInboundPoller = endpointConfigurer -> endpointConfigurer
.id("stockInboundPoller")
.autoStartup(true)
.poller(poller());
return IntegrationFlows
.from(sourceSpec, stockInboundPoller)
.transform(File.class, p -> {
// log step
LOG.info("flow=stockInboundFlowFromAFT, message=incoming file: " + p);
return p;
})
.channel(CHANNEL_INTERMEDIATE_STAGE)
.get();
}
@Bean
public IntegrationFlow stockIntermediateStageChannel() {
return IntegrationFlows
.from(CHANNEL_INTERMEDIATE_STAGE)
.transform(p -> {
//log step
LOG.info("flow=stockIntermediateStageChannel, message=rename file: " + p);
return p;
})
//TODO
.channel(new NullChannel())
.get();
}
public DefaultSftpSessionFactory createNewSftpSessionFactory(final PartnerConfigEntity pc) {
final DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(
false);
factory.setHost(pc.getServerIp());
factory.setPort(pc.getPort());
factory.setUser(pc.getUsername());
factory.setPassword(pc.getPassword());
factory.setAllowUnknownKeys(true);
return factory;
}
@Bean
public DelegatingSessionFactory<ChannelSftp.LsEntry> delegatingSFtpSessionFactory() {
final List<PartnerConfigEntity> partnerConnections = partnerConfigRepo
.findByTypeAndActiveIsTrue(PartnerConfigType.STOCK);
if (partnerConnections.isEmpty()) {
return null;
}
final Map<Object, SessionFactory<ChannelSftp.LsEntry>> factories = new LinkedHashMap<>(10);
for (PartnerConfigEntity pc : partnerConnections) {
// create a factory for every key containing server type, url and port
if (factories.get(pc.getKey()) == null) {
factories.put(pc.getKey(), createNewSftpSessionFactory(pc));
}
}
// use the first SF as the default
return new DelegatingSessionFactory<>(factories, factories.values().iterator().next());
}
@Bean
public RotatingServerAdvice advice() {
final List<PartnerConfigEntity> partnerConnections = partnerConfigRepo.findByTypeAndActiveIsTrue("A_TYPE");
final List<RotatingServerAdvice.KeyDirectory> keyDirectories = new ArrayList<>();
for (PartnerConfigEntity pc : partnerConnections) {
keyDirectories
.add(new RotatingServerAdvice.KeyDirectory(pc.getKey(), pc.getServerPath()));
}
return new RotatingServerAdvice(delegatingSFtpSessionFactory(), keyDirectories, true);
}
// private Boolean do_some_filtering_based_on_filename(final ChannelSftp.LsEntry f,
// final Session<ChannelSftp.LsEntry> session) {
// Object instance = session.getClientInstance();
// System.out.println(f);
// return true;
// }
//
// private class ModifiedFilter<F> extends AbstractFileListFilter<F> {
// private final Logger log = LoggerFactory.getLogger(ModifiedFilter.class);
//
// @Override public boolean accept(final F file) {
// log.info(file.toString());
// return false;
// }
// }
}
2 回答
我建议你实现
SftpSimplePatternFileListFilter
逻辑的自定义变体,并有一个用于在运行时更改模式的setter .然后你去
RotatingServerAdvice
并注入一个自定义RotationPolicy
,让我们说StandardRotationPolicy
的副本注入你的自定义FileListFilter
然后在configureSource()
之后:您执行数据库查询以获取
this.current.getDirectory()
的模式并将其设置为FileListFilter
.必须在代码中提到的
Sftp.inboundAdapter()
中使用相同的过滤器,而不是.patternFilter("*.xml")
.我认为我们需要将
StandardRotationPolicy
中的所有属性设置为protected
或使用getter来允许继承者访问 . 随意就此事提出JIRA:https://jira.spring.io/browse/INT!希望这会有所帮助 .
这是Artem Bilan提供的解决方案的源代码 . 如果有人想看到它 .
我复制了Spring集成类,并按照Artem Bilan的建议对其进行了修改 .
@Artem Bilan非常感谢你!