SFTP 适配器
SFTP 适配器
Spring Integration 支持通过 SFTP 进行文件传输作。
安全文件传输协议 (SFTP) 是一种网络协议,可让您通过任何可靠的流在 Internet 上的两台计算机之间传输文件。
SFTP 协议需要安全通道(例如 SSH),并在整个 SFTP 会话中查看客户端的身份。
Spring Integration 通过提供三个客户端端点来支持通过 SFTP 发送和接收文件:入站通道适配器、出站通道适配器和出站网关。 它还提供了方便的命名空间配置来定义这些客户端组件。
从 6.0 版开始,过时的 JCraft JSch 客户端已被现代 Apache MINA SSHD 框架取代。
这导致框架组件发生了许多重大变化。
但是,在大多数情况下,这种迁移隐藏在 Spring Integration API 后面。
最剧烈的变化发生在DefaultSftpSessionFactory 它现在基于org.apache.sshd.client.SshClient 并公开一些如果其配置属性。 |
您需要将此依赖项包含在您的项目中:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-sftp</artifactId>
<version>6.0.9</version>
</dependency>
compile "org.springframework.integration:spring-integration-sftp:6.0.9"
要在 xml 配置中包含 SFTP 命名空间,请在根元素上包含以下属性:
xmlns:int-sftp="http://www.springframework.org/schema/integration/sftp"
xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp
https://www.springframework.org/schema/integration/sftp/spring-integration-sftp.xsd"
SFTP 会话工厂
从 3.0 版开始,默认情况下不再缓存会话。 请参阅 SFTP 会话缓存。 |
在配置 SFTP 适配器之前,必须配置 SFTP 会话工厂。 您可以使用常规 bean 定义配置 SFTP 会话工厂,如以下示例所示:
<beans:bean id="sftpSessionFactory"
class="org.springframework.integration.sftp.session.DefaultSftpSessionFactory">
<beans:property name="host" value="localhost"/>
<beans:property name="privateKey" value="classpath:META-INF/keys/sftpTest"/>
<beans:property name="privateKeyPassphrase" value="springIntegration"/>
<beans:property name="port" value="22"/>
<beans:property name="user" value="kermit"/>
</beans:bean>
每次适配器从其SessionFactory
,将创建新的 SFTP 会话。
在幕后,SFTP 会话工厂依赖于 Apache MINA SSHD 库来提供 SFTP 功能。
但是,Spring Integration 还支持缓存 SFTP 会话。 有关详细信息,请参阅 SFTP 会话缓存。
这 使用此功能时,必须将会话工厂包装在缓存会话工厂中,如下所述,以便在作完成时不会物理关闭连接。 如果重置缓存,则仅当最后一个通道关闭时会话才会断开连接。 如果在新作获取会话时发现连接已断开连接,则会刷新连接。 |
现在,您需要做的就是将此 SFTP 会话工厂注入您的适配器。
为 SFTP 会话工厂提供值的更实用的方法是使用 Spring 的属性占位符支持。 |
配置属性
以下列表描述了DefaultSftpSessionFactory
.
isSharedSession
(构造函数参数)::当true
,单个SftpClient
用于所有请求的SftpSession
实例。
它默认为false
.
sftpVersionSelector
::一SftpVersionSelector
用于 SFTP 协议选择的实例。
默认值为SftpVersionSelector.CURRENT
.
host
::要连接的主机的 URL。
必填。
hostConfig
::一org.apache.sshd.client.config.hosts.HostConfigEntry
实例作为用户/主机/端口选项的替代方法。
可以使用代理跳转属性进行配置。
port
::建立SFTP连接的端口。
如果未指定,则此值默认为22
.
如果指定,则此属性必须为正数。
user
::要使用的远程用户。
必填。
knownHostsResource
::一org.springframework.core.io.Resource
用于主机密钥存储库的。
资源的内容必须与 OpenSSH 格式相同known_hosts
文件,并且是必需的,并且必须预填充allowUnknownKeys
是错误的。
password
::用于对远程主机进行身份验证的密码。
如果未提供密码,则privateKey
属性是必需的。
privateKey
::一org.springframework.core.io.Resource
表示用于对远程主机进行身份验证的私钥的位置。
如果privateKey
未提供,则password
属性是必需的。
privateKeyPassphrase
::私钥的密码。
如果您将userInfo
,privateKeyPassphrase
是不允许的。
密码是从该对象获取的。
自选。
timeout
::timeout 属性用作套接字超时参数,以及默认连接超时。
默认为0
,这意味着不会发生超时。
allowUnknownKeys
::设置为true
以允许连接到具有未知(或已更改)密钥的主机。
它的默认值是 'false'。
如果false
,预填充的knownHosts
file 是必填项。
userInteraction
::定制org.apache.sshd.client.auth.keyboard.UserInteraction
在身份验证期间使用。
委托会话工厂
4.2 版引入了DelegatingSessionFactory
,这允许在运行时选择实际的会话工厂。
在调用 SFTP 端点之前,您可以调用setThreadKey()
在工厂上将密钥与当前线程相关联。然后,该密钥用于查找要使用的实际会话工厂。您可以通过调用clearThreadKey()
使用后。
我们添加了便利方法,以便您可以更轻松地从消息流中执行此作,如以下示例所示:
<bean id="dsf" class="org.springframework.integration.file.remote.session.DelegatingSessionFactory">
<constructor-arg>
<bean class="o.s.i.file.remote.session.DefaultSessionFactoryLocator">
<!-- delegate factories here -->
</bean>
</constructor-arg>
</bean>
<int:service-activator input-channel="in" output-channel="c1"
expression="@dsf.setThreadKey(#root, headers['factoryToUse'])" />
<int-sftp:outbound-gateway request-channel="c1" reply-channel="c2" ... />
<int:service-activator input-channel="c2" output-channel="out"
expression="@dsf.clearThreadKey(#root)" />
使用会话缓存时(请参阅 SFTP 会话缓存),应缓存每个委托。
您无法缓存DelegatingSessionFactory 本身。 |
从 5.0.7 版本开始,DelegatingSessionFactory
可以与RotatingServerAdvice
轮询多个服务器;请参阅入站通道适配器:轮询多个服务器和目录。
SFTP 会话缓存
从 Spring Integration 版本 3.0 开始,默认情况下不再缓存会话。
这cache-sessions 端点不再支持属性。
如果要缓存会话,则必须使用CachingSessionFactory (请参阅下一个示例)。 |
在 3.0 之前的版本中,默认情况下会自动缓存会话。
一个cache-sessions
属性可用于禁用自动缓存,但该解决方案没有提供配置其他会话缓存属性的方法。
例如,您无法限制创建的会话数。
为了支持该要求和其他配置选项,我们添加了一个CachingSessionFactory
.
它提供了sessionCacheSize
和sessionWaitTimeout
性能。
顾名思义,sessionCacheSize
属性控制工厂在其缓存中维护的活动会话数(默认值为无限制)。
如果sessionCacheSize
已达到阈值,则任何获取另一个会话的尝试都会阻塞,直到其中一个缓存的会话变为可用或会话的等待时间到期(默认等待时间为Integer.MAX_VALUE
).
这sessionWaitTimeout
属性启用等待时间的配置。
如果希望缓存会话,请配置默认会话工厂(如前所述),然后将其包装在CachingSessionFactory
您可以在其中提供这些附加属性。
以下示例显示了如何执行此作:
<bean id="sftpSessionFactory"
class="org.springframework.integration.sftp.session.DefaultSftpSessionFactory">
<property name="host" value="localhost"/>
</bean>
<bean id="cachingSessionFactory"
class="org.springframework.integration.file.remote.session.CachingSessionFactory">
<constructor-arg ref="sftpSessionFactory"/>
<constructor-arg value="10"/>
<property name="sessionWaitTimeout" value="1000"/>
</bean>
前面的示例创建了一个CachingSessionFactory
与其sessionCacheSize
设置为10
及其sessionWaitTimeout
设置为一秒(1000毫秒)。
从 Spring Integration 3.0 版开始,CachingConnectionFactory
提供一个resetCache()
方法。
调用时,所有空闲会话都会立即关闭,正在使用的会话在返回到缓存时会关闭。
使用时isSharedSession=true
,通道关闭,共享会话仅在最后一个通道关闭时关闭。
新的会话请求会根据需要建立新的会话。
从 5.1 版本开始,CachingSessionFactory
有一个新属性testSession
.
如果为 true,则将通过执行REALPATH
命令为空路径以确保它仍然处于活动状态;如果没有,它将从缓存中删除;如果缓存中没有活动会话,则会创建新会话。
用RemoteFileTemplate
Spring Integration 3.0 版提供了一个新的抽象,对SftpSession
对象。
该模板提供了发送、检索的方法(作为InputStream
)、删除和重命名文件。
此外,我们还提供execute
方法,让调用方在会话上运行多个作。
在所有情况下,模板都会负责可靠地关闭会话。
有关更多信息,请参阅Javadoc 的RemoteFileTemplate
SFTP 有一个子类:SftpRemoteFileTemplate
.
我们在 4.1 版中添加了其他方法,包括getClientInstance()
.
它提供对底层ChannelSftp
,这允许访问低级 API。
5.0 版本引入了RemoteFileOperations.invoke(OperationsCallback<F, T> action)
方法。
此方法允许多个RemoteFileOperations
调用在同一线程绑定的范围内调用Session
.
当您需要执行RemoteFileTemplate
作为一个工作单元。
例如AbstractRemoteFileOutboundGateway
将其与mput
命令实现,其中我们执行put
对所提供目录中的每个文件进行作,并递归地对其子目录进行作。
有关更多信息,请参阅 Javadoc。
SFTP 入站通道适配器
SFTP 入站通道适配器是一个特殊的侦听器,它连接到服务器并侦听远程目录事件(例如正在创建的新文件),此时它会启动文件传输。 以下示例演示如何配置 SFTP 入站通道适配器:
<int-sftp:inbound-channel-adapter id="sftpAdapterAutoCreate"
session-factory="sftpSessionFactory"
channel="requestChannel"
filename-pattern="*.txt"
remote-directory="/foo/bar"
preserve-timestamp="true"
local-directory="file:target/foo"
auto-create-local-directory="true"
local-filename-generator-expression="#this.toUpperCase() + '.a'"
scanner="myDirScanner"
local-filter="myFilter"
temporary-file-suffix=".writing"
max-fetch-size="-1"
delete-remote-files="false">
<int:poller fixed-rate="1000"/>
</int-sftp:inbound-channel-adapter>
前面的配置示例演示了如何为各种属性提供值,包括以下内容:
-
local-directory
:文件将传输到的位置 -
remote-directory
:要从中传输文件的远程源目录 -
session-factory
:对我们之前配置的 bean 的引用
默认情况下,传输的文件与原始文件具有相同的名称。
如果要覆盖此行为,可以将local-filename-generator-expression
属性,它允许您提供 SpEL 表达式来生成本地文件的名称。
与出站网关和适配器不同,其中 SpEL 评估上下文的根对象是Message
,则此入站适配器在评估时尚未收到该消息,因为这是它最终使用传输的文件作为其有效负载生成的内容。
因此,SpEL 求值上下文的根对象是远程文件的原始名称 (String
).
入站通道适配器首先将文件检索到本地目录,然后根据轮询器配置发出每个文件。
从 5.0 版开始,当需要检索新文件时,您可以限制从 SFTP 服务器获取的文件数。
当目标文件很大或在具有持久文件列表过滤器的群集系统中运行时,这非常有用,本节稍后将讨论。
用max-fetch-size
为此目的。
负值(默认值)表示没有限制,并且会检索所有匹配的文件。
有关更多信息,请参阅入站通道适配器:控制远程文件获取。
从 5.0 版本开始,您还可以提供自定义DirectoryScanner
实现到inbound-channel-adapter
通过将scanner
属性。
从 Spring Integration 3.0 开始,您可以指定preserve-timestamp
属性(默认值为false
).
什么时候true
,则本地文件的修改时间戳设置为从服务器检索的值。
否则,它设置为当前时间。
从 4.2 版开始,您可以指定remote-directory-expression
而不是remote-directory
,它允许您动态确定每次轮询的目录 - 例如remote-directory-expression="@myBean.determineRemoteDir()"
.
有时,基于通过filename-pattern
属性可能不够。
如果是这种情况,您可以使用filename-regex
属性来指定正则表达式(例如,filename-regex=".*\.test$"
).
如果您需要完全控制,可以使用filter
属性来提供对org.springframework.integration.file.filters.FileListFilter
,这是一个用于过滤文件列表的策略接口。
此过滤器确定要检索哪些远程文件。
您还可以将基于模式的过滤器与其他过滤器(例如AcceptOnceFileListFilter
,以避免同步之前获取的文件),方法是使用CompositeFileListFilter
.
这AcceptOnceFileListFilter
将其状态存储在内存中。
如果您希望状态在系统重新启动后继续存在,请考虑使用SftpPersistentAcceptOnceFileListFilter
相反。
此过滤器将接受的文件名存储在MetadataStore
策略(请参阅元数据存储)。
此过滤器与文件名和远程修改时间相匹配。
从 4.0 版开始,此过滤器需要ConcurrentMetadataStore
.
与共享数据存储(例如Redis
使用RedisMetadataStore
),这允许在多个应用程序或服务器实例之间共享过滤器键。
从 5.0 版开始,SftpPersistentAcceptOnceFileListFilter
使用内存中的SimpleMetadataStore
默认情况下应用于SftpInboundFileSynchronizer
.
此过滤器也与regex
或pattern
选项,以及通过SftpInboundChannelAdapterSpec
在 Java DSL 中。
您可以使用以下命令来处理任何其他用例CompositeFileListFilter
(或ChainFileListFilter
).
上面的讨论是指在检索文件之前对其进行过滤。 检索文件后,将对文件系统上的文件应用额外的过滤器。 默认情况下,这是一个“AcceptOnceFileListFilter”,如本节所述,它将状态保留在内存中,并且不考虑文件的修改时间。 除非应用程序在处理后删除文件,否则适配器会在应用程序重新启动后默认重新处理磁盘上的文件。
此外,如果您配置filter
使用SftpPersistentAcceptOnceFileListFilter
并且远程文件时间戳更改(导致它被重新获取),默认本地过滤器不允许处理这个新文件。
有关此筛选器及其使用方式的详细信息,请参阅远程持久文件列表筛选器。
您可以使用local-filter
属性来配置本地文件系统过滤器的行为。从 V4.3.8 开始,FileSystemPersistentAcceptOnceFileListFilter
默认配置。此过滤器将接受的文件名和修改的时间戳存储在MetadataStore
策略(请参阅元数据存储)并检测对本地文件修改时间的更改。默认MetadataStore
是一个SimpleMetadataStore
将状态存储在内存中。
从 4.1.5 版开始,这些过滤器有一个名为flushOnUpdate
,这会导致它们在每次更新时刷新metadata store(如果 store 实现Flushable
).
此外,如果您使用分布式MetadataStore (例如 Redis 元数据存储),您可以拥有同一适配器或应用程序的多个实例,并确保只有一个实例处理一个文件。 |
实际的本地过滤器是CompositeFileListFilter
包含提供的过滤器和模式过滤器,该过滤器可防止处理正在下载的文件(基于temporary-file-suffix
). 下载带有此后缀的文件(默认值为.writing
),传输完成后,文件将重命名为最终名称,使它们对过滤器“可见”。
有关这些属性的更多详细信息,请参阅架构。
SFTP 入站通道适配器是轮询使用者。因此,您必须配置轮询器(全局默认值或本地元素)。将文件传输到本地目录后,带有java.io.File
由于其有效负载类型被生成并发送到channel
属性。
有关文件过滤和大文件的更多信息
有时,刚刚出现在受监视(远程)目录中的文件不完整。通常,此类文件是使用一些临时扩展名(例如.writing
在名为something.txt.writing
),然后在写入过程完成后重命名。
在大多数情况下,开发人员只对完整的文件感兴趣,并且只想过滤这些文件。
若要处理这些方案,可以使用filename-pattern
,filename-regex
和filter
属性。
如果需要自定义筛选器实现,可以通过将filter
属性。
以下示例显示了如何执行此作:
<int-sftp:inbound-channel-adapter id="sftpInbondAdapter"
channel="receiveChannel"
session-factory="sftpSessionFactory"
filter="customFilter"
local-directory="file:/local-test-dir"
remote-directory="/remote-test-dir">
<int:poller fixed-rate="1000" max-messages-per-poll="10" task-executor="executor"/>
</int-sftp:inbound-channel-adapter>
<bean id="customFilter" class="org.foo.CustomFilter"/>
从故障中恢复
您应该了解适配器的体系结构。
文件同步器获取文件,然后FileReadingMessageSource
为每个同步文件发出一条消息。
如前所述,涉及两个过滤器。
这filter
属性(和模式)引用远程 (SFTP) 文件列表,以避免获取已获取的文件。
这FileReadingMessageSource
使用local-filter
以确定哪些文件将作为消息发送。
同步器列出远程文件并查阅其过滤器。
然后传输文件。
如果在文件传输期间发生 IO 错误,则会删除已添加到过滤器中的任何文件,以便有资格在下一次轮询时重新提取它们。
仅当过滤器实现ReversibleFileListFilter
(例如AcceptOnceFileListFilter
).
如果在同步文件后,处理文件的下游流发生错误,则不会发生过滤器的自动回滚,因此默认情况下不会重新处理失败的文件。
如果希望在失败后重新处理此类文件,可以使用类似于以下内容的配置来方便从过滤器中删除失败的文件:
<int-sftp:inbound-channel-adapter id="sftpAdapter"
session-factory="sftpSessionFactory"
channel="requestChannel"
remote-directory-expression="'/sftpSource'"
local-directory="file:myLocalDir"
auto-create-local-directory="true"
filename-pattern="*.txt">
<int:poller fixed-rate="1000">
<int:transactional synchronization-factory="syncFactory" />
</int:poller>
</int-sftp:inbound-channel-adapter>
<bean id="acceptOnceFilter"
class="org.springframework.integration.file.filters.AcceptOnceFileListFilter" />
<int:transaction-synchronization-factory id="syncFactory">
<int:after-rollback expression="payload.delete()" />
</int:transaction-synchronization-factory>
<bean id="transactionManager"
class="org.springframework.integration.transaction.PseudoTransactionManager" />
上述配置适用于任何ResettableFileListFilter
.
从 5.0 版开始,入站通道适配器可以根据生成的本地文件名在本地构建子目录。
这也可以是一个远程子路径。
为了能够递归地读取本地目录以根据层次结构支持进行修改,您现在可以提供一个内部FileReadingMessageSource
使用新的RecursiveDirectoryScanner
基于Files.walk()
算法。
看AbstractInboundFileSynchronizingMessageSource.setScanner()
了解更多信息。
此外,您现在可以将AbstractInboundFileSynchronizingMessageSource
到WatchService
-基于DirectoryScanner
通过使用setUseWatchService()
选择。
它还配置为所有WatchEventType
实例对本地目录中的任何修改做出反应。
前面显示的重新处理示例基于FileReadingMessageSource.WatchServiceDirectoryScanner
,它使用ResettableFileListFilter.remove()
删除文件时 (StandardWatchEventKinds.ENTRY_DELETE
) 从本地目录中。
看WatchServiceDirectoryScanner
了解更多信息。
使用 Java 配置进行配置
以下 Spring Boot 应用程序显示了如何使用 Java 配置入站适配器的示例:
@SpringBootApplication
public class SftpJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(SftpJavaApplication.class)
.web(false)
.run(args);
}
@Bean
public SessionFactory<SftpClient.DirEntry> sftpSessionFactory() {
DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
factory.setHost("localhost");
factory.setPort(port);
factory.setUser("foo");
factory.setPassword("foo");
factory.setAllowUnknownKeys(true);
factory.setTestSession(true);
return new CachingSessionFactory<>(factory);
}
@Bean
public SftpInboundFileSynchronizer sftpInboundFileSynchronizer() {
SftpInboundFileSynchronizer fileSynchronizer = new SftpInboundFileSynchronizer(sftpSessionFactory());
fileSynchronizer.setDeleteRemoteFiles(false);
fileSynchronizer.setRemoteDirectory("foo");
fileSynchronizer.setFilter(new SftpSimplePatternFileListFilter("*.xml"));
return fileSynchronizer;
}
@Bean
@InboundChannelAdapter(channel = "sftpChannel", poller = @Poller(fixedDelay = "5000"))
public MessageSource<File> sftpMessageSource() {
SftpInboundFileSynchronizingMessageSource source =
new SftpInboundFileSynchronizingMessageSource(sftpInboundFileSynchronizer());
source.setLocalDirectory(new File("sftp-inbound"));
source.setAutoCreateLocalDirectory(true);
source.setLocalFilter(new AcceptOnceFileListFilter<File>());
source.setMaxFetchSize(1);
return source;
}
@Bean
@ServiceActivator(inputChannel = "sftpChannel")
public MessageHandler handler() {
return new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
System.out.println(message.getPayload());
}
};
}
}
使用 Java DSL 进行配置
以下 Spring Boot 应用程序显示了如何使用 Java DSL 配置入站适配器的示例:
@SpringBootApplication
public class SftpJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(SftpJavaApplication.class)
.web(false)
.run(args);
}
@Bean
public IntegrationFlow sftpInboundFlow() {
return IntegrationFlow
.from(Sftp.inboundAdapter(this.sftpSessionFactory)
.preserveTimestamp(true)
.remoteDirectory("foo")
.regexFilter(".*\\.txt$")
.localFilenameExpression("#this.toUpperCase() + '.a'")
.localDirectory(new File("sftp-inbound")),
e -> e.id("sftpInboundAdapter")
.autoStartup(true)
.poller(Pollers.fixedDelay(5000)))
.handle(m -> System.out.println(m.getPayload()))
.get();
}
}
SFTP 流式入站通道适配器
版本 4.3 引入了流式入站通道适配器。
此适配器生成有效负载类型为InputStream
,让您无需写入本地文件系统即可获取文件。
由于会话保持打开状态,因此使用应用程序负责在文件使用时关闭会话。
会话在closeableResource
标头 (IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE
).
标准框架组件,例如FileSplitter
和StreamTransformer
,自动关闭会话。
有关这些组件的更多信息,请参阅文件拆分器和流转换器。
以下示例演示如何配置 SFTP 流式入站通道适配器:
<int-sftp:inbound-streaming-channel-adapter id="ftpInbound"
channel="ftpChannel"
session-factory="sessionFactory"
filename-pattern="*.txt"
filename-regex=".*\.txt"
filter="filter"
filter-expression="@myFilterBean.check(#root)"
remote-file-separator="/"
comparator="comparator"
max-fetch-size="1"
remote-directory-expression="'foo/bar'">
<int:poller fixed-rate="1000" />
</int-sftp:inbound-streaming-channel-adapter>
您只能使用其中一种filename-pattern
,filename-regex
,filter
或filter-expression
.
从 5.0 版开始,默认情况下,SftpStreamingMessageSource 适配器通过使用SftpPersistentAcceptOnceFileListFilter 基于内存中的SimpleMetadataStore .
默认情况下,此过滤器也与文件名模式(或正则表达式)一起应用。
如果您需要允许重复,您可以使用AcceptAllFileListFilter .
您可以使用以下命令来处理任何其他用例CompositeFileListFilter (或ChainFileListFilter ).
稍后显示的 Java 配置显示了一种在处理后删除远程文件的技术,避免重复。 |
有关SftpPersistentAcceptOnceFileListFilter
及其使用方式,请参阅远程持久文件列表过滤器。
您可以使用max-fetch-size
属性来限制在需要获取时每次轮询时获取的文件数量。
将其设置为1
并在集群环境中运行时使用持久过滤器。
有关更多信息,请参阅入站通道适配器:控制远程文件获取。
适配器将远程目录和文件名放在标头 (FileHeaders.REMOTE_DIRECTORY
和FileHeaders.REMOTE_FILE
,分别)。
从 5.0 版开始,FileHeaders.REMOTE_FILE_INFO
header 提供额外的远程文件信息(JSON 格式)。
如果您将fileInfoJson
属性SftpStreamingMessageSource
自false
,标头包含一个SftpFileInfo
对象。
您可以访问SftpClient.DirEntry
由底层提供的对象SftpClient
通过使用SftpFileInfo.getFileInfo()
方法。
这fileInfoJson
属性在使用 XML 配置时不可用,但您可以通过注入SftpStreamingMessageSource
到您的配置类之一。
另请参阅远程文件信息。
使用 Java 配置进行配置
以下 Spring Boot 应用程序显示了如何使用 Java 配置入站适配器的示例:
@SpringBootApplication
public class SftpJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(SftpJavaApplication.class)
.web(false)
.run(args);
}
@Bean
@InboundChannelAdapter(channel = "stream")
public MessageSource<InputStream> ftpMessageSource() {
SftpStreamingMessageSource messageSource = new SftpStreamingMessageSource(template());
messageSource.setRemoteDirectory("sftpSource/");
messageSource.setFilter(new AcceptAllFileListFilter<>());
messageSource.setMaxFetchSize(1);
return messageSource;
}
@Bean
@Transformer(inputChannel = "stream", outputChannel = "data")
public org.springframework.integration.transformer.Transformer transformer() {
return new StreamTransformer("UTF-8");
}
@Bean
public SftpRemoteFileTemplate template() {
return new SftpRemoteFileTemplate(sftpSessionFactory());
}
@ServiceActivator(inputChannel = "data", adviceChain = "after")
@Bean
public MessageHandler handle() {
return System.out::println;
}
@Bean
public ExpressionEvaluatingRequestHandlerAdvice after() {
ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
advice.setOnSuccessExpression(
"@template.remove(headers['file_remoteDirectory'] + headers['file_remoteFile'])");
advice.setPropagateEvaluationFailures(true);
return advice;
}
}
请注意,在此示例中,转换器下游的消息处理程序具有advice
在处理后删除远程文件。
入站通道适配器:轮询多个服务器和目录
从 5.0.7 版本开始,RotatingServerAdvice
可用;当配置为轮询器通知时,入站适配器可以轮询多个服务器和目录。
配置通知并将其正常添加到轮询器的通知链中。
一个DelegatingSessionFactory
用于选择服务器,请参阅委托会话工厂了解更多信息。
通知配置由以下列表组成RotationPolicy.KeyDirectory
对象。
@Bean
public RotatingServerAdvice advice() {
List<RotationPolicy.KeyDirectory> keyDirectories = new ArrayList<>();
keyDirectories.add(new RotationPolicy.KeyDirectory("one", "foo"));
keyDirectories.add(new RotationPolicy.KeyDirectory("one", "bar"));
keyDirectories.add(new RotationPolicy.KeyDirectory("two", "baz"));
keyDirectories.add(new RotationPolicy.KeyDirectory("two", "qux"));
keyDirectories.add(new RotationPolicy.KeyDirectory("three", "fiz"));
keyDirectories.add(new RotationPolicy.KeyDirectory("three", "buz"));
return new RotatingServerAdvice(delegatingSf(), keyDirectories);
}
此建议将轮询目录foo
在服务器上one
直到没有新文件存在,然后移动到目录bar
然后是目录baz
在服务器上two
等。
可以使用fair
构造函数 arg:
@Bean
public RotatingServerAdvice advice() {
...
return new RotatingServerAdvice(delegatingSf(), keyDirectories, true);
}
在这种情况下,通知将移动到下一个服务器/目录,而不管上一个轮询是否返回了文件。
或者,您可以提供自己的RotationPolicy
要根据需要重新配置消息源,请执行以下作:
public interface RotationPolicy {
void beforeReceive(MessageSource<?> source);
void afterReceive(boolean messageReceived, MessageSource<?> source);
}
和
@Bean
public RotatingServerAdvice advice() {
return new RotatingServerAdvice(myRotationPolicy());
}
这local-filename-generator-expression
属性 (localFilenameGeneratorExpression
)现在可以包含#remoteDirectory
变量。
这允许从不同目录检索到的文件下载到本地的类似目录:
@Bean
public IntegrationFlow flow() {
return IntegrationFlow.from(Sftp.inboundAdapter(sf())
.filter(new SftpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "rotate"))
.localDirectory(new File(tmpDir))
.localFilenameExpression("#remoteDirectory + T(java.io.File).separator + #root")
.remoteDirectory("."),
e -> e.poller(Pollers.fixedDelay(1).advice(advice())))
.channel(MessageChannels.queue("files"))
.get();
}
不要配置TaskExecutor 使用此建议时对轮询者;有关详细信息,请参阅消息源的条件轮询器。 |
入站通道适配器:控制远程文件获取
配置入站通道适配器时,应考虑两个属性。max-messages-per-poll
与所有轮询器一样,可用于限制每次轮询时发出的消息数(如果准备就绪的消息超过配置值)。max-fetch-size
(从 5.0 版开始)可以限制一次从远程服务器检索的文件数量。
以下方案假定启动状态为空的本地目录:
-
max-messages-per-poll=2
和max-fetch-size=1
:适配器获取一个文件,发出它,获取下一个文件,然后发出它。 然后它会休眠,直到下一次投票。 -
max-messages-per-poll=2
和max-fetch-size=2
):适配器获取两个文件,然后发出每个文件。 -
max-messages-per-poll=2
和max-fetch-size=4
:适配器最多可获取 4 个文件(如果可用)并发出前两个文件(如果至少有两个)。 接下来的两个文件将在下一次轮询时发出。 -
max-messages-per-poll=2
和max-fetch-size
未指定:适配器获取所有远程文件并发出前两个文件(如果至少有两个)。 后续文件在后续轮询(一次两个)时发出。 当所有文件都被消耗掉时,将再次尝试远程获取,以拾取任何新文件。
当您部署应用程序的多个实例时,我们建议将max-fetch-size ,以避免一个实例“抓取”所有文件并使其他实例饥饿。 |
另一种用途max-fetch-size
是当您想要停止获取远程文件但继续处理已获取的文件时。
设置maxFetchSize
属性MessageSource
(以编程方式,通过 JMX 或通过控制总线)有效地阻止适配器获取更多文件,但允许轮询器继续为之前获取的文件发出消息。
如果在属性更改时轮询器处于活动状态,则更改将在下一次轮询时生效。
从 5.1 版开始,同步器可以为Comparator<?>
.
这在限制使用maxFetchSize
.
SFTP 出站通道适配器
SFTP 出站通道适配器是一个特殊的MessageHandler
连接到远程目录,并为作为传入有效负载接收的每个文件启动文件传输Message
.
它还支持文件的多种表示形式,因此您不仅限于File
对象。
与 FTP 出站适配器类似,SFTP 出站通道适配器支持以下有效负载:
-
java.io.File
:实际文件对象 -
byte[]
:表示文件内容的字节数组 -
java.lang.String
:表示文件内容的文本 -
java.io.InputStream
:要传输到远程文件的数据流 -
org.springframework.core.io.Resource
:用于将数据传输到远程文件的资源
以下示例演示如何配置 SFTP 出站通道适配器:
<int-sftp:outbound-channel-adapter id="sftpOutboundAdapter"
session-factory="sftpSessionFactory"
channel="inputChannel"
charset="UTF-8"
remote-file-separator="/"
remote-directory="foo/bar"
remote-filename-generator-expression="payload.getName() + '-mysuffix'"
filename-generator="fileNameGenerator"
use-temporary-filename="true"
chmod="600"
mode="REPLACE"/>
有关这些属性的更多详细信息,请参阅架构。
SpEL 和 SFTP 出站适配器
与 Spring Integration 中的许多其他组件一样,您可以通过指定两个属性来配置 SFTP 出站通道适配器时使用 Spring 表达式语言 (SpEL):remote-directory-expression
和remote-filename-generator-expression
(前面描述)。
表达式求值上下文将消息作为其根对象,它允许您使用可以根据消息中的数据(来自“有效负载”或“标头”)动态计算文件名或现有目录路径的表达式。
在前面的示例中,我们定义了remote-filename-generator-expression
属性,其表达式值根据文件名的原始名称计算文件名,同时附加后缀:“-mysuffix”。
从 4.1 版开始,您可以指定mode
传输文件时。
默认情况下,将覆盖现有文件。
模式由FileExistsMode
枚举,其中包括以下值:
-
REPLACE
(默认) -
REPLACE_IF_MODIFIED
-
APPEND
-
APPEND_NO_FLUSH
-
IGNORE
-
FAIL
跟IGNORE
和FAIL
,则不会传输文件。FAIL
导致抛出异常,而IGNORE
静默忽略传输(尽管DEBUG
生成日志条目)。
4.3 版引入了chmod
属性,可用于在上传后更改远程文件权限。
您可以使用传统的 Unix 八进制格式(例如600
仅允许文件所有者读写)。
使用 java 配置适配器时,您可以使用setChmodOctal("600")
或setChmod(0600)
.
避免部分写入的文件
处理文件传输时的常见问题之一是处理部分文件的可能性。 文件可能会在实际传输完成之前出现在文件系统中。
为了解决这个问题,Spring Integration SFTP 适配器使用一种通用算法,其中文件以临时名称传输,并在完全传输后重命名。
默认情况下,正在传输的每个文件都显示在文件系统中,并带有一个附加后缀,默认情况下,该后缀为.writing
.
您可以通过设置temporary-file-suffix
属性。
但是,在某些情况下,您可能不想使用此技术(例如,如果服务器不允许重命名文件)。
对于这种情况,您可以通过设置use-temporary-file-name
自false
(默认值为true
).
当此属性为false
,则该文件是用其最终名称写入的,并且使用应用程序需要一些其他机制来检测文件是否已完全上传,然后才能访问它。
使用 Java 配置进行配置
以下 Spring Boot 应用程序显示了如何使用 Java 配置出站适配器的示例:
@SpringBootApplication
@IntegrationComponentScan
public class SftpJavaApplication {
public static void main(String[] args) {
ConfigurableApplicationContext context =
new SpringApplicationBuilder(SftpJavaApplication.class)
.web(false)
.run(args);
MyGateway gateway = context.getBean(MyGateway.class);
gateway.sendToSftp(new File("/foo/bar.txt"));
}
@Bean
public SessionFactory<SftpClient.DirEntry> sftpSessionFactory() {
DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
factory.setHost("localhost");
factory.setPort(port);
factory.setUser("foo");
factory.setPassword("foo");
factory.setAllowUnknownKeys(true);
factory.setTestSession(true);
return new CachingSessionFactory<SftpClient.DirEntry>(factory);
}
@Bean
@ServiceActivator(inputChannel = "toSftpChannel")
public MessageHandler handler() {
SftpMessageHandler handler = new SftpMessageHandler(sftpSessionFactory());
handler.setRemoteDirectoryExpressionString("headers['remote-target-dir']");
handler.setFileNameGenerator(new FileNameGenerator() {
@Override
public String generateFileName(Message<?> message) {
return "handlerContent.test";
}
});
return handler;
}
@MessagingGateway
public interface MyGateway {
@Gateway(requestChannel = "toSftpChannel")
void sendToSftp(File file);
}
}
使用 Java DSL 进行配置
以下 Spring Boot 应用程序显示了如何使用 Java DSL 配置出站适配器的示例:
@SpringBootApplication
public class SftpJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(SftpJavaApplication.class)
.web(false)
.run(args);
}
@Bean
public IntegrationFlow sftpOutboundFlow() {
return IntegrationFlow.from("toSftpChannel")
.handle(Sftp.outboundAdapter(this.sftpSessionFactory, FileExistsMode.FAIL)
.useTemporaryFileName(false)
.remoteDirectory("/foo")
).get();
}
}
SFTP 出站网关
SFTP 出站网关提供了一组有限的命令,可让您与远程 SFTP 服务器进行交互:
-
ls
(列出文件) -
nlst
(列出文件名) -
get
(检索文件) -
mget
(检索多个文件) -
rm
(删除文件) -
mv
(移动和重命名文件) -
put
(发送文件) -
mput
(发送多个文件)
使用ls
命令
ls
列出远程文件并支持以下选项:
-
-1
:检索文件名列表。默认值是检索FileInfo
对象 -
-a
:包括所有文件(包括以“.”开头的文件) -
-f
:不对列表进行排序 -
-dirs
:包括目录(默认排除) -
-links
:包括符号链接(默认排除) -
-R
:递归列出远程目录
此外,文件名过滤的提供方式与inbound-channel-adapter
.
由ls
作是文件名列表或FileInfo
对象(取决于您是否使用-1
开关)。
这些对象提供修改时间、权限等信息。
远程目录ls
命令在file_remoteDirectory
页眉。
使用递归选项 (-R
)、fileName
包含任何子目录元素,并表示文件的相对路径(相对于远程目录)。
如果您使用-dirs
选项,则每个递归目录也将作为列表中的元素返回。
在这种情况下,我们建议您不要使用-1
选项,因为您将无法区分文件与目录,而您可以在使用FileInfo
对象。
如果列表的远程路径以符号开头,则 SFTP 将其视为绝对路径;without - 作为当前用户主页中的相对路径。/
用nlst
命令
版本 5 引入了对nlst
命令。
nlst
列出远程文件名,仅支持一个选项:
-
-f
:不对列表进行排序
由nlst
作是文件名的列表。
这file_remoteDirectory
标头保存远程目录,其上nlst
指挥部采取行动。
SFTP 协议不提供列出名称的功能。
此命令等效于ls
命令将-1
选项,并为方便起见在此处添加。
使用get
命令
get
检索远程文件并支持以下选项:
-
-P
:保留远程文件的时间戳。 -
-stream
:将远程文件检索为流。 -
-D
:传输成功后删除远程文件。 如果忽略传输,则不会删除远程文件,因为FileExistsMode
是IGNORE
并且本地文件已经存在。
这file_remoteDirectory
标头保存远程目录,而file_remoteFile
header 保存文件名。
由get
作是一个File
表示检索到的文件的对象。
如果您使用-stream
选项,则有效负载是InputStream
而不是File
.
对于文本文件,一个常见的用例是将此作与文件拆分器或流转换器相结合。
当将远程文件作为流使用时,您负责关闭Session
在流被消耗后。为方便起见,该Session
在closeableResource
header 和IntegrationMessageHeaderAccessor
提供便捷方法:
Closeable closeable = new IntegrationMessageHeaderAccessor(message).getCloseableResource();
if (closeable != null) {
closeable.close();
}
以下示例演示如何将文件作为流使用:
<int-sftp:outbound-gateway session-factory="ftpSessionFactory"
request-channel="inboundGetStream"
command="get"
command-options="-stream"
expression="payload"
remote-directory="ftpTarget"
reply-channel="stream" />
<int-file:splitter input-channel="stream" output-channel="lines" />
如果您在自定义组件中使用输入流,则必须关闭Session .
您可以在自定义代码中执行此作,也可以将消息的副本路由到service-activator 并使用 SpEL,如以下示例所示: |
<int:service-activator input-channel="closeSession"
expression="headers['closeableResource'].close()" />
使用mget
命令
mget
根据模式检索多个远程文件,并支持以下选项:
-
-P
:保留远程文件的时间戳。 -
-R
:递归检索整个目录树。 -
-x
:如果没有与模式匹配的文件,则抛出异常(否则,返回空列表)。 -
-D
:传输成功后删除每个远程文件。 如果忽略传输,则不会删除远程文件,因为FileExistsMode
是IGNORE
并且本地文件已经存在。
由mget
作是一个List<File>
对象(即List
之File
对象,每个对象代表一个检索到的文件)。
从 5.0 版开始,如果FileExistsMode 是IGNORE ,则输出消息的有效负载不再包含由于文件已存在而未获取的文件。
以前,数组包含所有文件,包括那些已经存在的文件。 |
您使用的表达式确定远程路径应生成结果,例如*
myfiles/*
在myfiles
.
从 5.0 版开始,您可以使用递归MGET
,结合FileExistsMode.REPLACE_IF_MODIFIED
模式,以定期在本地同步整个远程目录树。
此模式将本地文件的上次修改时间戳设置为远程文件的时间戳,无论-P
(保留时间戳)选项。
使用递归(
-R )该模式将被忽略并被假定。
默认情况下,检索整个远程树。
但是,您可以通过提供 如果过滤子目录,则不会对该子目录执行其他遍历。 这 通常,您可以使用 |
持久文件列表过滤器现在具有布尔属性forRecursion
.
将此属性设置为true
,也设置alwaysAcceptDirectories
,这意味着出站网关上的递归作 (ls
和mget
) 现在每次都会遍历完整的目录树。
这是为了解决未检测到目录树深处更改的问题。
另外forRecursion=true
导致文件的完整路径用作元数据存储键;这解决了如果具有相同名称的文件多次出现在不同目录中,则过滤器无法正常工作的问题。
重要提示:这意味着对于顶级目录下的文件,将找不到持久元数据存储中的现有密钥。
因此,该属性是false
默认情况下;这可能会在将来的版本中更改。
从 5.0 版开始,您可以配置SftpSimplePatternFileListFilter
和SftpRegexPatternFileListFilter
通过设置alwaysAcceptDirectorties
自true
.
这样做允许简单模式的递归,如以下示例所示:
<bean id="starDotTxtFilter"
class="org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter">
<constructor-arg value="*.txt" />
<property name="alwaysAcceptDirectories" value="true" />
</bean>
<bean id="dotStarDotTxtFilter"
class="org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter">
<constructor-arg value="^.*\.txt$" />
<property name="alwaysAcceptDirectories" value="true" />
</bean>
您可以使用filter
网关上的属性。
也可以看看出站网关部分成功 (mget
和mput
).
使用put
命令
put
将文件发送到远程服务器。
消息的有效负载可以是java.io.File
一个byte[]
或String
.
一个remote-filename-generator
(或表达式)用于命名远程文件。
其他可用属性包括remote-directory
,temporary-remote-directory
和他们的*-expression
等价物:use-temporary-file-name
和auto-create-directory
.
有关更多信息,请参阅架构文档。
由put
作是一个String
其中包含传输后服务器上文件的完整路径。
4.3 版引入了chmod
属性,该属性在上传后更改远程文件权限。
您可以使用传统的 Unix 八进制格式(例如600
仅允许文件所有者读写)。
使用 java 配置适配器时,您可以使用setChmod(0600)
.
使用mput
命令
mput
向服务器发送多个文件,并支持以下选项:
-
-R
: 递归 — 发送目录和子目录中的所有文件(可能经过过滤)
消息有效负载必须是java.io.File
(或String
) 表示本地目录。
从 5.1 版本开始,集合File
或String
也受支持。
与put
命令被支持。
此外,您可以使用以下选项之一过滤本地目录中的文件mput-pattern
,mput-regex
,mput-filter
或mput-filter-expression
.
过滤器与递归一起工作,只要子目录本身通过过滤器即可。
未通过过滤器的子目录不会递归。
由mput
作是一个List<String>
对象(即List
传输产生的远程文件路径)。
也可以看看出站网关部分成功 (mget
和mput
).
4.3 版引入了chmod
属性,允许您在上传后更改远程文件权限。
您可以使用传统的 Unix 八进制格式(例如600
仅允许文件所有者读写)。
使用 Java 配置适配器时,您可以使用setChmodOctal("600")
或setChmod(0600)
.
使用rm
命令
这rm
命令没有选项。
如果删除作成功,则生成的消息有效负载为Boolean.TRUE
.
否则,消息有效负载为Boolean.FALSE
.
这file_remoteDirectory
标头保存远程目录,而file_remoteFile
header 保存文件名。
使用mv
命令
这mv
命令没有选项。
这expression
属性定义了“from”路径,而rename-expression
属性定义“to”路径。
默认情况下,rename-expression
是headers['file_renameTo']
.
此表达式的计算结果不得为 null 或空String
.
如有必要,将创建所需的任何远程目录。
结果消息的有效负载为Boolean.TRUE
.
这file_remoteDirectory
标头保存原始远程目录,而file_remoteFile
header 保存文件名。
这file_renameTo
header 保存新路径。
从 5.5.6 版本开始,remoteDirectoryExpression
可用于mv
为了方便起见,命令。
如果“from”文件不是完整的文件路径,则remoteDirectoryExpression
用作远程目录。
这同样适用于“to”文件,例如,如果任务只是重命名某个目录中的远程文件。
其他命令信息
这get
和mget
命令支持local-filename-generator-expression
属性。
它定义了一个 SpEL 表达式,用于在传输期间生成本地文件的名称。
评估上下文的根对象是请求消息。
这remoteFileName
变量也可用。
它特别有用mget
(例如:local-filename-generator-expression="#remoteFileName.toUpperCase() + headers.foo"
).
这get
和mget
命令支持local-directory-expression
属性。
它定义了一个 SpEL 表达式,用于在传输过程中生成本地目录的名称。
评估上下文的根对象是请求消息。
这remoteDirectory
变量也可用。
它对 mget 特别有用(例如:local-directory-expression="'/tmp/local/' + #remoteDirectory.toUpperCase() + headers.myheader"
).
此属性与local-directory
属性。
对于所有命令,网关的“expression”属性保存命令执行的路径。
对于mget
命令,则表达式的计算结果可能为 ,这意味着检索所有文件,*
somedirectory/*
以及以 结尾的其他值。*
以下示例显示了为ls
命令:
<int-ftp:outbound-gateway id="gateway1"
session-factory="ftpSessionFactory"
request-channel="inbound1"
command="ls"
command-options="-1"
expression="payload"
reply-channel="toSplitter"/>
发送到toSplitter
channel 是String
对象,每个对象都包含一个文件名。
如果您省略了command-options="-1"
,则有效负载将是FileInfo
对象。
您可以将选项作为空格分隔的列表提供(例如,command-options="-1 -dirs -links"
).
从 4.2 版开始,GET
,MGET
,PUT
和MPUT
命令支持FileExistsMode
属性 (mode
使用命名空间支持时)。
这会影响本地文件存在时的行为 (GET
和MGET
) 或远程文件存在 (PUT
和MPUT
).
支持的模式包括REPLACE
,APPEND
,FAIL
和IGNORE
.
为了向后兼容,默认模式PUT
和MPUT
作是REPLACE
.
为GET
和MGET
作,默认值为FAIL
.
使用 Java 配置进行配置
以下 Spring Boot 应用程序显示了如何使用 Java 配置出站网关的示例:
@SpringBootApplication
public class SftpJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(SftpJavaApplication.class)
.web(false)
.run(args);
}
@Bean
@ServiceActivator(inputChannel = "sftpChannel")
public MessageHandler handler() {
return new SftpOutboundGateway(ftpSessionFactory(), "ls", "'my_remote_dir/'");
}
}
使用 Java DSL 进行配置
以下 Spring Boot 应用程序显示了如何使用 Java DSL 配置出站网关的示例:
@SpringBootApplication
public class SftpJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(SftpJavaApplication.class)
.web(false)
.run(args);
}
@Bean
public SessionFactory<SftpClient.DirEntry> sftpSessionFactory() {
DefaultSftpSessionFactory sf = new DefaultSftpSessionFactory();
sf.setHost("localhost");
sf.setPort(port);
sf.setUsername("foo");
sf.setPassword("foo");
factory.setTestSession(true);
return new CachingSessionFactory<>(sf);
}
@Bean
public QueueChannelSpec remoteFileOutputChannel() {
return MessageChannels.queue();
}
@Bean
public IntegrationFlow sftpMGetFlow() {
return IntegrationFlow.from("sftpMgetInputChannel")
.handle(Sftp.outboundGateway(sftpSessionFactory(),
AbstractRemoteFileOutboundGateway.Command.MGET, "payload")
.options(AbstractRemoteFileOutboundGateway.Option.RECURSIVE)
.regexFileNameFilter("(subSftpSource|.*1.txt)")
.localDirectoryExpression("'myDir/' + #remoteDirectory")
.localFilenameExpression("#remoteFileName.replaceFirst('sftpSource', 'localTarget')"))
.channel("remoteFileOutputChannel")
.get();
}
}
出站网关部分成功 (mget
和mput
)
对多个文件执行作时(使用mget
和mput
) 在传输一个或多个文件后一段时间可能会发生异常。
在这种情况下(从 4.2 版开始),一个PartialSuccessException
被抛出。
以及通常的MessagingException
属性 (failedMessage
和cause
),则此异常具有两个附加属性:
-
partialResults
:成功传输结果。 -
derivedInput
:从请求消息生成的文件列表(例如,要为mput
).
这些属性允许您确定哪些文件已成功传输,哪些文件未成功传输。
在递归的情况下mput
这PartialSuccessException
可能已经嵌套PartialSuccessException
实例。
请考虑以下目录结构:
root/
|- file1.txt
|- subdir/
| - file2.txt
| - file3.txt
|- zoo.txt
如果异常发生在file3.txt
这PartialSuccessException
由网关抛出derivedInput
之file1.txt
,subdir
和zoo.txt
和partialResults
之file1.txt
.
其cause
是另一个PartialSuccessException
跟derivedInput
之file2.txt
和file3.txt
和partialResults
之file2.txt
.
消息会话回调
从 Spring Integration 4.2 版开始,您可以使用MessageSessionCallback<F, T>
使用<int-sftp:outbound-gateway/>
(SftpOutboundGateway
) 对Session<SftpClient.DirEntry>
使用requestMessage
上下文。
您可以将其用于任何非标准或低级 SFTP作(或多个),例如允许从集成流定义访问,或功能接口 (lambda) 实现注入。
以下示例使用 lambda:
@Bean
@ServiceActivator(inputChannel = "sftpChannel")
public MessageHandler sftpOutboundGateway(SessionFactory<SftpClient.DirEntry> sessionFactory) {
return new SftpOutboundGateway(sessionFactory,
(session, requestMessage) -> session.list(requestMessage.getPayload()));
}
另一个示例可能是对正在发送或检索的文件数据进行预处理或后处理。
使用 XML 配置时,<int-sftp:outbound-gateway/>
提供一个session-callback
属性,该属性允许您指定MessageSessionCallback
bean 名称。
这session-callback 与command 和expression 属性。
使用 Java 配置时,SftpOutboundGateway class 提供不同的构造函数。 |
Apache Mina SFTP 服务器事件
这ApacheMinaSftpEventListener
,在 5.2 版中添加,监听某些 Apache Mina SFTP 服务器事件并将它们发布为ApplicationEvent
可以由任何ApplicationListener
豆@EventListener
bean 方法或事件入站通道适配器。
目前,支持的事件包括:
-
SessionOpenedEvent
- 打开了客户端会话 -
DirectoryCreatedEvent
- 创建了一个目录 -
FileWrittenEvent
- 文件已写入 -
PathMovedEvent
- 文件或目录已重命名 -
PathRemovedEvent
- 文件或目录被删除 -
SessionClosedEvent
- 客户端已断开连接
其中每一个都是ApacheMinaSftpEvent
;您可以配置单个侦听器来接收所有事件类型。
这source
每个事件的属性是ServerSession
,您可以从中获取客户地址等信息;一个方便的getSession()
方法。
要使用侦听器(必须是 Spring bean)配置服务器,只需将其添加到SftpSubsystemFactory
:
server = SshServer.setUpDefaultServer();
...
SftpSubsystemFactory sftpFactory = new SftpSubsystemFactory();
sftpFactory.addSftpEventListener(apacheMinaSftpEventListenerBean);
...
要使用 Spring Integration 事件适配器使用这些事件:
@Bean
public ApplicationEventListeningMessageProducer eventsAdapter() {
ApplicationEventListeningMessageProducer producer =
new ApplicationEventListeningMessageProducer();
producer.setEventTypes(ApacheMinaSftpEvent.class);
producer.setOutputChannel(eventChannel());
return producer;
}
远程文件信息
从 5.2 版开始,SftpStreamingMessageSource
(SFTP 流式入站通道适配器)、SftpInboundFileSynchronizingMessageSource
(SFTP 入站通道适配器)和 “read”命令SftpOutboundGateway
(SFTP 出站网关)在消息中提供其他标头以生成有关远程文件的信息:
-
FileHeaders.REMOTE_HOST_PORT
- 文件传输作期间远程会话已连接的 host:port 对; -
FileHeaders.REMOTE_DIRECTORY
- 已执行作的远程目录; -
FileHeaders.REMOTE_FILE
- 远程文件名;仅适用于单文件作。
由于SftpInboundFileSynchronizingMessageSource
不会针对远程文件生成消息,但使用本地副本,AbstractInboundFileSynchronizer
将有关远程文件的信息存储在MetadataStore
(可以在外部配置)的 URI 样式 (protocol://host:port/remoteDirectory#remoteFileName
) 在同步作期间。
此元数据由SftpInboundFileSynchronizingMessageSource
轮询本地文件时。
删除本地文件时,建议删除其元数据条目。
这AbstractInboundFileSynchronizer
提供一个removeRemoteFileMetadata()
回调。
此外,还有一个setMetadataStorePrefix()
用于元数据键。
建议将此前缀与MetadataStore
-基于FileListFilter
实现,当相同时MetadataStore
instance 在这些组件之间共享,以避免条目覆盖,因为 filter 和AbstractInboundFileSynchronizer
对元数据条目键使用相同的本地文件名。