Spring-Integration-SFTP 다운로드 및 프로세스 파일

Question

아래 제안 사항을 사용했으며 원격 서버에서 파일을 다운로드 할 수있었습니다. 다운로드가 완료된 후 파일을 처리하려면 어떻게해야합니까? 파일이 내 로컬 시스템에

spring integration + cron + quartz in cluster?

ApplicationContext.xml을 다운로드하면 누군가가 제어를 구할 수있는 곳은 어떻게/날 안내 주실 래요

<?xml version="1.0" encoding="UTF-8"?> 

<beans xmlns="http://www.springframework.org/schema/beans" 
    xmlns:context="http://www.springframework.org/schema/context" 
    xmlns:int="http://www.springframework.org/schema/integration" 
    xmlns:int-sftp="http://www.springframework.org/schema/integration/sftp" 
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 
    xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd 
     http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd 
     http://www.springframework.org/schema/integration/sftp http://www.springframework.org/schema/integration/sftp/spring-integration-sftp.xsd 
     http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.1.xsd">  

    <context:component-scan base-package="com.reports"/>  

    <context:property-placeholder location="classpath:reports.properties"/> 

    <bean class="org.springframework.integration.sftp.session.DefaultSftpSessionFactory" 
     id="sftpSessionFactory"> 
     <property name="host" value="${sftp.host}"/> 
     <property name="privateKey" value="${sftp.key.location}"/> 
     <property name="port" value="${sftp.port}"/> 
     <property name="user" value="${sftp.user}"/> 
    </bean> 

    <int-sftp:inbound-channel-adapter 
     id="sftpInboundAdapter" 
     auto-startup="false" 
     auto-create-local-directory="false" 
     channel="receiveChannel" 
     delete-remote-files="false" 
     filter = "customfilter" 
     remote-file-separator="/" 
     local-directory="${local.directory}" 
     remote-directory="${remote.directory}" 
     session-factory="sftpSessionFactory" 
     preserve-timestamp="true" 
     local-filter="acceptAll"> 

     <int:poller trigger="sftp-trigger"/> 
    </int-sftp:inbound-channel-adapter> 

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

    <bean id="customfilter" class="com.reports.filters.ReportsFileListFilter"/> 
    <bean id="acceptAll" class="org.springframework.integration.file.filters.AcceptAllFileListFilter" />   
    <bean id="sftp-trigger" class="com.reports.jobs.ReportsTrigger"/> 

    <!-- Quartz job --> 
    <!-- (1) Trigger - This is used in the Scheduler below --> 
    <bean id="ReportCronTrigger" class="org.springframework.scheduling.quartz.CronTriggerFactoryBean"> 
     <property name="jobDetail" ref="ReportsJob"/> 
     <property name="cronExpression" value="${cron.expression}"/> 
    </bean> 

    <!-- (2) Job --> 
    <bean name="ReportsJob" class="org.springframework.scheduling.quartz.JobDetailFactoryBean"> 
     <property name="jobClass" value="com.reports.jobs.ReportsJob"/> 
     <property name="name" value="ReportsJob" /> 
     <property name="durability" value="true" /> 
    </bean> 

    <!-- (3) Scheduler --> 
    <bean class="org.springframework.scheduling.quartz.SchedulerFactoryBean"> 
     <property name="schedulerContextAsMap"> 
      <map>    
       <entry key="inputEndpoint"><ref bean="sftpInboundAdapter" /></entry> 
       <entry key="inputEndpointTrigger"><ref bean="sftp-trigger" /></entry> 
      </map> 
     </property> 
     <property name="dataSource" ref="dataSource" /> 
     <property name="overwriteExistingJobs" value="true" />  
     <property name="quartzProperties"> 
      <props> 
       <prop key="org.quartz.scheduler.instanceName">ReportsBatchScheduler</prop> 
       <prop key="org.quartz.scheduler.instanceId">AUTO</prop> 
       <prop key="org.quartz.jobStore.misfireThreshold">60000</prop> 
       <prop key="org.quartz.jobStore.class">org.quartz.impl.jdbcjobstore.JobStoreTX</prop> 
       <prop key="org.quartz.jobStore.driverDelegateClass">org.quartz.impl.jdbcjobstore.StdJDBCDelegate</prop> 
       <prop key="org.quartz.jobStore.tablePrefix">dbo.QRTZ_</prop> 
       <prop key="org.quartz.jobStore.isClustered">true</prop> 
       <prop key="org.quartz.threadPool.class">org.quartz.simpl.SimpleThreadPool</prop> 
       <prop key="org.quartz.threadPool.threadCount">1</prop> 
       <prop key="org.quartz.threadPool.threadPriority">5</prop> 
       <prop key="org.quartz.jobStore.selectWithLockSQL">SELECT * FROM {0}LOCKS UPDLOCK WHERE LOCK_NAME = ?</prop> 
      </props> 
     </property> 
     <property name="triggers"> 
      <list> 
       <ref bean="ReportCronTrigger" /> 
      </list> 
     </property> 
    </bean> 

    <bean id="dataSource" class="org.springframework.jndi.JndiObjectFactoryBean"> 
     <property name="jndiName"><value>java:/MSSQLDS_APP</value></property> 
    </bean> 

