从远程读取大文件的Mule流程设计

Question

我有一个流程，包括以下步骤：

1)从源SFTP服务器中选择一个文件

2)复制到本地存储

3) 进程文件，使用本地存储中的副本

4)将处理过的文件(将被转换)放置到目标SFTP服务器中

5)将源SFTP中的文件移动到源SFTP服务器上的另一个文件夹中(我找不到这样做的方法，因此我将从临时位置复制回SFTP处理文件夹)

这似乎是一个标准的工作流，但是我没有找到任何关于如何在Mule中具体实现这个工作流程的建议。

​

​

我目前的实现如下：

<file:connector name="tempFile" workDirectory="${temp.file.location}/work"
    workFileNamePattern="#[message.inboundProperties.originalFilename]"
    autoDelete="true" streaming="false" validateConnections="true"
    doc:name="File" />

<sftp:connector name="InputSFTP" validateConnections="true" keepFileOnError="true" doc:name="SFTP" >
    <reconnect frequency="${reconnectfrequency}" count="5"/>
</sftp:connector>

<sftp:connector name="DestinationSFTP" validateConnections="true" pollingFrequency="30000" doc:name="SFTP">
    <reconnect frequency="${reconnectfrequency}" count="5"/>
</sftp:connector>
<smtp:gmail-connector name="Gmail" contentType="text/plain" validateConnections="true" doc:name="Gmail"/>


<flow name="DownloadFTPFileIntoLocalFlow" processingStrategy="synchronous" tracking:enable-default-events="true">
    <sftp:inbound-endpoint connector-ref="InputSFTP" host="${source.host}" port="22" path="${source.path}" user="${source.username}" 
    password="${source.password}" responseTimeout="90000" pollingFrequency="120000" sizeCheckWaitTime="1000" doc:name="InputSFTP" autoDelete="true">
        <file:filename-regex-filter pattern="[Z].*\.csv" caseSensitive="false" />
    </sftp:inbound-endpoint>
    <file:outbound-endpoint path="${temp.file.location}" responseTimeout="10000" doc:name="Templocation" outputPattern="#[message.inboundProperties.originalFilename]" connector-ref="tempFile" />
    <exception-strategy ref="Default_Exception_Strategy" doc:name="Reference Exception Strategy"/>
</flow>
<flow name="ProcessCSVFlow" processingStrategy="synchronous" tracking:enable-default-events="true">
    <file:inbound-endpoint path="${temp.file.location}" connector-ref="tempFile" pollingFrequency="180000" fileAge="10000" responseTimeout="10000" doc:name="TempFileLocation"/>
    <transformer ref="enrichWithHeaderAndEndOfFileTransformer" doc:name="headerAndEOFEnricher" />
    <set-variable variableName="outputfilename" value="#['Mercury'+server.dateTime.year+server.dateTime.month+server.dateTime.dayOfMonth+server.dateTime.hours +server.dateTime.minutes+server.dateTime.seconds+'.csv']" doc:name="outputfilename"/>
    <sftp:outbound-endpoint exchange-pattern="one-way" connector-ref="DestinationSFTP" host="${destination.host}" port="22" responseTimeout="10000" doc:name="DestinationSFTP" 
    outputPattern="#[outputfilename]" path="${destination.path}" user="${destination.username}" password="${destination.password}"/>
    <gzip-compress-transformer/>
    <sftp:outbound-endpoint exchange-pattern="one-way" connector-ref="InputSFTP" host="${source.host}" port="22" responseTimeout="10000" doc:name="SourceArchiveSFTP" 
    outputPattern="#[outputfilename].gzip" path="Archive" user="${source.username}" password="${source.password}"/>
    <set-payload value="Hello world" doc:name="Set Payload"/>
    <smtp:outbound-endpoint host="${smtp.host}" port="${smtp.port}" user="${smtp.from.address}" password="${smtp.from.password}" 
                            to="${smtp.to.address}" from="${smtp.from.address}" subject="${mail.success.subject}" responseTimeout="10000" 
                            doc:name="SuccessEmail" connector-ref="Gmail"/>
    <logger message="Process completed successfully" level="INFO" doc:name="Logger"/>
    <tracking:transaction id="#[server.dateTime]"/>
    <exception-strategy ref="Default_Exception_Strategy" doc:name="Reference Exception Strategy"/>

</flow>
<catch-exception-strategy name="Default_Exception_Strategy">
    <logger message="Exception has occured Payload is #[payload] and Message is #[message]" level="ERROR" doc:name="Logger"/>
    <!-- <smtp:outbound-endpoint host="localhost" responseTimeout="10000" doc:name="Failure Email"/> -->
</catch-exception-strategy>

Answer 1

您试过在SFTP连接器上启用autoDelete="true“强制删除吗？

另外，是否不可能执行flow1: SFTP-in ->转换->文件输出，flow2: file-in -> SFTP- out？

HTH