【发布时间】：2023-03-17 11:08:01
【问题描述】：
我有一个用户界面，我可以在其中添加或删除 FTP 服务器（添加端口和用户名..等），这些服务器保存在 postgres 数据库中，我的应用程序将使用 spring MVC 和 Spring Integration 读取数据库中的连接对于动态 FTP（我正在使用委派会话工厂和 Rotating 建议），当我运行它并读取数据库中可用的连接时，该应用程序正在工作，因此将传输我指定的 FTP 目录中可用的内容。我的问题是，如果我使用该接口删除或添加新服务器，除非我停止并再次运行它，否则该应用程序不会使用数据库中持久的新连接，我希望在添加时使其在运行时工作并自动删除服务器。 这是我用于设置集成流程的 conf 类，我不确定是否有任何注释可以使其正常工作。有人可以指导吗？ 如果需要更多信息，请告诉我
@Configuration
@Component
@EnableIntegration
public class FTIntegration {
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(FTIntegration.class);
private static final String CHANNEL_INTERMEDIATE_STAGE = "intermediateChannel";
@Autowired
private IntegrationFlowContext flowContext;
/* pulling the server config from postgres DB*/
private final BranchRepository branchRepository;
@Value("${app.temp-dir}")
private String localTempPath;
public FTIntegration(BranchRepository branchRepository) {
this.branchRepository = branchRepository;
}
@Bean
public Branch myBranch(){
return new Branch();
}
/**
* 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)
.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
*/
public IntegrationFlow fileInboundFlowFromFTPServer(Branch myBranch){
final FtpInboundChannelAdapterSpec sourceSpecFtp = Ftp.inboundAdapter(createNewFtpSessionFactory(myBranch))
.preserveTimestamp(true)
.patternFilter("*.csv")
.deleteRemoteFiles(true)
.maxFetchSize(MAX_MESSAGES_PER_POLL)
.remoteDirectory(myBranch.getFolderPath())
.localDirectory(new File(localTempPath))
.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());
IntegrationFlow flow = IntegrationFlows
.from(sourceSpecFtp, stockInboundPoller)
.transform(File.class, p ->{
// log step
LOG.info("flow=stockInboundFlowFromAFT, message=incoming file: " + p);
return p;
})
.channel(CHANNEL_INTERMEDIATE_STAGE)
.get();
// this.flowContext.registration(flow).id(myBranch.getId().toString()).register().toString();
//this.flowContext.registration(flow).id("fileInb").register();
return flow;
}
@Bean
public IntegrationFlow stockIntermediateStageChannel() {
IntegrationFlow flow = IntegrationFlows
.from(CHANNEL_INTERMEDIATE_STAGE)
.transform(p -> {
//log step
LOG.info("flow=stockIntermediateStageChannel, message=rename file: " + p);
return p;
})
//TODO
.channel(new NullChannel())
.get();
return flow;
}
public DefaultFtpSessionFactory createNewFtpSessionFactory(Branch branch){
final DefaultFtpSessionFactory factory = new DefaultFtpSessionFactory();
factory.setHost(branch.getHost());
factory.setUsername(branch.getUsern());
factory.setPort(branch.getFtpPort());
factory.setPassword(branch.getPassword());
return factory;
}
}
这里是删除服务器的控制器部分。
@Controller
public class BranchController {
private BranchService branchService;
private BranchToBranchForm branchToBranchForm;
//@Autowired
private Branch branch;
@Autowired
private FTIntegration ftIntegration;
private static final Logger LOG = LoggerFactory.getLogger(FTIntegration.class);
private static final String CHANNEL_INTERMEDIATE_STAGE = "intermediateChannel";
@Autowired
private IntegrationFlowContext flowContext;
@Autowired
public void setBranchService(BranchService branchService) {
this.branchService = branchService;
}
@Autowired
public void setBranchToBranchForm(BranchToBranchForm branchToBranchForm) {
this.branchToBranchForm = branchToBranchForm;
}
@RequestMapping( "/")
public String branch(){return "redirect:/branch/list";}
@RequestMapping({"/branch/list","/branch"})
public String listBranches(Model model){
model.addAttribute("branches",branchService.listAll());
return "branch/list";
}
@RequestMapping("/branch/showbranch/{id}")
public String getBranch (@PathVariable String id, Model model){
model.addAttribute("branch", branchService.getById(Long.valueOf(id)));
//addFlowFtp(id);
addFlowftp(id);
return "/branch/showbranch";
}
@RequestMapping("/branch/edit/{id}")
public String edit(@PathVariable String id, Model model){
Branch branch = branchService.getById(Long.valueOf(id));
BranchForm branchForm = branchToBranchForm.convert(branch);
model.addAttribute("branchForm",branchForm);
return "branch/branchform";
}
@RequestMapping("/branch/new")
public String newBranch(Model model){
model.addAttribute("branchForm", new BranchForm());
return "branch/branchform";
}
@RequestMapping(value = "/branch", method = RequestMethod.POST)
public String saveOrUpdateBranch(@Valid BranchForm branchForm, BindingResult bindingResult){
if(bindingResult.hasErrors()){
return "branch/branchform";
}
Branch savedBranch = branchService.saveOrUpdateBranchForm(branchForm);
return "redirect:/branch/showbranch/" + savedBranch.getId();
}
@RequestMapping("/branch/delete/{id}")
private String delete(@PathVariable String id){
branchService.delete(Long.valueOf(id));
flowContext.remove(id);
return "redirect:/branch/list";
}
private void addFlowftp(String name) {
branch = branchService.getById(Long.valueOf(name));
System.out.println(branch.getBranchCode());
IntegrationFlow flow = ftIntegration.fileInboundFlowFromFTPServer(branch);
this.flowContext.registration(flow).id(name).register();
}
【问题讨论】：
标签： java spring-mvc spring-integration