我使用 RotatingServerAdvice 从 1 个服务器中的 2 个不同目录轮询文件，并且工作正常，问题是一旦我启动 inboundtest.start ()，我就无法停止轮询。主要思想是检索那些目录下的所有文件，然后发送inboundtest.stop()，这就是代码。
@Bean
public SessionFactory<LsEntry> sftpSessionFactory() {
DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(false);
factory.setHost(host);
factory.setPort(port);
factory.setUser(user);
factory.setPassword(password);
factory.setAllowUnknownKeys(true);
//factory.setTestSession(true);
return factory;
}
@Bean
public SftpInboundFileSynchronizer sftpInboundFileSynchronizer() {
SftpInboundFileSynchronizer fileSynchronizer = new SftpInboundFileSynchronizer(sftpSessionFactory());
fileSynchronizer.setDeleteRemoteFiles(true);
fileSynchronizer.setRemoteDirectory(sftpRemoteDirectory);
fileSynchronizer.setFilter(new SftpRegexPatternFileListFilter(".*?\\.(txt|TXT?)"));
return fileSynchronizer;
}
@Bean(name = "sftpMessageSource")
@EndpointId("inboundtest")
@InboundChannelAdapter(channel = "sftpChannel",poller = @Poller("fileReadingMessageSourcePollerMetadata"), autoStartup = "false")
public MessageSource<File> sftpMessageSource() {
SftpInboundFileSynchronizingMessageSource source =
new SftpInboundFileSynchronizingMessageSource(sftpInboundFileSynchronizer());
source.setLocalDirectory(new File(sftpLocalDirectoryDownloadUpload));
source.setAutoCreateLocalDirectory(true);
source.setLocalFilter(new AcceptOnceFileListFilter<File>());
return source;
}
@Bean
public DelegatingSessionFactory<LsEntry> sessionFactory() {
Map<Object, SessionFactory<LsEntry>> factories = new LinkedHashMap<>();
factories.put("one", sftpSessionFactory());
// use the first SF as the default
return new DelegatingSessionFactory<LsEntry>(factories, factories.values().iterator().next());
}
@Bean
public RotatingServerAdvice advice() {
List<RotationPolicy.KeyDirectory> keyDirectories = new ArrayList<>();
keyDirectories.add(new RotationPolicy.KeyDirectory("one", sftpRemoteDirectory));
keyDirectories.add(new RotationPolicy.KeyDirectory("one", sftpRemoteDirectoryNonUpload));
return new RotatingServerAdvice(sessionFactory(), keyDirectories, false);
}
@Bean
MessageChannel controlChannel() {
return new DirectChannel();
}
@Bean
@ServiceActivator(inputChannel = "controlChannel")
ExpressionControlBusFactoryBean controlBus() {
return new ExpressionControlBusFactoryBean();
}
@Bean
public PollerMetadata fileReadingMessageSourcePollerMetadata() {
PollerMetadata meta = new PollerMetadata();
meta.setTrigger(new PeriodicTrigger(1000));
meta.setAdviceChain(List.of(advice()));
meta.setMaxMessagesPerPoll(1);
meta.setErrorHandler(throwable -> new IOException());
return meta;
}
Allways 正在等待 2 个目录之一中的新文件，但这不是想法，想法是在检索所有文件时停止轮询
从另一个类我调用 inbound.start() trouhg 控制通道这里的代码：
@Autowired
private MessageChannel controlChannel;
public void startProcessingFiles() throws InterruptedException {
controlChannel.send(new GenericMessage<>("@inboundtest.start()"));
}
我试图停止这门课，但不起作用
@Component
public class StopPollingAdvice implements ReceiveMessageAdvice {
@Autowired
private MessageChannel controlChannel;
@Override
public Message<?> afterReceive(Message<?> message, Object o) {
System.out.println("There is no more files, stopping connection" + message.getPayload());
if(message == null) {
System.out.println("There is no more files, stopping connection" + message.getPayload());
Message operation = MessageBuilder.withPayload("@inboundtest.stop()").build();
controlChannel.send(operation);
}
return message;
}
}