How to invoke sftp:outbound-gateway from spring batch?

Question

I would like to invoke sftp:outbound-gateway from batch tasklet in order to download a file from sftp server. I've seen other posts related to this subject but I'm not sure what am I doing wrong. Could anybody give me a hint based on my configuration? My batch works so the problem is just to ivoke the sftp component in batch step. I've marked the Spring Integration section with comment so it is easier to read just a relevant configuration.

I can see in my logs: DEBUG [o.s.i.e.SourcePollingChannelAdapter] Received no Message during the poll, returning 'false'. So I am not receiving a file but why?

Thanks in advance for your time spend on analysis!

<bean id="ftsSftpClientFactory" class="org.springframework.integration.sftp.session.DefaultSftpSessionFactory">
    <property name="host" value="${my.import.sftp.localhost}"/>
    <property name="user" value="${my.import.sftp.username}"/>
    <property name="password" value="${my.import.sftp.passwort}"/>
</bean>

  <!-- Start: Spring Integration -->

    <int:channel id="replyChannel" >
        <int:queue/>
    </int:channel>

    <int:channel id="requestChannel" />

    <int-sftp:outbound-gateway id="sftpGateway"
                               session-factory="ftsSftpClientFactory"
                               request-channel="requestChannel"
                               reply-channel="replyChannel"
                               auto-startup="true"
                               command="get"
                               command-options="-P"
                               expression="payload"
                               remote-directory="."
                               local-directory="${my.import.sftp.copy.file.destinationpath}">
    </int-sftp:outbound-gateway>

<bean name="copyFileTasklet" class="com.mydomain.CopyFileTasklet">
    <property name="channel" ref="replyChannel" />
    <property name="pollableChannel" ref="requestChannel" />
</bean>

<!-- Start: Spring Batch -->
<bean name="myImportTask" class="com.mydomain.MyImportTask">
    <property name="job" ref="unternehmungImportJob"/>
    <property name="jobLauncher" ref="jobLauncher"/>
</bean>

<bean id="jobDetail"
      class="com.mydomain.MyImportJob">
    <property name="myImportTask" ref="myImportTask" />
</bean>

<!--suppress SpringBatchModel -->
<batch:job id="myImportJob">
    <batch:step id="copy-file-step" next="my-import-step">
        <batch:tasklet ref="copyFileTasklet"/>
    </batch:step>
    <batch:step id="my-import-step">
        <batch:tasklet>
            <batch:chunk reader="myItemReader"
                         writer="myItemWriter"
                         commit-interval="10000">
                <!--
                skip-limit="10000"
                <batch:skippable-exception-classes>
                   <batch:include class="java.lang.Exception"/>
                   <batch:exclude class="java.io.FileNotFoundException"/>
                </batch:skippable-exception-classes> -->
            </batch:chunk>
            <batch:transaction-attributes isolation="DEFAULT" propagation="REQUIRED"/>
        </batch:tasklet>
    </batch:step>
</batch:job>

<bean id="myItemReader" scope="step" class="org.springframework.batch.item.file.FlatFileItemReader">
    <property name="linesToSkip" value="1"/>
    <property name="encoding" value="${my.import.batch.encoding}" />
    <property name="resource" value="${my.import.batch.input.resource}"/>
    <property name="lineMapper">
        <bean class="org.springframework.batch.item.file.mapping.DefaultLineMapper">
            <property name="lineTokenizer" ref="lineTokenizer"/>
            <property name="fieldSetMapper">
                <bean class="com.mydomain.MyImportMapper"/>
            </property>
        </bean>
    </property>
</bean>

<bean id="myItemWriter" class="com.mydomain.MyItemWriter">
    <property name="myApplicationService" ref="defaultmyApplicationService" />
</bean>

<bean id="lineTokenizer" class="com.mydomain.DelimitedLineTokenizerWithEOF">
    <property name="delimiter" value="${my.import.batch.delimiter}" />
    <property name="eofMarker" value="${my.import.batch.eof.marker}" />
</bean>

public class CopyFileTasklet implements Tasklet {

private MessageChannel requestChannel;

private PollableChannel replyChannel;

public void setRequestChannel(MessageChannel requestChannel) {
    this.requestChannel = requestChannel;
}

public void setReplyChannel(PollableChannel replyChannel) {
    this.replyChannel = replyChannel;
}

@Override
public RepeatStatus execute(StepContribution stepContribution, ChunkContext chunkContext) throws Exception {

    Message<?> result = replyChannel.receive(10000);
    Object file = result.getPayload();       
    return RepeatStatus.FINISHED;
}

}

Answer 1

Your issue that you don't inititate Integration Flow from your custom Tasklet. Of course you can't receive anything from the replyChannel, if you haven't sent request before.

If you just need to process Integration Flow and get result from it, it would be better to use POJI <gateway> from that Tasklet:

public interface SftpGateway {

   File download(String fileName);

}

<gateway id="sftpGateway" service-interface="com.my.proj.SftpGateway"
    default-request-channel="requestChannel"/>


<bean name="copyFileTasklet" class="com.mydomain.CopyFileTasklet">
    <property name="sftpGateway" ref="sftpGateway" />
</bean>

Something like that.