How To Stop Polling InboundChannelAdapter

Im polling files from 2 different directories in 1 server using RotatingServerAdvice and that´s working fine, the problem is that I can´t stop polling once time I start the inboundtest.start (). The main idea is retrive all the files in those directories, and then send inboundtest.stop (), this is the code.

    @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 is waiting for a new file in one of the 2 directories, but thats no the idea, the idea is stop polling when all the files be retrived

From another class I call inbound.start() trouhg the control chanel here the code:

@Autowired
private MessageChannel controlChannel;

public void startProcessingFiles() throws InterruptedException {
    controlChannel.send(new GenericMessage<>("@inboundtest.start()"));
    
}

I was tryong stop with this class but doesn´t works

 @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;
}
}

Answer

OK. Now I see your point. The RotatingServerAdvice does move to other server only when the first is exhausted (by default, see that fair option). So, when you stop it in the advice it cannot go to other dir for fetching any more. You need to think about some other stopping solution. Something what is not tied to the advice and this afterReceive(), somewhere downstream in your flow…

Or you can provide a custom RotationPolicy (extension of StandardRotationPolicy) and in its overridden afterReceive() check for all the dirs processed and then send stop command.

Advertisement

Answer