Spring Integration : Inbound Channel Adapter configuration via annotations

Question

How can I configure the inbound channel adapter via annotations instead of the regular configuration file? I was able to define the bean for the session factory though as under:

@Bean
public DefaultFtpSessionFactory ftpSessionFactory() {
        DefaultFtpSessionFactory ftpSessionFactory = new  
       DefaultFtpSessionFactory();
        ftpSessionFactory.setHost(host);
        ftpSessionFactory.setPort(port);
        ftpSessionFactory.setUsername(username);
        ftpSessionFactory.setPassword(password);
        return ftpSessionFactory;
    }

How can I configure the inbound channel adapter given as under via annotations?

<int-ftp:inbound-channel-adapter id="ftpInbound"
                                 channel="ftpChannel"
                                 session-factory="ftpSessionFactory"
                                 filename-pattern="*.xml"
                                 auto-create-local-directory="true"
                                 delete-remote-files="false"
                                 remote-directory="/"
                                 local-directory="ftp-inbound"
                                 local-filter="acceptOnceFilter">

    <int:poller fixed-delay="60000" max-messages-per-poll="-1">
        <int:transactional synchronization-factory="syncFactory" />
    </int:poller>

</int-ftp:inbound-channel-adapter>

@Artem Bilan The modified code is as under

@EnableIntegration
@Configuration
public class FtpConfiguration {
    @Value("${ftp.host}")
    private String host;
    @Value("${ftp.port}")
    private Integer port;
    @Value("${ftp.username}")
    private String username;
    @Value("${ftp.password}")
    private String password;
    @Value("${ftp.fixed.delay}")
    private Integer fixedDelay;
    @Value("${ftp.local.directory}")
    private String localDirectory;

    private final static Logger LOGGER =   LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());

@Bean
public SessionFactory<FTPFile> ftpSessionFactory() {
    DefaultFtpSessionFactory sessionFactory = new DefaultFtpSessionFactory();
    sessionFactory.setHost(host);
    sessionFactory.setPort(port);
    sessionFactory.setUsername(username);
    sessionFactory.setPassword(password);
    return new CachingSessionFactory<FTPFile>(sessionFactory);
}

@Bean
public FtpInboundFileSynchronizer ftpInboundFileSynchronizer() {
    FtpInboundFileSynchronizer fileSynchronizer = new FtpInboundFileSynchronizer(ftpSessionFactory());
    fileSynchronizer.setDeleteRemoteFiles(false);
    fileSynchronizer.setRemoteDirectory("/");
    fileSynchronizer.setFilter(new FtpSimplePatternFileListFilter("*.xml"));
    return fileSynchronizer;
}

@Bean
@InboundChannelAdapter(value = "ftpChannel",
        poller = @Poller(fixedDelay = "60000", maxMessagesPerPoll = "-1"))
public MessageSource<File> ftpMessageSource() {
    FtpInboundFileSynchronizingMessageSource source =
            new FtpInboundFileSynchronizingMessageSource(ftpInboundFileSynchronizer());
    source.setLocalDirectory(new File(localDirectory));
    source.setAutoCreateLocalDirectory(true);
    source.setLocalFilter(new AcceptOnceFileListFilter<File>());
    return source;
}

}

While running this,I get an exception as under No bean named 'ftpChannel' is defined

Please note that 'channel' keyword in not available while wiring the Inbound channel adapter,its 'value' instead.

I tried wiring the channel with PollableChannel,that also went in vain though. It is as under:

@Bean
public MessageChannel ftpChannel() {
    return new PollableChannel() {
        @Override
        public Message<?> receive() {
            return this.receive();
        }

        @Override
        public Message<?> receive(long l) {
            return null;
        }

        @Override
        public boolean send(Message<?> message) {
            return false;
        }

        @Override
        public boolean send(Message<?> message, long l) {
            return false;
        }
    };
}

I got an error "failed to send message within timeout: -1".Am I doing something wrong still?

What I'm looking for is to wire up all the beans on application start-up, and then expose some method to start polling the server,process them and then delete them from local,something like this

public void startPollingTheServer() {
  getPollableChannel().receive();
}

where getPollableChannel() gives me the bean I had wired for Polling.

Artem Bilan Artem Bilan · Accepted Answer · 2016-07-14T18:05:40

There is an @InboundChannelAdapter for you.

@Bean
public FtpInboundFileSynchronizer ftpInboundFileSynchronizer() {
    FtpInboundFileSynchronizer fileSynchronizer = new FtpInboundFileSynchronizer(ftpSessionFactory());
    fileSynchronizer.setDeleteRemoteFiles(false);
    fileSynchronizer.setRemoteDirectory("/");
    fileSynchronizer.setFilter(new FtpSimplePatternFileListFilter("*.xml"));
    return fileSynchronizer;
}

@Bean
@InboundChannelAdapter(channel = "ftpChannel")
public MessageSource<File> ftpMessageSource() {
    FtpInboundFileSynchronizingMessageSource source =
            new FtpInboundFileSynchronizingMessageSource(ftpInboundFileSynchronizer());
    source.setLocalDirectory(new File("ftp-inbound"));
    source.setAutoCreateLocalDirectory(true);
    source.setLocalFilter(new AcceptOnceFileListFilter<File>());
    return source;
}

Plus take a look into the Reference Manual.

Also pay attention, please, for Java DSL for Spring Integration, where the same might look like:

@Bean
public IntegrationFlow ftpInboundFlow() {
    return IntegrationFlows
            .from(s -> s.ftp(this.ftpSessionFactory)
                            .preserveTimestamp(true)
                            .remoteDirectory("ftpSource")
                            .regexFilter(".*\\.txt$")
                            .localFilename(f -> f.toUpperCase() + ".a")
                            .localDirectory(this.ftpServer.getTargetLocalDirectory()),
                    e -> e.id("ftpInboundAdapter").autoStartup(false))
            .channel(MessageChannels.queue("ftpInboundResultChannel"))
            .get();
}

Spring Integration : Inbound Channel Adapter configuration via annotations

1 Answers