Argthtjtr

I have created an application that adds inbound adaptors at run time for ftp servers and registers them for removing at a certain stage if needed, this app will pull a csv file from ftp server(s) and place it in my local in a folder having the name of the ftp server, so every server I add will have a separate local folder created and the csv file is saved in it, now this is accomplished smoothly, the second part is I want to change the format of that file and then send it back to the respective server, so basically I need to use outbound adaptor, in this case I would need to create outbound adaptors at run time at the same time when creating inbound adaptor or adding a server, this should be done through controller same as the inbound, I searched for possible solutions and tried a one that is below but did not work or did not perform any sending of files to destination, any solution on how I can accomplish this?
In the configuration class I added the below:

public IntegrationFlow ftpOutboundFlow(Branch myBranch){



    return IntegrationFlows.from(OUTBOUND_CHANNEL)

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

                    .useTemporaryFileName(true)

                    .remoteFileSeparator("/")

                    //.fileNameExpression("BEY/FEFOexportBEY.csv")

            .remoteDirectory(myBranch.getFolderPath()))

            .get();



}



@MessagingGateway

public interface MyGateway {



    @Gateway(requestChannel = OUTBOUND_CHANNEL)

    void sendToFtp(File file);



}







 @Bean

    public IntegrationFlow ftpOutboundChannel() {



        return IntegrationFlows.from(OUTBOUND_CHANNEL)

                .publishSubscribeChannel(p -> p

                        .subscribe(t -> t.handle(System.out::println)))

                /*.transform(p -> {

                    LOG.info("Outbound intermediate Channel, message=rename file:" + p);

                    return p;

                })*/

                .channel(new NullChannel())

                .get();

    }

And in the Controller class

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



}



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

      //  this.flowContext.registration(flow).id(name).register();

        myGateway.sendToFtp(new File("BEY/FEFOexportBEY.csv"));

    }

Here is what I get in the consol as error when I enable the register before sending the file:

java.lang.IllegalArgumentException: An IntegrationFlow 'IntegrationFlowRegistration{integrationFlow=StandardIntegrationFlow{integrationComponents={org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizer@aadab28=98.org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizer#0, org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean@aa290b3=stockInboundPoller, org.springframework.integration.transformer.MethodInvokingTransformer@e5f85cd=98.org.springframework.integration.transformer.MethodInvokingTransformer#0, 98.channel#0=98.channel#0, org.springframework.integration.config.ConsumerEndpointFactoryBean@319dff2=98.org.springframework.integration.config.ConsumerEndpointFactoryBean#0}}, id='98', inputChannel=null}' with flowId '98' is already registered.

An existing IntegrationFlowRegistration must be destroyed before overriding.

After my second trial where I removed the registration from the first method and only tried with the second method but nothing was sent to the FTP:

GenericMessage [payload=BEYFEFOexportBEY.csv, headers={id=43cfc2db-41e9-0866-8e4c-8e95968189ff, timestamp=1542702869007}]

2018-11-20 10:34:29.011  INFO 13716 --- [nio-8081-exec-6] f.s.s.configuration.FTIntegration        : Outbound intermediate Channel, message=rename file:BEYFEFOexportBEY.csv

You need to perform this.flowContext.registration(flow).id(name).register(); before sending the file via myGateway.sendToFtp(new File("BEY/FEFOexportBEY.csv"));.

That's first.

Other concern I see in your code that you have a ftpOutboundChannel bean for an IntegrationFlow which is subscribed to the same OUTBOUND_CHANNEL. If that one is not declared as a PublishSubscribeChannel, then you'll end up with the round-robin distribution. And I believe you would like to have file sent to the FTP and logged. So, indeed you need to declare that channel as a PublishSubscribeChannel.

You don't have any error because that OUTBOUND_CHANNEL has your ftpOutboundChannel IntegrationFlow as subscriber.