spring-projects/spring-integration

Hello, can anyone help me / give me advice how to setup the following;
I have a SFTP server which i'm polling (every 10sec?) for a new file, than I'll retrieve this file locally, and invoke a handler method (e.g. to read the file / create a flux of elements, save it to the local database, etc)
Afterwards I want to pass along that flux of elements to another server (using RSocketOutboundGateway??)
so the second service retrieves that same flux of elements, and can do some processing as well.

Currently I have the following setup;

@Bean
    public SftpInboundFileSynchronizer sftpInboundFileSynchronizer() {
        SftpInboundFileSynchronizer fileSynchronizer = new SftpInboundFileSynchronizer(sftpSessionFactory());
        fileSynchronizer.setDeleteRemoteFiles(true);
        fileSynchronizer.setRemoteDirectory(excelRemoteDirectory);
        fileSynchronizer.setPreserveTimestamp(true);
        fileSynchronizer.setFilter(new SftpSimplePatternFileListFilter("testdennis*.xlsx"));

        return fileSynchronizer;
    }

    @Bean(name = "directChannel")
    public MessageChannel directChannel() {
        return new DirectChannel();
    }

    @Bean
    @InboundChannelAdapter(channel = "directChannel", poller = @Poller(fixedDelay = "10000"))
    public MessageSource<File> myMessageSource() {
        SftpInboundFileSynchronizingMessageSource messageSource = new SftpInboundFileSynchronizingMessageSource(sftpInboundFileSynchronizer());
        messageSource.setLocalDirectory(new File("sftp-inbound"));
        messageSource.setAutoCreateLocalDirectory(true);
        messageSource.setLocalFilter(new SimplePatternFileListFilter("testdennis*.xlsx"));

        return messageSource;
    }

    @Bean(name = "fluxMessageChannel")
    public MessageChannel fluxMessageChannel() {
        return new FluxMessageChannel();
    }

    @Bean
    @Gateway(requestChannel = "fluxMessageChannel")
    public RSocketOutboundGateway rsocketOutboundGateway() {
        RSocketOutboundGateway rsocketOutboundGateway = new RSocketOutboundGateway("my-rsocket-route");
        rsocketOutboundGateway.setInteractionModel(RSocketInteractionModel.requestChannel);
        rsocketOutboundGateway.setExpectedResponseType(Void.class);
        rsocketOutboundGateway.setClientRSocketConnector(clientRSocketConnector());

        return rsocketOutboundGateway;
    }

    @Bean
    public ClientRSocketConnector clientRSocketConnector() {
        return new ClientRSocketConnector(URI.create("ws://localhost:8880"));
    }

With the following (intermediate) handler / service activator defined;

@Service
public class ExcelFileHandler {

    @ServiceActivator(inputChannel = "directChannel", outputChannel = "fluxMessageChannel")
    public Flux<MyElement> handleMessage(Message<File> message) {
        File file = message.getPayload();
        log.info("Handling excel file '{}'", file.getName());

        return Flux.just(....);
    }
}