Btukfyl

I have an issue where I'm polling data from FTP server (s) to local folder for example file FEFOexportBEY.csv, once this file is in remote directory, I'm polling it to my local smoothly with not issue, then I'm using this file to generate a new file called finalBEY.csv in my local folder, then I want to push down stream this file to the ftp folder where I got the original one, my problem is that I was able to send the finalBEY.csv one time only, this process will happen frequently so if I pull FEFOexportBEY.csv 3 times per day then I will generate finalBEY.csv three times and will send the same one three times to the down stream, it is not working with me to do so, it is only sending it one time and if I try to remove the finalBEY.csv and generated a new one , the app is not sending it, below is my whole code in config file and controller file, please assist on letting me know how I can keep on watching or polling the local folder for example BEY for new finalBEY.csv and send it to destination.

     @Configuration



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



        private static final String OUTBOUND_CHANNEL = "outboundChannel";



        /* 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

         */





          //@Bean

           public IntegrationFlow fileInboundFlowFromFTPServer(Branch myBranch){



            final FtpInboundChannelAdapterSpec sourceSpecFtp = Ftp.inboundAdapter(createNewFtpSessionFactory(myBranch))

                    .preserveTimestamp(true)

                    //.patternFilter("*.csv")

                    .maxFetchSize(MAX_MESSAGES_PER_POLL)

                    .remoteDirectory(myBranch.getFolderPath())

                    .regexFilter("FEFOexport"+myBranch.getBranchCode()+".csv")

                    .deleteRemoteFiles(true)

                    .localDirectory(new File(myBranch.getBranchCode()))

                    .localFilter(new AcceptAllFileListFilter())

                        .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();



            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;



        }



        /*

        * Creating the outbound adaptor

        *

        * */



        public IntegrationFlow localToFtpFlow(Branch myBranch){

    return IntegrationFlows.from(Files.inboundAdapter(new File(myBranch.getBranchCode()))

            .filter(new FileSystemPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "foo")),

                    e -> e.poller(Pollers.fixedDelay(10_000)))

            .log()

            .handle(Ftp.outboundAdapter(createNewFtpSessionFactory(myBranch), FileExistsMode.REPLACE)

                    .useTemporaryFileName(true)

                    .autoCreateDirectory(true)

                    .remoteDirectory(myBranch.getFolderPath()))

            .get();

}





public interface SendToFtpDirect{

          void send(byte bytes, @Header(FileHeaders.FILENAME) String filename);



             }



        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 class is:

    @Controller

public class BranchController {



    private BranchService branchService;

    private BranchToBranchForm branchToBranchForm;



    //@Autowired

    private Branch branch;



    @Autowired

    private FTIntegration.MyGateway myGateway;



    @Autowired

    private FTIntegration ftIntegration;



    @Autowired

    private IntegrationFlowContext flowContext;



private FTIntegration.SendToFtpDirect gate;



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

       addFlowftpOutbound(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";

    }



    //@PostMapping

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

        flowContext.remove(id+"o");

        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();

    }



   private void addFlowftpOutbound(String name) {

    branch = branchService.getById(Long.valueOf(name));

    System.out.println(branch.getBranchCode());

    IntegrationFlow flow = ftIntegration.localToFtpFlow(branch);//ftpOutboundFlow(branch);

    this.flowContext.registration(flow).id(name +"o").register();

    //gate.send("BEY".getBytes(),"final"+ branch.getBranchCode()+ ".csv" );



}





}

If you are using a recent version, a new version of the file should pass, as long as its modified timestamp has changed.

See the documentation.

You can use the local-filter attribute to configure the behavior of the local file system filter. Starting with version 4.3.8, a FileSystemPersistentAcceptOnceFileListFilter is configured by default. This filter stores the accepted file names and modified timestamp in an instance of the MetadataStore strategy (see Section 12.5, “Metadata Store”) and detects changes to the local file modified time. The default MetadataStore is a SimpleMetadataStore, which stores state in memory.

Check what's in the local filter; also turn on DEBUG logging to see if that provides you with more information.

EDIT

This works fine for me...

@SpringBootApplication

public class So53521657Application {



    public static void main(String args) {

        SpringApplication.run(So53521657Application.class, args);

    }



    @Bean

    public CachingSessionFactory<FTPFile> sf() {

        DefaultFtpSessionFactory sf = new DefaultFtpSessionFactory();

        sf.setHost("10.0.0.3");

        sf.setUsername("ftptest");

        sf.setPassword("ftptest");

        return new CachingSessionFactory<>(sf);

    }



    @Bean

    public IntegrationFlow webToFtpFlow() {

        return IntegrationFlows.from(SendToFtpDirect.class)

                .log()

                .handle(Ftp.outboundAdapter(sf()).remoteDirectory("foo"))

                .get();

    }



    @Bean

    public IntegrationFlow ftpToLocalFlow() {

        return IntegrationFlows.from(Ftp.inboundAdapter(sf())

                    .remoteDirectory("foo")

                    .deleteRemoteFiles(true)

                    .localFilter(new SimplePatternFileListFilter("*.csv"))

                    .localDirectory(new File("/tmp/foo")), e ->

                        e.poller(Pollers.fixedDelay(5_000)))

                .log()

                .<File>handle((p, h) -> {

                        File newFile = new File("/tmp/bar/" + p.getName().replace(".csv", ".txt"));

                        newFile.delete();

                        System.out.println("renaming " + p + " to " + newFile);

                        p.renameTo(newFile);

                        return p;

                    })

                .log()

                .nullChannel();

    }



    @Bean

    public IntegrationFlow localToFtpFlow() {

        return IntegrationFlows.from(Files.inboundAdapter(new File("/tmp/bar"))

                        .filter(new FileSystemPersistentAcceptOnceFileListFilter(

                                new SimpleMetadataStore(), "foo")), e ->

                    e.poller(Pollers.fixedDelay(10_000)))

                .log()

                .handle(Ftp.outboundAdapter(sf())

                        .remoteDirectory("bar"))

                .get();

    }



}



@RestController

@DependsOn("webToFtpFlow")

class Controller {



    @Autowired

    private SendToFtpDirect gate;



    @GetMapping(path = "/send/{name}")

    public String send(@PathVariable String name) {

        gate.send("foo".getBytes(), name + ".csv");

        return name + " sent";

    }



}



interface SendToFtpDirect {



    void send(byte bytes, @Header(FileHeaders.FILENAME) String filename);



}