</beans> 

사용자 정의 필터

public class ReportsFileListFilter extends AbstractFileListFilter<LsEntry> {  
    SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd"); 
    @Override 
    protected boolean accept(LsEntry file) { 
     if (".".equals(file.getFilename()) || "..".equals(file.getFilename())) { 
      return false; 
     } 
     String dt = sdf.format(new Date());   
     if (file != null && file.getFilename().contains(dt)) { 
      Log.logDebug(ReportsFileListFilter.class, "Downloading File :: "+file.getFilename());    
      return true; 
     } 
     return false;  
    } 
} 

트리거

public class ReportsTrigger implements Trigger { 
    boolean done; 

    public Date nextExecutionTime(TriggerContext triggerContext) {   
     if (done) { 
      return null; 
     } 
     done = true; 
     Log.logDebug(ReportsTrigger.class, "Job started for date :: "+new Date());   
     return new Date();   
    } 

    public void reset() { 
     Log.logDebug(ReportsTrigger.class, "Reset is called"); 
     done = false; 
    } 
} 

석영 작업은

public class ReportsJob extends QuartzJobBean { 

    private AbstractEndpoint inputEndpoint; 
    private ReportsTrigger inputEndpointTrigger;  

    public void setInputEndpoint(final AbstractEndpoint pInputEndpoint) { 
     this.inputEndpoint = pInputEndpoint; 
    } 
    public void setInputEndpointTrigger(final ReportsTrigger pInputEndpointTrigger) { 
     this.inputEndpointTrigger = pInputEndpointTrigger; 
    } 
    @Override 
    protected void executeInternal(final JobExecutionContext pParamJobExecutionContext) 
    throws JobExecutionException {  
     inputEndpoint.stop(); 
     inputEndpointTrigger.reset(); 
     inputEndpoint.start(); 
     Log.logDebug(ReportsJob.class, "Job runnnnnnning"); 
    } 
} 

Answer 1

는 사실은 내가 다운로드 한 파일을 구문 분석하고 데이터베이스 테이블

음에 삽입하려고합니다. FTP 어댑터는 원격 디렉토리를 폴링하여 해당 파일을 국지 디렉토리에 저장하고 하나씩 Message.payload (File 오브젝트)로 전송하므로 다운 스트림 플로우에서 각 사용자 정의 프로세스를 수행 할 중지가 없습니다.

예. <int-file:file-to-string-transformer>을 사용하고 추가 사용자 정의 파서를 <transformer ref="">으로 제공 할 수 있습니다.

결국 <int-jdbc:outbound-channel-adapter>을 처리하여 payload의 데이터를 DB에 저장하는 것만으로 충분합니다.