Spring AMQP version 1.6 introduced a mechanism to allow the specification of a default user ID for outbound messages.
It has always been possible to set the AmqpHeaders.USER_ID
header, which now takes precedence over the default.
This might be useful to message recipients.
For inbound messages, if the message publisher sets the property, it is made available in the AmqpHeaders.RECEIVED_USER_ID
header.
Note that RabbitMQ validates that the user ID is the actual user ID for the connection or that the connection allows impersonation.
To configure a default user ID for outbound messages, configure it on a RabbitTemplate
and configure the outbound adapter or gateway to use that template.
Similarly, to set the user ID property on replies, inject an appropriately configured template into the inbound gateway.
See the Spring AMQP documentation for more information.
14.11. Delayed Message Exchange
Spring AMQP supports the RabbitMQ Delayed Message Exchange Plugin.
For inbound messages, the x-delay
header is mapped to the AmqpHeaders.RECEIVED_DELAY
header.
Setting the AMQPHeaders.DELAY
header causes the corresponding x-delay
header to be set in outbound messages.
You can also specify the delay
and delayExpression
properties on outbound endpoints (delay-expression
when using XML configuration).
These properties take precedence over the AmqpHeaders.DELAY
header.
14.12. AMQP-backed Message Channels
There are two message channel implementations available.
One is point-to-point, and the other is publish-subscribe.
Both of these channels provide a wide range of configuration attributes for the underlying AmqpTemplate
and
SimpleMessageListenerContainer
(as shown earlier in this chapter for the channel adapters and gateways).
However, the examples we show here have minimal configuration.
Explore the XML schema to view the available attributes.
A point-to-point channel might look like the following example:
<int-amqp:channel id="p2pChannel"/>
Under the covers, the preceding example causes a Queue
named si.p2pChannel
to be declared, and this channel sends to that Queue
(technically, by sending to the no-name direct exchange with a routing key that matches the name of this Queue
).
This channel also registers a consumer on that Queue
.
If you want the channel to be “pollable” instead of message-driven, provide the message-driven
flag with a value of false
, as the following example shows:
<int-amqp:channel id="p2pPollableChannel" message-driven="false"/>
A publish-subscribe channel might look like the following:
<int-amqp:publish-subscribe-channel id="pubSubChannel"/>
Under the covers, the preceding example causes a fanout exchange named si.fanout.pubSubChannel
to be declared, and this channel sends to that fanout exchange.
This channel also declares a server-named exclusive, auto-delete, non-durable Queue
and binds that to the fanout exchange while registering a consumer on that Queue
to receive messages.
There is no “pollable” option for a publish-subscribe-channel.
It must be message-driven.
Starting with version 4.1, AMQP-backed message channels (in conjunction with channel-transacted
) support
template-channel-transacted
to separate transactional
configuration for the AbstractMessageListenerContainer
and
for the RabbitTemplate
.
Note that, previously, channel-transacted
was true
by default.
Now, by default, it is false
for the AbstractMessageListenerContainer
.
Prior to version 4.3, AMQP-backed channels only supported messages with Serializable
payloads and headers.
The entire message was converted (serialized) and sent to RabbitMQ.
Now, you can set the extract-payload
attribute (or setExtractPayload()
when using Java configuration) to true
.
When this flag is true
, the message payload is converted and the headers are mapped, in a manner similar to when you use channel adapters.
This arrangement lets AMQP-backed channels be used with non-serializable payloads (perhaps with another message converter, such as the Jackson2JsonMessageConverter
).
See AMQP Message Headers for more about the default mapped headers.
You can modify the mapping by providing custom mappers that use the outbound-header-mapper
and inbound-header-mapper
attributes.
You can now also specify a default-delivery-mode
, which is used to set the delivery mode when there is no amqp_deliveryMode
header.
By default, Spring AMQP MessageProperties
uses PERSISTENT
delivery mode.
| As with other persistence-backed channels, AMQP-backed channels are intended to provide message persistence to avoid message loss. They are not intended to distribute work to other peer applications. For that purpose, use channel adapters instead. |
Starting with version 5.0, the pollable channel now blocks the poller thread for the specified receiveTimeout (the default is 1 second).
Previously, unlike other PollableChannel implementations, the thread returned immediately to the scheduler if no message was available, regardless of the receive timeout.
Blocking is a little more expensive than using a basicGet() to retrieve a message (with no timeout), because a consumer has to be created to receive each message.
To restore the previous behavior, set the poller’s receiveTimeout to 0.
|
14.12.1. Configuring with Java Configuration
The following example shows how to configure the channels with Java configuration:
@Bean
public AmqpChannelFactoryBean pollable(ConnectionFactory connectionFactory) {
AmqpChannelFactoryBean factoryBean = new AmqpChannelFactoryBean();
factoryBean.setConnectionFactory(connectionFactory);
factoryBean.setQueueName("foo");
factoryBean.setPubSub(false);
return factoryBean;
}
@Bean
public AmqpChannelFactoryBean messageDriven(ConnectionFactory connectionFactory) {
AmqpChannelFactoryBean factoryBean = new AmqpChannelFactoryBean(true);
factoryBean.setConnectionFactory(connectionFactory);
factoryBean.setQueueName("bar");
factoryBean.setPubSub(false);
return factoryBean;
}
@Bean
public AmqpChannelFactoryBean pubSub(ConnectionFactory connectionFactory) {
AmqpChannelFactoryBean factoryBean = new AmqpChannelFactoryBean(true);
factoryBean.setConnectionFactory(connectionFactory);
factoryBean.setQueueName("baz");
factoryBean.setPubSub(false);
return factoryBean;
}
14.12.2. Configuring with the Java DSL
The following example shows how to configure the channels with the Java DSL:
@Bean
public IntegrationFlow pollableInFlow(ConnectionFactory connectionFactory) {
return IntegrationFlows.from(...)
...
.channel(Amqp.pollableChannel(connectionFactory)
.queueName("foo"))
...
.get();
}
@Bean
public IntegrationFlow messageDrivenInFow(ConnectionFactory connectionFactory) {
return IntegrationFlows.from(...)
...
.channel(Amqp.channel(connectionFactory)
.queueName("bar"))
...
.get();
}
@Bean
public IntegrationFlow pubSubInFlow(ConnectionFactory connectionFactory) {
return IntegrationFlows.from(...)
...
.channel(Amqp.publisSubscribeChannel(connectionFactory)
.queueName("baz"))
...
.get();
}
14.13. AMQP Message Headers
14.13.1. Overview
The Spring Integration AMQP Adapters automatically map all AMQP properties and headers.
(This is a change from 4.3 - previously, only standard headers were mapped).
By default, these properties are copied to and from Spring Integration MessageHeaders
by using the
DefaultAmqpHeaderMapper
.
You can pass in your own implementation of AMQP-specific header mappers, as the adapters have properties to support doing so.
Any user-defined headers within the AMQP MessageProperties
are copied to or from an AMQP message, unless explicitly negated by the requestHeaderNames
or replyHeaderNames
properties of the DefaultAmqpHeaderMapper
.
By default, for an outbound mapper, no x-*
headers are mapped.
See the caution that appears later in this section for why.
To override the default and revert to the pre-4.3 behavior, use STANDARD_REQUEST_HEADERS
and
STANDARD_REPLY_HEADERS
in the properties.
When mapping user-defined headers, the values can also contain simple wildcard patterns (such as 'thing*' or thing ) to be matched.
matches all headers.
|
Starting with version 4.1, the AbstractHeaderMapper
(a DefaultAmqpHeaderMapper
superclass) lets the NON_STANDARD_HEADERS
token be configured for the requestHeaderNames
and replyHeaderNames
properties (in addition to the existing STANDARD_REQUEST_HEADERS
and STANDARD_REPLY_HEADERS
) to map all user-defined headers.
The org.springframework.amqp.support.AmqpHeaders
class identifies the default headers that are used by the
DefaultAmqpHeaderMapper
:
-
amqp_appId
-
amqp_clusterId
-
amqp_contentEncoding
-
amqp_contentLength
-
content-type
(see contentType Header) -
amqp_correlationId
-
amqp_delay
-
amqp_deliveryMode
-
amqp_deliveryTag
-
amqp_expiration
-
amqp_messageCount
-
amqp_messageId
-
amqp_receivedDelay
-
amqp_receivedDeliveryMode
-
amqp_receivedExchange
-
amqp_receivedRoutingKey
-
amqp_redelivered
-
amqp_replyTo
-
amqp_timestamp
-
amqp_type
-
amqp_userId
-
amqp_publishConfirm
-
amqp_publishConfirmNackCause
-
amqp_returnReplyCode
-
amqp_returnReplyText
-
amqp_returnExchange
-
amqp_returnRoutingKey
-
amqp_channel
-
amqp_consumerTag
-
amqp_consumerQueue
As mentioned earlier in this section, using a header mapping pattern of`` is a common way to copy all headers.
However, this can have some unexpected side effects, because certain RabbitMQ proprietary properties/headers are also
copied.
For example, when you use federation, the received message may have
a property named x-received-from , which contains the node that sent the message.
If you use the wildcard character for the request and reply header mapping on the inbound gateway, this header is copied, which may cause some issues with federation.
This reply message may be federated back to the sending broker, which may think that a message is looping and, as a result, silently drop it.
If you wish to use the convenience of wildcard header mapping, you may need to filter out some headers in the downstream flow.
For example, to avoid copying the x-received-from header back to the reply you can use <int:header-filter … header-names="x-received-from"> before sending the reply to the AMQP inbound gateway.
Alternatively, you can explicitly list those properties that you actually want mapped, instead of using wildcards.
For these reasons, for inbound messages, the mapper (by default) does not map any x-* headers.
It also does not map the deliveryMode to the amqp_deliveryMode header, to avoid propagation of that header from an inbound message to an outbound message.
Instead, this header is mapped to amqp_receivedDeliveryMode , which is not mapped on output.
|
Starting with version 4.3, patterns in the header mappings can be negated by preceding the pattern with !
.
Negated patterns get priority, so a list such as STANDARD_REQUEST_HEADERS,thing1,ba*,!thing2,!thing3,qux,!thing1
does not map thing1
(nor thing2
nor thing3
).
The standard headers plus bad
and qux
are mapped.
If you have a user-defined header that begins with ! that you do wish to map, you need to escape it with \ , as follows: STANDARD_REQUEST_HEADERS,\!myBangHeader .
The header named !myBangHeader is now mapped.
|
Starting with version 5.1, the DefaultAmqpHeaderMapper will fall back to mapping MessageHeaders.ID and MessageHeaders.TIMESTAMP to MessageProperties.messageId and MessageProperties.timestamp respectively, if the corresponding amqp_messageId or amqp_timestamp headers are not present on outbound messages.
Inbound properties will be mapped to the amqp_* headers as before.
It is useful to populate the messageId property when message consumers are using stateful retry.
|
14.13.2. contentType Header
Unlike other headers, the AmqpHeaders.CONTENT_TYPE
is not prefixed with amqp_
; this allows transparent passing of the contentType header across different technologies.
For example an inbound HTTP message sent to a RabbitMQ queue.
The contentType
header is mapped to Spring AMQP’s MessageProperties.contentType
property and that is subsequently mapped to RabbitMQ’s content_type
property.
Prior to version 5.1, this header was also mapped as an entry in the MessageProperties.headers
map; this was incorrect and, furthermore, the value could be wrong since the underlying Spring AMQP message converter might have changed the content type.
Such a change would be reflected in the first-class content_type
property, but not in the RabbitMQ headers map.
Inbound mapping ignored the headers map value.
contentType
is no longer mapped to an entry in the headers map.
14.14. Strict Message Ordering
This section describes message ordering for inbound and outbound messages.
14.14.1. Inbound
If you require strict ordering of inbound messages, you must configure the inbound listener container’s prefetchCount
property to 1
.
This is because, if a message fails and is redelivered, it arrives after existing prefetched messages.
Since Spring AMQP version 2.0, the prefetchCount
defaults to 250
for improved performance.
Strict ordering requirements come at the cost of decreased performance.
14.14.2. Outbound
Consider the following integration flow:
@Bean
public IntegrationFlow flow(RabbitTemplate template) {
return IntegrationFlows.from(Gateway.class)
.split(s -> s.delimiters(","))
.<String, String>transform(String::toUpperCase)
.handle(Amqp.outboundAdapter(template).routingKey("rk"))
.get();
}
Suppose we send messages A
, B
and C
to the gateway.
While it is likely that messages A
, B
, C
are sent in order, there is no guarantee.
This is because the template “borrows” a channel from the cache for each send operation, and there is no guarantee that the same channel is used for each message.
One solution is to start a transaction before the splitter, but transactions are expensive in RabbitMQ and can reduce performance several hundred fold.
To solve this problem in a more efficient manner, starting with version 5.1, Spring Integration provides the BoundRabbitChannelAdvice
which is a HandleMessageAdvice
.
See Handling Message Advice.
When applied before the splitter, it ensures that all downstream operations are performed on the same channel and, optionally, can wait until publisher confirmations for all sent messages are received (if the connection factory is configured for confirmations).
The following example shows how to use BoundRabbitChannelAdvice
:
@Bean
public IntegrationFlow flow(RabbitTemplate template) {
return IntegrationFlows.from(Gateway.class)
.split(s -> s.delimiters(",")
.advice(new BoundRabbitChannelAdvice(template, Duration.ofSeconds(10))))
.<String, String>transform(String::toUpperCase)
.handle(Amqp.outboundAdapter(template).routingKey("rk"))
.get();
}
Notice that the same RabbitTemplate
(which implements RabbitOperations
) is used in the advice and the outbound adapter.
The advice runs the downstream flow within the template’s invoke
method so that all operations run on the same channel.
If the optional timeout is provided, when the flow completes, the advice calls the waitForConfirmsOrDie
method, which throws an exception if the confirmations are not received within the specified time.
There must be no thread handoffs in the downstream flow (QueueChannel , ExecutorChannel , and others).
|
14.15. AMQP Samples
To experiment with the AMQP adapters, check out the samples available in the Spring Integration samples git repository at https://github.com/SpringSource/spring-integration-samples
Currently, one sample demonstrates the basic functionality of the Spring Integration AMQP adapter by using an outbound channel adapter and an inbound channel adapter. As AMQP broker implementation in the sample uses RabbitMQ.
| In order to run the example, you need a running instance of RabbitMQ. A local installation with just the basic defaults suffices. For detailed RabbitMQ installation procedures, see https://www.rabbitmq.com/install.html |
Once the sample application is started, enter some text on the command prompt and a message containing that entered text is dispatched to the AMQP queue. In return, that message is retrieved by Spring Integration and printed to the console.
The following image illustrates the basic set of Spring Integration components used in this sample.
15. Spring ApplicationEvent
Support
Spring Integration provides support for inbound and outbound ApplicationEvents
, as defined by the underlying Spring Framework.
For more information about Spring’s support for events and listeners, see the Spring Reference Manual.
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-event</artifactId>
<version>5.1.5.RELEASE</version>
</dependency>
compile "org.springframework.integration:spring-integration-event:5.1.5.RELEASE"
15.1. Receiving Spring Application Events
To receive events and send them to a channel, you can define an instance of Spring Integration’s ApplicationEventListeningMessageProducer
.
This class is an implementation of Spring’s ApplicationListener
interface.
By default, it passes all received events as Spring Integration messages.
To limit based on the type of event, you can use the 'eventTypes' property to configure the list of event types that you want to receive.
If a received event has a Message
instance as its 'source', that Message
is passed as-is.
Otherwise, if a SpEL-based payloadExpression
has been provided, that is evaluated against the ApplicationEvent
instance.
If the event’s source is not a Message
instance and no payloadExpression
has been provided, the ApplicationEvent
itself is passed as the payload.
Starting with version 4.2, the ApplicationEventListeningMessageProducer
implements GenericApplicationListener
and can be configured to accept not only ApplicationEvent
types but any type for treating payload events (which are also supported since Spring Framework 4.2).
When the accepted event is an instance of PayloadApplicationEvent
, its payload
is used for the message to send.
For convenience, namespace support is provided to configure an ApplicationEventListeningMessageProducer
with the inbound-channel-adapter
element, as the following example shows:
<int-event:inbound-channel-adapter channel="eventChannel"
error-channel="eventErrorChannel"
event-types="example.FooEvent, example.BarEvent, java.util.Date"/>
<int:publish-subscribe-channel id="eventChannel"/>
In the preceding example, all application context events that match one of the types specified by the 'event-types' (optional) attribute are delivered as Spring Integration messages to the message channel named 'eventChannel'.
If a downstream component throws an exception, a MessagingException
that contains the failed message and exception is sent to the channel named 'eventErrorChannel'.
If no error-channel
is specified and the downstream channels are synchronous, the exception is propagated to the caller.
15.2. Sending Spring Application Events
To send Spring ApplicationEvents
, create an instance of the ApplicationEventPublishingMessageHandler
and register it within an endpoint.
This implementation of the MessageHandler
interface also implements Spring’s ApplicationEventPublisherAware
interface and consequently acts as a bridge between Spring Integration messages and ApplicationEvents
.
For convenience, namespace support is provided to configure an ApplicationEventPublishingMessageHandler
with the outbound-channel-adapter
element, as the following example shows:
<int:channel id="eventChannel"/>
<int-event:outbound-channel-adapter channel="eventChannel"/>
If you use a PollableChannel
(such as a Queue
), you can also provide poller
as a child element of the outbound-channel-adapter
element.
You can also optionally provide a task-executor
reference for that poller.
The following example demonstrates both:
<int:channel id="eventChannel">
<int:queue/>
</int:channel>
<int-event:outbound-channel-adapter channel="eventChannel">
<int:poller max-messages-per-poll="1" task-executor="executor" fixed-rate="100"/>
</int-event:outbound-channel-adapter>
<task:executor id="executor" pool-size="5"/>
In the preceding example, all messages sent to the 'eventChannel' channel are published as ApplicationEvent
instances to any relevant ApplicationListener
instances that are registered within the same Spring ApplicationContext
.
If the payload of the message is an ApplicationEvent
, it is passed as-is.
Otherwise, the message itself is wrapped in a MessagingEvent
instance.
Starting with version 4.2, you can configure the ApplicationEventPublishingMessageHandler
(<int-event:outbound-channel-adapter>
) with the publish-payload
boolean attribute to publish to the application context payload
as is, instead of wrapping it to a MessagingEvent
instance.
16. Feed Adapter
Spring Integration provides support for syndication through feed adapters. The implementation is based on the ROME Framework.
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-feed</artifactId>
<version>5.1.5.RELEASE</version>
</dependency>
compile "org.springframework.integration:spring-integration-feed:5.1.5.RELEASE"
Web syndication is a way to publish material such as news stories, press releases, blog posts, and other items typically available on a website but also made available in a feed format such as RSS or ATOM.
Spring integration provides support for web syndication through its 'feed' adapter and provides convenient namespace-based configuration for it. To configure the 'feed' namespace, include the following elements within the headers of your XML configuration file:
xmlns:int-feed="http://www.springframework.org/schema/integration/feed"
xsi:schemaLocation="http://www.springframework.org/schema/integration/feed
https://www.springframework.org/schema/integration/feed/spring-integration-feed.xsd"
16.1. Feed Inbound Channel Adapter
The only adapter you really need to provide support for retrieving feeds is an inbound channel adapter. It lets you subscribe to a particular URL. The following example shows a possible configuration:
<int-feed:inbound-channel-adapter id="feedAdapter"
channel="feedChannel"
url="https://feeds.bbci.co.uk/news/rss.xml">
<int:poller fixed-rate="10000" max-messages-per-poll="100" />
</int-feed:inbound-channel-adapter>
In the preceding configuration, we are subscribing to a URL identified by the url
attribute.
As news items are retrieved, they are converted to messages and sent to a channel identified by the channel
attribute.
The payload of each message is a com.sun.syndication.feed.synd.SyndEntry
instance.
Each one encapsulates various data about a news item (content, dates, authors, and other details).
The inbound feed channel adapter is a polling consumer.
That means that you must provide a poller configuration.
However, one important thing you must understand with regard to a feed is that its inner workings are slightly different then most other polling consumers.
When an inbound feed adapter is started, it does the first poll and receives a com.sun.syndication.feed.synd.SyndEntryFeed
instance.
That object contains multiple SyndEntry
objects.
Each entry is stored in the local entry queue and is released based on the value in the max-messages-per-poll
attribute, such that each message contains a single entry.
If, during retrieval of the entries from the entry queue, the queue has become empty, the adapter attempts to update the feed, thereby populating the queue with more entries (SyndEntry
instances), if any are available.
Otherwise the next attempt to poll for a feed is determined by the trigger of the poller (every ten seconds in the preceding configuration).
16.2. Duplicate Entries
Polling for a feed can result in entries that have already been processed (“I already read that news item, why are you showing it to me again?”).
Spring Integration provides a convenient mechanism to eliminate the need to worry about duplicate entries.
Each feed entry has a “published date” field.
Every time a new Message
is generated and sent, Spring Integration stores the value of the latest published date in an instance of the MetadataStore
strategy (see Metadata Store).
The key used to persist the latest published date is the value of the (required) id attribute of the feed inbound channel adapter component plus the feedUrl (if any) from the adapter’s configuration.
|
16.3. Other Options
Starting with version 5.0, the deprecated com.rometools.fetcher.FeedFetcher
option has been removed and an overloaded FeedEntryMessageSource
constructor for an org.springframework.core.io.Resource
is provided.
This is useful when the feed source is not an HTTP endpoint but is any other resource (such as local or remote on FTP).
In the FeedEntryMessageSource
logic, such a resource (or provided URL
) is parsed by the SyndFeedInput
to the SyndFeed
object for the processing mentioned earlier.
You can also inject a customized SyndFeedInput
(for example, with the allowDoctypes
option) instance into the FeedEntryMessageSource
.
16.4. Java DSL Configuration
The following Spring Boot application shows an example of how to configure the inbound adapter with the Java DSL:
@SpringBootApplication
public class FeedJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(FeedJavaApplication.class)
.web(false)
.run(args);
}
@Value("org/springframework/integration/feed/sample.rss")
private Resource feedResource;
@Bean
public MetadataStore metadataStore() {
PropertiesPersistingMetadataStore metadataStore = new PropertiesPersistingMetadataStore();
metadataStore.setBaseDirectory(tempFolder.getRoot().getAbsolutePath());
return metadataStore;
}
@Bean
public IntegrationFlow feedFlow() {
return IntegrationFlows
.from(Feed.inboundAdapter(this.feedResource, "feedTest")
.metadataStore(metadataStore()),
e -> e.poller(p -> p.fixedDelay(100)))
.channel(c -> c.queue("entries"))
.get();
}
}
17. File Support
Spring Integration’s file support extends the Spring Integration core with a dedicated vocabulary to deal with reading, writing, and transforming files.
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-file</artifactId>
<version>5.1.5.RELEASE</version>
</dependency>
compile "org.springframework.integration:spring-integration-file:5.1.5.RELEASE"
It provides a namespace that enables elements defining channel adapters dedicated to files and support for transformers that can read file contents into strings or byte arrays.
This section explains the workings of FileReadingMessageSource
and FileWritingMessageHandler
and how to configure them as beans.
It also discusses the support for dealing with files through file-specific implementations of Transformer
.
Finally, it explains the file-specific namespace.
17.1. Reading Files
A FileReadingMessageSource
can be used to consume files from the filesystem.
This is an implementation of MessageSource
that creates messages from a file system directory.
The following example shows how to configure a FileReadingMessageSource
:
<bean id="pollableFileSource"
class="org.springframework.integration.file.FileReadingMessageSource"
p:directory="${input.directory}"/>
To prevent creating messages for certain files, you can supply a FileListFilter
.
By default, we use the following filters:
-
IgnoreHiddenFileListFilter
-
AcceptOnceFileListFilter
The IgnoreHiddenFileListFilter
ensures that hidden files are not processed.
Note that the exact definition of hidden is system-dependent.
For example, on UNIX-based systems, a file beginning with a period character is considered to be hidden.
Microsoft Windows, on the other hand, has a dedicated file attribute to indicate hidden files.
|
Version 4.2 introduced the |
The AcceptOnceFileListFilter
ensures files are picked up only once from the directory.
|
The Since version 4.0, this filter requires a Since version 4.1.5, this filter has a new property ( |
The following example configures a FileReadingMessageSource
with a filter:
<bean id="pollableFileSource"
class="org.springframework.integration.file.FileReadingMessageSource"
p:inputDirectory="${input.directory}"
p:filter-ref="customFilterBean"/>
A common problem with reading files is that a file may be detected before it is ready (that is, some other process may still be writing the file).
The default AcceptOnceFileListFilter
does not prevent this.
In most cases, this can be prevented if the file-writing process renames each file as soon as it is ready for reading.
A filename-pattern
or filename-regex
filter that accepts only files that are ready (perhaps based on a known suffix), composed with the default AcceptOnceFileListFilter
, allows for this situation.
The CompositeFileListFilter
enables the composition, as the following example shows:
<bean id="pollableFileSource"
class="org.springframework.integration.file.FileReadingMessageSource"
p:inputDirectory="${input.directory}"
p:filter-ref="compositeFilter"/>
<bean id="compositeFilter"
class="org.springframework.integration.file.filters.CompositeFileListFilter">
<constructor-arg>
<list>
<bean class="o.s.i.file.filters.AcceptOnceFileListFilter"/>
<bean class="o.s.i.file.filters.RegexPatternFileListFilter">
<constructor-arg value="^test.*$"/>
</bean>
</list>
</constructor-arg>
</bean>
If it is not possible to create the file with a temporary name and rename to the final name, Spring Integration provides another alternative.
Version 4.2 added the LastModifiedFileListFilter
.
This filter can be configured with an age
property so that only files older than this value are passed by the filter.
The age defaults to 60 seconds, but you should choose an age that is large enough to avoid picking up a file early (due to, say, network glitches).
The following example shows how to configure a LastModifiedFileListFilter
:
<bean id="filter" class="org.springframework.integration.file.filters.LastModifiedFileListFilter">
<property name="age" value="120" />
</bean>
Starting with version 4.3.7, a ChainFileListFilter
(an extension of CompositeFileListFilter
) has been introduced to allow scenarios when subsequent filters should only see the result of the previous filter.
(With the CompositeFileListFilter
, all filters see all the files, but it passes only files that have passed all filters).
An example of where the new behavior is required is a combination of LastModifiedFileListFilter
and AcceptOnceFileListFilter
, when we do not wish to accept the file until some amount of time has elapsed.
With the CompositeFileListFilter
, since the AcceptOnceFileListFilter
sees all the files on the first pass, it does not pass it later when the other filter does.
The CompositeFileListFilter
approach is useful when a pattern filter is combined with a custom filter that looks for a secondary file to indicate that file transfer is complete.
The pattern filter might only pass the primary file (such as something.txt
) but the “done” filter needs to see whether (for example) something.done
is present.
Say we have files a.txt
, a.done
, and b.txt
.
The pattern filter passes only a.txt
and b.txt
, while the “done” filter sees all three files and passes only a.txt
.
The final result of the composite filter is that only a.txt
is released.
With the ChainFileListFilter , if any filter in the chain returns an empty list, the remaining filters are not invoked.
|
Version 5.0 introduced an ExpressionFileListFilter
to execute SpEL expression against a file as a context evaluation root object.
For this purpose, all the XML components for file handling (local and remote), along with an existing filter
attribute, have been supplied with the filter-expression
option, as the following example shows:
<int-file:inbound-channel-adapter
directory="${inputdir}"
filter-expression="name matches '.text'"
auto-startup="false"/>
Version 5.0.5 introduced the DiscardAwareFileListFilter
implementations that have an interest in rejected files.
For this purpose, such a filter implementation should be supplied with a callback through addDiscardCallback(Consumer<File>)
.
In the framework, this functionality is used from the FileReadingMessageSource.WatchServiceDirectoryScanner
, in combination with LastModifiedFileListFilter
.
Unlike the regular DirectoryScanner
, the WatchService
provides files for processing according to the events on the target file system.
At the moment of polling an internal queue with those files, the LastModifiedFileListFilter
may discard them because they are too young relative to its configured age
.
Therefore, we lose the file for future possible considerations.
The discard callback hook lets us retain the file in the internal queue so that it is available to be checked against the age
in subsequent polls.
The CompositeFileListFilter
also implements a DiscardAwareFileListFilter
and populates a discard callback to all its DiscardAwareFileListFilter
delegates.
Since CompositeFileListFilter matches the files against all delegates, the discardCallback may be called several times for the same file.
|
Starting with version 5.1, the FileReadingMessageSource
doesn’t check a directory for existence and doesn’t create it until its start()
is called (typically via wrapping SourcePollingChannelAdapter
).
Previously, there was no simple way to prevent an operation system permissions error when referencing the directory, for example from tests, or when permissions are applied later.
17.1.1. Message Headers
Starting with version 5.0, the FileReadingMessageSource
(in addition to the payload
as a polled File
) populates the following headers to the outbound Message
:
-
FileHeaders.FILENAME
: TheFile.getName()
of the file to send. Can be used for subsequent rename or copy logic. -
FileHeaders.ORIGINAL_FILE
: TheFile
object itself. Typically, this header is populated automatically by framework components (such as splitters or transformers) when we lose the originalFile
object. However, for consistency and convenience with any other custom use cases, this header can be useful to get access to the original file. -
FileHeaders.RELATIVE_PATH
: A new header introduced to represent the part of file path relative to the root directory for the scan. This header can be useful when the requirement is to restore a source directory hierarchy in the other places. For this purpose, theDefaultFileNameGenerator
(see "`Generating File Names) can be configured to use this header.
17.1.2. Directory Scanning and Polling
The FileReadingMessageSource
does not produce messages for files from the directory immediately.
It uses an internal queue for 'eligible files' returned by the scanner
.
The scanEachPoll
option is used to ensure that the internal queue is refreshed with the latest input directory content on each poll.
By default (scanEachPoll = false
), the FileReadingMessageSource
empties its queue before scanning the directory again.
This default behavior is particularly useful to reduce scans of large numbers of files in a directory.
However, in cases where custom ordering is required, it is important to consider the effects of setting this flag to true
.
The order in which files are processed may not be as expected.
By default, files in the queue are processed in their natural (path
) order.
New files added by a scan, even when the queue already has files, are inserted in the appropriate position to maintain that natural order.
To customize the order, the FileReadingMessageSource
can accept a Comparator<File>
as a constructor argument.
It is used by the internal (PriorityBlockingQueue
) to reorder its content according to the business requirements.
Therefore, to process files in a specific order, you should provide a comparator to the FileReadingMessageSource
rather than ordering the list produced by a custom DirectoryScanner
.
Version 5.0 introduced RecursiveDirectoryScanner
to perform file tree visiting.
The implementation is based on the Files.walk(Path start, int maxDepth, FileVisitOption… options)
functionality.
The root directory (DirectoryScanner.listFiles(File)
) argument is excluded from the result.
All other sub-directories inclusions and exclusions are based on the target FileListFilter
implementation.
For example, the SimplePatternFileListFilter
filters out directories by default.
See AbstractDirectoryAwareFileListFilter
and its implementations for more information.
17.1.3. Namespace Support
The configuration for file reading can be simplified by using the file-specific namespace. To do so, use the following template:
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-file="http://www.springframework.org/schema/integration/file"
xsi:schemaLocation="http://www.springframework.org/schema/beans
https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
https://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/file
https://www.springframework.org/schema/integration/file/spring-integration-file.xsd">
</beans>
Within this namespace, you can reduce the FileReadingMessageSource
and wrap it in an inbound Channel Adapter, as follows:
<int-file:inbound-channel-adapter id="filesIn1"
directory="file:${input.directory}" prevent-duplicates="true" ignore-hidden="true"/>
<int-file:inbound-channel-adapter id="filesIn2"
directory="file:${input.directory}"
filter="customFilterBean" />
<int-file:inbound-channel-adapter id="filesIn3"
directory="file:${input.directory}"
filename-pattern="test*" />
<int-file:inbound-channel-adapter id="filesIn4"
directory="file:${input.directory}"
filename-regex="test[0-9]+\.txt" />
The first channel adapter example relies on the default FileListFilter
implementations:
-
IgnoreHiddenFileListFilter
(do not process hidden files) -
AcceptOnceFileListFilter
(prevent duplication)
Therefore, you can also leave off the prevent-duplicates
and ignore-hidden
attributes, as they are true
by default.
|
Spring Integration 4.2 introduced the |
The second channel adapter example uses a custom filter, the third uses the filename-pattern
attribute to add an AntPathMatcher
based filter, and the fourth uses the filename-regex
attribute to add a regular expression pattern-based filter to the FileReadingMessageSource
.
The filename-pattern
and filename-regex
attributes are each mutually exclusive with the regular filter
reference attribute.
However, you can use the filter
attribute to reference an instance of CompositeFileListFilter
that combines any number of filters, including one or more pattern-based filters to fit your particular needs.
When multiple processes read from the same directory, you may want to lock files to prevent them from being picked up concurrently.
To do so, you can use a FileLocker
.
There is a java.nio
-based implementation available, but it is also possible to implement your own locking scheme.
The nio
locker can be injected as follows:
<int-file:inbound-channel-adapter id="filesIn"
directory="file:${input.directory}" prevent-duplicates="true">
<int-file:nio-locker/>
</int-file:inbound-channel-adapter>
You can configure a custom locker as follows:
<int-file:inbound-channel-adapter id="filesIn"
directory="file:${input.directory}" prevent-duplicates="true">
<int-file:locker ref="customLocker"/>
</int-file:inbound-channel-adapter>
When a file inbound adapter is configured with a locker, it takes responsibility for acquiring a lock before the file is allowed to be received.
It does not assume the responsibility to unlock the file.
If you have processed the file and keep the locks hanging around, you have a memory leak.
If this is a problem, you should call FileLocker.unlock(File file) yourself at the appropriate time.
|
When filtering and locking files is not enough, you might need to control the way files are listed entirely.
To implement this type of requirement, you can use an implementation of DirectoryScanner
.
This scanner lets you determine exactly what files are listed in each poll.
This is also the interface that Spring Integration uses internally to wire FileListFilter
instances and FileLocker
to the FileReadingMessageSource
.
You can inject a custom DirectoryScanner
into the <int-file:inbound-channel-adapter/>
on the scanner
attribute, as the following example shows:
<int-file:inbound-channel-adapter id="filesIn" directory="file:${input.directory}"
scanner="customDirectoryScanner"/>
Doing so gives you full freedom to choose the ordering, listing, and locking strategies.
It is also important to understand that filters (including patterns
, regex
, prevent-duplicates
, and others) and locker
instances are actually used by the scanner
.
Any of these attributes set on the adapter are subsequently injected into the internal scanner
.
For the case of an external scanner
, all filter and locker attributes are prohibited on the FileReadingMessageSource
.
They must be specified (if required) on that custom DirectoryScanner
.
In other words, if you inject a scanner
into the FileReadingMessageSource
, you should supply filter
and locker
on that scanner
, not on the FileReadingMessageSource
.
By default, the DefaultDirectoryScanner uses an IgnoreHiddenFileListFilter and an AcceptOnceFileListFilter .
To prevent their use, you can configure your own filter (such as AcceptAllFileListFilter ) or even set it to null .
|
17.1.4. WatchServiceDirectoryScanner
The FileReadingMessageSource.WatchServiceDirectoryScanner
relies on file-system events when new files are added to the directory.
During initialization, the directory is registered to generate events.
The initial file list is also built during initialization.
While walking the directory tree, any subdirectories encountered are also registered to generate events.
On the first poll, the initial file list from walking the directory is returned.
On subsequent polls, files from new creation events are returned.
If a new subdirectory is added, its creation event is used to walk the new subtree to find existing files and register any new subdirectories found.
There is an issue with WatchKey when its internal events queue is not drained by the program as quickly as the directory modification events occur.
If the queue size is exceeded, a StandardWatchEventKinds.OVERFLOW is emitted to indicate that some file system events may be lost.
In this case, the root directory is re-scanned completely.
To avoid duplicates, consider using an appropriate FileListFilter (such as the AcceptOnceFileListFilter ) or removing files when processing is complete.
|
The WatchServiceDirectoryScanner
can be enabled through the FileReadingMessageSource.use-watch-service
option, which is mutually exclusive with the scanner
option.
An internal FileReadingMessageSource.WatchServiceDirectoryScanner
instance is populated for the provided directory
.
In addition, now the WatchService
polling logic can track the StandardWatchEventKinds.ENTRY_MODIFY
and StandardWatchEventKinds.ENTRY_DELETE
.
If you need to track the modification of existing files as well as new files, you should implement the ENTRY_MODIFY
events logic in the FileListFilter
.
Otherwise, the files from those events are treated the same way.
The ResettableFileListFilter
implementations pick up the ENTRY_DELETE
events.
Consequently, their files are provided for the remove()
operation.
When this event is enabled, filters such as the AcceptOnceFileListFilter
have the file removed
As a result, if a file with the same name appears, it passes the filter and is sent as a message.
For this purpose, the watch-events
property (FileReadingMessageSource.setWatchEvents(WatchEventType… watchEvents)
) has been introduced.
(WatchEventType
is a public inner enumeration in FileReadingMessageSource
.)
With such an option, we can use one downstream flow logic for new files and use some other logic for modified files.
The following example shows how to configure different logic for create and modify events in the same directory:
<int-file:inbound-channel-adapter id="newFiles"
directory="${input.directory}"
use-watch-service="true"/>
<int-file:inbound-channel-adapter id="modifiedFiles"
directory="${input.directory}"
use-watch-service="true"
filter="acceptAllFilter"
watch-events="MODIFY"/> <!-- The default is CREATE. -->
17.1.5. Limiting Memory Consumption
You can use a HeadDirectoryScanner
to limit the number of files retained in memory.
This can be useful when scanning large directories.
With XML configuration, this is enabled by setting the queue-size
property on the inbound channel adapter.
Prior to version 4.2, this setting was incompatible with the use of any other filters.
Any other filters (including prevent-duplicates="true"
) overwrote the filter used to limit the size.
|
The use of a Generally, instead of using an |
17.1.6. Configuring with Java Configuration
The following Spring Boot application shows an example of how to configure the inbound adapter with Java configuration:
@SpringBootApplication
public class FileReadingJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(FileReadingJavaApplication.class)
.web(false)
.run(args);
}
@Bean
public MessageChannel fileInputChannel() {
return new DirectChannel();
}
@Bean
@InboundChannelAdapter(value = "fileInputChannel", poller = @Poller(fixedDelay = "1000"))
public MessageSource<File> fileReadingMessageSource() {
FileReadingMessageSource source = new FileReadingMessageSource();
source.setDirectory(new File(INBOUND_PATH));
source.setFilter(new SimplePatternFileListFilter("*.txt"));
return source;
}
@Bean
@Transformer(inputChannel = "fileInputChannel", outputChannel = "processFileChannel")
public FileToStringTransformer fileToStringTransformer() {
return new FileToStringTransformer();
}
}
17.1.7. Configuring with the Java DSL
The following Spring Boot application shows an example of how to configure the inbound adapter with the Java DSL:
@SpringBootApplication
public class FileReadingJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(FileReadingJavaApplication.class)
.web(false)
.run(args);
}
@Bean
public IntegrationFlow fileReadingFlow() {
return IntegrationFlows
.from(Files.inboundAdapter(new File(INBOUND_PATH))
.patternFilter("*.txt"),
e -> e.poller(Pollers.fixedDelay(1000)))
.transform(Files.toStringTransformer())
.channel("processFileChannel")
.get();
}
}
17.1.8. 'tail’ing Files
Another popular use case is to get 'lines' from the end (or tail) of a file, capturing new lines when they are added.
Two implementations are provided.
The first, OSDelegatingFileTailingMessageProducer
, uses the native tail
command (on operating systems that have one).
This is generally the most efficient implementation on those platforms.
For operating systems that do not have a tail
command, the second implementation, ApacheCommonsFileTailingMessageProducer
, uses the Apache commons-io
Tailer
class.
In both cases, file system events, such as files being unavailable and other events, are published as ApplicationEvent
instances by using the normal Spring event publishing mechanism.
Examples of such events include the following:
[message=tail: cannot open `/tmp/somefile' for reading:
No such file or directory, file=/tmp/somefile]
[message=tail: `/tmp/somefile' has become accessible, file=/tmp/somefile]
[message=tail: `/tmp/somefile' has become inaccessible:
No such file or directory, file=/tmp/somefile]
[message=tail: `/tmp/somefile' has appeared;
following end of new file, file=/tmp/somefile]
The sequence of events shown in the preceding example might occur, for example, when a file is rotated.
Starting with version 5.0, a FileTailingIdleEvent
is emitted when there is no data in the file during idleEventInterval
.
The following example shows what such an event looks like:
[message=Idle timeout, file=/tmp/somefile] [idle time=5438]
Not all platforms that support a tail command provide these status messages.
|
Messages emitted from these endpoints have the following headers:
-
FileHeaders.ORIGINAL_FILE
: TheFile
object -
FileHeaders.FILENAME
: The file name (File.getName()
)
In versions prior to version 5.0, the FileHeaders.FILENAME header contained a string representation of the file’s absolute path.
You can now obtain that string representation by calling getAbsolutePath() on the original file header.
|
The following example creates a native adapter with the default options ('-F -n 0', meaning to follow the file name from the current end).
<int-file:tail-inbound-channel-adapter id="native"
channel="input"
task-executor="exec"
file="/tmp/foo"/>
The following example creates a native adapter with '-F -n +0' options (meaning follow the file name, emitting all existing lines).
<int-file:tail-inbound-channel-adapter id="native"
channel="input"
native-options="-F -n +0"
task-executor="exec"
file-delay=10000
file="/tmp/foo"/>
If the tail
command fails (on some platforms, a missing file causes the tail
to fail, even with -F
specified), the command is retried every 10 seconds.
By default, native adapters capture from standard output and send the content as messages.
They also capture from standard error to raise events.
Starting with version 4.3.6, you can discard the standard error events by setting the enable-status-reader
to false
, as the following example shows:
<int-file:tail-inbound-channel-adapter id="native"
channel="input"
enable-status-reader="false"
task-executor="exec"
file="/tmp/foo"/>
In the following example, IdleEventInterval
is set to 5000
, meaning that, if no lines are written for five seconds, FileTailingIdleEvent
is triggered every five seconds:
<int-file:tail-inbound-channel-adapter id="native"
channel="input"
idle-event-interval="5000"
task-executor="exec"
file="/tmp/somefile"/>
This can be useful when you need to stop the adapter.
The following example creates an Apache commons-io
Tailer
adapter that examines the file for new lines every two seconds and checks for existence of a missing file every ten seconds:
<int-file:tail-inbound-channel-adapter id="apache"
channel="input"
task-executor="exec"
file="/tmp/bar"
delay="2000"
end="false" (1)
reopen="true" (2)
file-delay="10000"/>
| 1 | The file is tailed from the beginning (end="false" ) instead of the end (which is the default). |
| 2 | The file is reopened for each chunk (the default is to keep the file open). |
Specifying the delay , end or reopen attributes forces the use of the Apache commons-io adapter and makes the native-options attribute unavailable.
|
17.1.9. Dealing With Incomplete Data
A common problem in file-transfer scenarios is how to determine that the transfer is complete so that you do not start reading an incomplete file.
A common technique to solve this problem is to write the file with a temporary name and then atomically rename it to the final name.
This technique, together with a filter that masks the temporary file from being picked up by the consumer, provides a robust solution.
This technique is used by Spring Integration components that write files (locally or remotely).
By default, they append .writing
to the file name and remove it when the transfer is complete.
Another common technique is to write a second “marker” file to indicate that the file transfer is complete.
In this scenario, you should not consider somefile.txt
(for example) to be available for use until somefile.txt.complete
is also present.
Spring Integration version 5.0 introduced new filters to support this mechanism.
Implementations are provided for the file system (FileSystemMarkerFilePresentFileListFilter
), FTP and SFTP.
They are configurable such that the marker file can have any name, although it is usually related to the file being transferred.
See the Javadoc for more information.
17.2. Writing files
To write messages to the file system, you can use a FileWritingMessageHandler
.
This class can deal with the following payload types:
-
File
-
String
-
byte array
-
InputStream
(since version 4.2)
For a String payload, you can configure the encoding and the charset.
To make things easier, you can configure the FileWritingMessageHandler
as part of an outbound channel adapter or outbound gateway by using the XML namespace.
Starting with version 4.3, you can specify the buffer size to use when writing files.
Starting with version 5.1, you can provide a BiConsumer<File, Message<?>>
newFileCallback
which is triggered if you use FileExistsMode.APPEND
or FileExistsMode.APPEND_NO_FLUSH
and a new file has to be created.
This callback receives a newly created file and the message which triggered it.
This callback could be used to write a CSV header defined in the message header, for an example.
17.2.1. Generating File Names
In its simplest form, the FileWritingMessageHandler
requires only a destination directory for writing the files.
The name of the file to be written is determined by the handler’s FileNameGenerator
.
The default implementation looks for a message header whose key matches the constant defined as FileHeaders.FILENAME
.
Alternatively, you can specify an expression to be evaluated against the message to generate a file name — for example, headers['myCustomHeader'] + '.something'
.
The expression must evaluate to a String
.
For convenience, the DefaultFileNameGenerator
also provides the setHeaderName
method, letting you explicitly specify the message header whose value is to be used as the filename.
Once set up, the DefaultFileNameGenerator
employs the following resolution steps to determine the filename for a given message payload:
-
Evaluate the expression against the message and, if the result is a non-empty
String
, use it as the filename. -
Otherwise, if the payload is a
java.io.File
, use theFile
object’s filename. -
Otherwise, use the message ID appended with .
msg
as the filename.
When you use the XML namespace support, both the file outbound channel adapter and the file outbound gateway support the following mutually exclusive configuration attributes:
-
filename-generator
(a reference to aFileNameGenerator
implementation) -
filename-generator-expression
(an expression that evaluates to aString
)
While writing files, a temporary file suffix is used (its default is .writing
).
It is appended to the filename while the file is being written.
To customize the suffix, you can set the temporary-file-suffix
attribute on both the file outbound channel adapter and the file outbound gateway.
When using the APPEND file mode , the temporary-file-suffix attribute is ignored, since the data is appended to the file directly.
|
Starting with ,version 4.2.5, the generated file name (as a result of filename-generator
or filename-generator-expression
evaluation) can represent a child path together with the target file name.
It is used as a second constructor argument for File(File parent, String child)
as before.
However, in the past we did not create (mkdirs()
) directories for the child path, assuming only the file name.
This approach is useful for cases when we need to restore the file system tree to match the source directory — for example, when unzipping the archive and saving all the files in the target directory in the original order.
17.2.2. Specifying the Output Directory
Both, the file outbound channel adapter and the file outbound gateway provide two mutually exclusive configuration attributes for specifying the output directory:
-
directory
-
directory-expression
Spring Integration 2.2 introduced the directory-expression attribute.
|
Using the directory
Attribute
When you use the directory
attribute, the output directory is set to a fixed value, which is set when the FileWritingMessageHandler
is initialized.
If you do not specify this attribute, you must use the directory-expression
attribute.
Using the directory-expression
Attribute
If you want to have full SpEL support, you can use the directory-expression
attribute.
This attribute accepts a SpEL expression that is evaluated for each message being processed.
Thus, you have full access to a message’s payload and its headers when you dynamically specify the output file directory.
The SpEL expression must resolve to either a String
or to java.io.File
.
Furthermore, the resulting String
or File
must point to a directory.
If you do not specify the directory-expression
attribute, then you must set the directory
attribute.
Using the auto-create-directory
Attribute
By default, if the destination directory does not exist, the respective destination directory and any non-existing parent directories are automatically created.
To prevent that behavior, you can set the auto-create-directory
attribute to false
.
This attribute applies to both the directory
and the directory-expression
attributes.
|
When using the Instead of checking for the existence of the destination directory when the adapter is initialized, this check is now performed for each message being processed. Furthermore, if |
17.2.3. Dealing with Existing Destination Files
When you write files and the destination file already exists, the default behavior is to overwrite that target file.
You can change this behavior by setting the mode
attribute on the relevant file outbound components.
The following options exist:
-
REPLACE
(Default) -
REPLACE_IF_MODIFIED
-
APPEND
-
APPEND_NO_FLUSH
-
FAIL
-
IGNORE
Spring Integration 2.2 introduced the mode attribute and the APPEND , FAIL , and IGNORE options.
|
REPLACE
-
If the target file already exists, it is overwritten. If the
mode
attribute is not specified, this is the default behavior when writing files. REPLACE_IF_MODIFIED
-
If the target file already exists, it is overwritten only if the last modified timestamp differs from that of the source file. For
File
payloads, the payloadlastModified
time is compared to the existing file. For other payloads, theFileHeaders.SET_MODIFIED
(file_setModified
) header is compared to the existing file. If the header is missing or has a value that is not aNumber
, the file is always replaced. APPEND
-
This mode lets you append message content to the existing file instead of creating a new file each time. Note that this attribute is mutually exclusive with the
temporary-file-suffix
attribute because, when it appends content to the existing file, the adapter no longer uses a temporary file. The file is closed after each message. APPEND_NO_FLUSH
-
This option has the same semantics as
APPEND
, but the data is not flushed and the file is not closed after each message. This can provide a significant performance at the risk of data loss in the event of a failure. See Flushing Files When UsingAPPEND_NO_FLUSH
for more information. FAIL
-
If the target file exists, a
MessageHandlingException
is thrown. IGNORE
-
If the target file exists, the message payload is silently ignored.
When using a temporary file suffix (the default is .writing ), the IGNORE option applies if either the final file name or the temporary file name exists.
|
17.2.4. Flushing Files When Using APPEND_NO_FLUSH
The APPEND_NO_FLUSH
mode was added in version 4.3.
Using it can improve performance because the file is not closed after each message.
However, this can cause data loss in the event of a failure.
Spring Integration provides several flushing strategies to mitigate this data loss:
-
Use
flushInterval
. If a file is not written to for this period of time, it is automatically flushed. This is approximate and may be up to1.33x
this time (with an average of1.167x
).  Files with absolute path names matching the pattern are flushed. -
Provide the handler with a custom
MessageFlushPredicate
implementation to modify the action taken when a message is sent to thetrigger
method. -
Invoke one of the handler’s
flushIfNeeded
methods by passing in a customFileWritingMessageHandler.FlushPredicate
orFileWritingMessageHandler.MessageFlushPredicate
implementation.
The predicates are called for each open file. See the Javadoc for these interfaces for more information. Note that, since version 5.0, the predicate methods provide another parameter: the time that the current file was first written to if new or previously closed.
When using flushInterval
, the interval starts at the last write.
The file is flushed only if it is idle for the interval.
Starting with version 4.3.7, an additional property (flushWhenIdle
) can be set to false
, meaning that the interval starts with the first write to a previously flushed (or new) file.
17.2.5. File Timestamps
By default, the destination file’s lastModified
timestamp is the time when the file was created (except that an in-place rename retains the current timestamp).
Starting with version 4.3, you can now configure preserve-timestamp
(or setPreserveTimestamp(true)
when using Java configuration).
For File
payloads, this transfers the timestamp from the inbound file to the outbound (regardless of whether a copy was required).
For other payloads, if the FileHeaders.SET_MODIFIED
header (file_setModified
) is present, it is used to set the destination file’s lastModified
timestamp, as long as the header is a Number
.
17.2.6. File Permissions
Starting with version 5.0, when writing files to a file system that supports Posix permissions, you can specify those permissions on the outbound channel adapter or gateway.
The property is an integer and is usually supplied in the familiar octal format — for example, 0640
, meaning that the owner has read/write permissions, the group has read-only permission, and others have no access.
17.2.7. File Outbound Channel Adapter
The following example configures a file outbound channel adapter:
<int-file:outbound-channel-adapter id="filesOut" directory="${input.directory.property}"/>
The namespace-based configuration also supports a delete-source-files
attribute.
If set to true
, it triggers the deletion of the original source files after writing to a destination.
The default value for that flag is false
.
The following example shows how to set it to true
:
<int-file:outbound-channel-adapter id="filesOut"
directory="${output.directory}"
delete-source-files="true"/>
The delete-source-files attribute has an effect only if the inbound message has a File payload or if the FileHeaders.ORIGINAL_FILE header value contains either the source File instance or a String representing the original file path.
|
Starting with version 4.2, the FileWritingMessageHandler
supports an append-new-line
option.
If set to true
, a new line is appended to the file after a message is written.
The default attribute value is false
.
The following example shows how to use the append-new-line
option:
<int-file:outbound-channel-adapter id="newlineAdapter"
append-new-line="true"
directory="${output.directory}"/>
17.2.8. Outbound Gateway
In cases where you want to continue processing messages based on the written file, you can use the outbound-gateway
instead.
It plays a role similar to that of the outbound-channel-adapter
.
However, after writing the file, it also sends it to the reply channel as the payload of a message.
The following example configures an outbound gateway:
<int-file:outbound-gateway id="mover" request-channel="moveInput"
reply-channel="output"
directory="${output.directory}"
mode="REPLACE" delete-source-files="true"/>
As mentioned earlier, you can also specify the mode
attribute, which defines the behavior of how to deal with situations where the destination file already exists.
See Dealing with Existing Destination Files for further details.
Generally, when using the file outbound gateway, the result file is returned as the message payload on the reply channel.
This also applies when specifying the IGNORE
mode.
In that case the pre-existing destination file is returned.
If the payload of the request message was a file, you still have access to that original file through the message header.
See FileHeaders.ORIGINAL_FILE.
The 'outbound-gateway' works well in cases where you want to first move a file and then send it through a processing pipeline.
In such cases, you may connect the file namespace’s inbound-channel-adapter element to the outbound-gateway and then connect that gateway’s reply-channel to the beginning of the pipeline.
|
If you have more elaborate requirements or need to support additional payload types as input to be converted to file content, you can extend the FileWritingMessageHandler
, but a much better option is to rely on a Transformer
.
17.2.9. Configuring with Java Configuration
The following Spring Boot application shows an example of how to configure the inbound adapter with Java configuration:
@SpringBootApplication
@IntegrationComponentScan
public class FileWritingJavaApplication {
public static void main(String[] args) {
ConfigurableApplicationContext context =
new SpringApplicationBuilder(FileWritingJavaApplication.class)
.web(false)
.run(args);
MyGateway gateway = context.getBean(MyGateway.class);
gateway.writeToFile("foo.txt", new File(tmpDir.getRoot(), "fileWritingFlow"), "foo");
}
@Bean
@ServiceActivator(inputChannel = "writeToFileChannel")
public MessageHandler fileWritingMessageHandler() {
Expression directoryExpression = new SpelExpressionParser().parseExpression("headers.directory");
FileWritingMessageHandler handler = new FileWritingMessageHandler(directoryExpression);
handler.setFileExistsMode(FileExistsMode.APPEND);
return handler;
}
@MessagingGateway(defaultRequestChannel = "writeToFileChannel")
public interface MyGateway {
void writeToFile(@Header(FileHeaders.FILENAME) String fileName,
@Header(FileHeaders.FILENAME) File directory, String data);
}
}
17.2.10. Configuring with the Java DSL
The following Spring Boot application shows an example of how to configure the inbound adapter with the Java DSL:
@SpringBootApplication
public class FileWritingJavaApplication {
public static void main(String[] args) {
ConfigurableApplicationContext context =
new SpringApplicationBuilder(FileWritingJavaApplication.class)
.web(false)
.run(args);
MessageChannel fileWritingInput = context.getBean("fileWritingInput", MessageChannel.class);
fileWritingInput.send(new GenericMessage<>("foo"));
}
@Bean
public IntegrationFlow fileWritingFlow() {
return IntegrationFlows.from("fileWritingInput")
.enrichHeaders(h -> h.header(FileHeaders.FILENAME, "foo.txt")
.header("directory", new File(tmpDir.getRoot(), "fileWritingFlow")))
.handle(Files.outboundGateway(m -> m.getHeaders().get("directory")))
.channel(MessageChannels.queue("fileWritingResultChannel"))
.get();
}
}
17.3. File Transformers
To transform data read from the file system to objects and the other way around, you need to do some work.
Unlike FileReadingMessageSource
and to a lesser extent FileWritingMessageHandler
, you probably need your own mechanism to get the job done.
For this, you can implement the Transformer
interface.
Alternatively, you can extend the AbstractFilePayloadTransformer
for inbound messages.
Spring Integration provides some obvious implementations.
See the Javadoc for the Transformer
interface to see which Spring Integration classes implement it.
Similarly, you can check the Javadoc for the AbstractFilePayloadTransformer
class to see which Spring Integration classes extend it.
FileToByteArrayTransformer
extends AbstractFilePayloadTransformer
and transforms a File
object into a byte[]
by using Spring’s FileCopyUtils
.
It is often better to use a sequence of transformers than to put all transformations in a single class.
In that case the File
to byte[]
conversion might be a logical first step.
FileToStringTransformer
extends AbstractFilePayloadTransformer
convert a File
object to a String
.
If nothing else, this can be useful for debugging (consider using it with a wire tap).
To configure file-specific transformers, you can use the appropriate elements from the file namespace, as the following example shows:
<int-file:file-to-bytes-transformer input-channel="input" output-channel="output"
delete-files="true"/>
<int-file:file-to-string-transformer input-channel="input" output-channel="output"
delete-files="true" charset="UTF-8"/>
The delete-files
option signals to the transformer that it should delete the inbound file after the transformation is complete.
This is in no way a replacement for using an AcceptOnceFileListFilter
when the FileReadingMessageSource
is being used in a multi-threaded environment (such as when you use Spring Integration in general).
17.4. File Splitter
The FileSplitter
was added in version 4.1.2, and its namespace support was added in version 4.2.
The FileSplitter
splits text files into individual lines, based on BufferedReader.readLine()
.
By default, the splitter uses an Iterator
to emit lines one at a time as they are read from the file.
Setting the iterator
property to false
causes it to read all the lines into memory before emitting them as messages.
One use case for this might be if you want to detect I/O errors on the file before sending any messages containing lines.
However, it is only practical for relatively short files.
Inbound payloads can be File
, String
(a File
path), InputStream
, or Reader
.
Other payload types are emitted unchanged.
The following listing shows all the possible attributes for <int-file:splitter>
:
<int-file:splitter id="splitter" (1)
iterator="" (2)
markers="" (3)
markers-json="" (4)
apply-sequence="" (5)
requires-reply="" (6)
charset="" (7)
first-line-as-header="" (8)
input-channel="" (9)
output-channel="" (10)
send-timeout="" (11)
auto-startup="" (12)
order="" (13)
phase="" /> (14)
| 1 | The bean name of the splitter. |
| 2 | Set to true (the default) to use an iterator or false to load the file into memory before sending lines. |
| 3 | Set to true to emit start-of-file and end-of-file marker messages before and after the file data.
Markers are messages with FileSplitter.FileMarker payloads (with START and END values in the mark property).
You might use markers when sequentially processing files in a downstream flow where some lines are filtered.
They enable the downstream processing to know when a file has been completely processed.
In addition, a file_marker header that contains START or END is added to these messages.
The END marker includes a line count.
If the file is empty, only START and END markers are emitted with 0 as the lineCount .
The default is false .
When true , apply-sequence is false by default.
See also markers-json (the next attribute). |
| 4 | When markers is true, set this to true to have the FileMarker objects be converted to a JSON string.
Requires a supported JSON processor library (Jackson or Boon) on the classpath. |
| 5 | Set to false to disable the inclusion of sequenceSize and sequenceNumber headers in messages.
The default is true , unless markers is true .
When true and markers is true , the markers are included in the sequencing.
When true and iterator is true , the sequenceSize header is set to 0 , because the size is unknown. |
| 6 | Set to true to cause a RequiresReplyException to be thrown if there are no lines in the file.
The default is false . |
| 7 | Set the charset name to be used when reading text data into String payloads.
The default is the platform charset. |
| 8 | The header name for the first line to be carried as a header in the messages emitted for the remaining lines. Since version 5.0. |
| 9 | Set the input channel used to send messages to the splitter. |
| 10 | Set the output channel to which messages are sent. |
| 11 | Set the send timeout.
Only applies if the output-channel can block — such as a full QueueChannel . |
| 12 | Set to false to disable automatically starting the splitter when the context is refreshed.
The default is true . |
| 13 | Set the order of this endpoint if the input-channel is a <publish-subscribe-channel/> . |
| 14 | Set the startup phase for the splitter (used when auto-startup is true ). |
The FileSplitter
also splits any text-based InputStream
into lines.
Starting with version 4.3, when used in conjunction with an FTP or SFTP streaming inbound channel adapter or an FTP or SFTP outbound gateway that uses the stream
option to retrieve a file, the splitter automatically closes the session that supports the stream when the file is completely consumed
See FTP Streaming Inbound Channel Adapter and SFTP Streaming Inbound Channel Adapter as well as FTP Outbound Gateway and SFTP Outbound Gateway for more information about these facilities.
When using Java configuration, an additional constructor is available, as the following example shows:
public FileSplitter(boolean iterator, boolean markers, boolean markersJson)
When markersJson
is true, the markers are represented as a JSON string, as long as a suitable JSON processor library (such as Jackson or Boon) is on the classpath.
Version 5.0 introduced the firstLineAsHeader
option to specify that the first line of content is a header (such as column names in a CSV file).
The argument passed to this property is the header name under which the first line is carried as a header in the messages emitted for the remaining lines.
This line is not included in the sequence header (if applySequence
is true) nor in the lineCount
associated with FileMarker.END
.
If a file contains only the header line, the file is treated as empty and, therefore, only FileMarker
instances are emitted during splitting (if markers are enabled — otherwise, no messages are emitted).
By default (if no header name is set), the first line is considered to be data and becomes the payload of the first emitted message.
If you need more complex logic about header extraction from the file content (not first line, not the whole content of the line, not one particular header, and so on), consider using header enricher ahead of the FileSplitter
.
Note that the lines that have been moved to the headers might be filtered downstream from the normal content process.
17.4.1. Configuring with Java Configuration
The following Spring Boot application shows an example of how to configure a file splitter with Java configuration:
@Splitter(inputChannel="toSplitter")
@Bean
public MessageHandler fileSplitter() {
FileSplitter splitter = new FileSplitter(true, true);
splitter.setApplySequence(true);
splitter.setOutputChannel(outputChannel);
return splitter;
}
17.4.2. Configuring with the Java DSL
The following Spring Boot application shows an example of how to configure a file splitter with the Java DSL:
@SpringBootApplication
public class FileSplitterApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(FileSplitterApplication.class)
.web(false)
.run(args);
}
@Bean
public IntegrationFlow fileSplitterFlow() {
return IntegrationFlows
.from(Files.inboundAdapter(tmpDir.getRoot())
.filter(new ChainFileListFilter<File>()
.addFilter(new AcceptOnceFileListFilter<>())
.addFilter(new ExpressionFileListFilter<>(
new FunctionExpression<File>(f -> "foo.tmp".equals(f.getName()))))))
.split(Files.splitter()
.markers()
.charset(StandardCharsets.US_ASCII)
.firstLineAsHeader("fileHeader")
.applySequence(true))
.channel(c -> c.queue("fileSplittingResultChannel"))
.get();
}
}
18. FTP/FTPS Adapters
Spring Integration provides support for file transfer operations with FTP and FTPS.
The File Transfer Protocol (FTP) is a simple network protocol that lets you transfer files between two computers on the Internet. FTPS stands for “FTP over SSL”.
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-ftp</artifactId>
<version>5.1.5.RELEASE</version>
</dependency>
compile "org.springframework.integration:spring-integration-ftp:5.1.5.RELEASE"
There are two actors when it comes to FTP communication: client and server. To transfer files with FTP or FTPS, you use a client that initiates a connection to a remote computer that is running an FTP server. After the connection is established, the client can choose to send or receive copies of files.
Spring Integration supports sending and receiving files over FTP or FTPS by providing three client-side endpoints: inbound channel adapter, outbound channel adapter, and outbound gateway. It also provides convenient namespace-based configuration options for defining these client components.
To use the FTP namespace, add the following to the header of your XML file:
xmlns:int-ftp="http://www.springframework.org/schema/integration/ftp"
xsi:schemaLocation="http://www.springframework.org/schema/integration/ftp
https://www.springframework.org/schema/integration/ftp/spring-integration-ftp.xsd"
18.1. FTP Session Factory
Spring Integration provides factories you can use to create FTP (or FTPS) sessions.
18.1.1. Default Factories
| Starting with version 3.0, sessions are no longer cached by default. See FTP Session Caching. |
Before configuring FTP adapters, you must configure an FTP session factory.
You can configure the FTP Session Factory with a regular bean definition where the implementation class is o.s.i.ftp.session.DefaultFtpSessionFactory
.
The following example shows a basic configuration:
<bean id="ftpClientFactory"
class="org.springframework.integration.ftp.session.DefaultFtpSessionFactory">
<property name="host" value="localhost"/>
<property name="port" value="22"/>
<property name="username" value="kermit"/>
<property name="password" value="frog"/>
<property name="clientMode" value="0"/>
<property name="fileType" value="2"/>
<property name="bufferSize" value="100000"/>
</bean>
For FTPS connections, you can use o.s.i.ftp.session.DefaultFtpsSessionFactory
instead.
The following example shows a complete configuration:
<bean id="ftpClientFactory"
class="org.springframework.integration.ftp.session.DefaultFtpsSessionFactory">
<property name="host" value="localhost"/>
<property name="port" value="22"/>
<property name="username" value="oleg"/>
<property name="password" value="password"/>
<property name="clientMode" value="1"/>
<property name="fileType" value="2"/>
<property name="useClientMode" value="true"/>
<property name="cipherSuites" value="a,b.c"/>
<property name="keyManager" ref="keyManager"/>
<property name="protocol" value="SSL"/>
<property name="trustManager" ref="trustManager"/>
<property name="prot" value="P"/>
<property name="needClientAuth" value="true"/>
<property name="authValue" value="oleg"/>
<property name="sessionCreation" value="true"/>
<property name="protocols" value="SSL, TLS"/>
<property name="implicit" value="true"/>
</bean>
If you experience connectivity problems and would like to trace session creation as well as see which sessions are polled, you can enable session tracing by setting the logger to the TRACE level (for example, log4j.category.org.springframework.integration.file=TRACE ).
|
Now you need only inject these session factories into your adapters. The protocol (FTP or FTPS) that an adapter uses depends on the type of session factory that has been injected into the adapter.
| A more practical way to provide values for FTP or FTPS session factories is to use Spring’s property placeholder support (See docs.spring.io/spring/docs/current/spring-framework-reference/html/beans.html#beans-factory-placeholderconfigurer). |
18.2. Advanced Configuration
DefaultFtpSessionFactory
provides an abstraction over the underlying client API, which (since Spring Integration 2.0) is Apache Commons Net.
This spares you from the low-level configuration details of the org.apache.commons.net.ftp.FTPClient
.
Several common properties are exposed on the session factory (since version 4.0, this now includes connectTimeout
, defaultTimeout
, and dataTimeout
).
However, you sometimes need access to lower level FTPClient
configuration to achieve more advanced configuration (such as setting the port range for active mode).
For that purpose, AbstractFtpSessionFactory
(the base class for all FTP Session Factories) exposes hooks, in the form of the two post-processing methods shown in the following listing:
/**
* Will handle additional initialization after client.connect() method was invoked,
* but before any action on the client has been taken
*/
protected void postProcessClientAfterConnect(T t) throws IOException {
// NOOP
}
/**
* Will handle additional initialization before client.connect() method was invoked.
*/
protected void postProcessClientBeforeConnect(T client) throws IOException {
// NOOP
}
As you can see, there is no default implementation for these two methods.
However, by extending DefaultFtpSessionFactory
, you can override these methods to provide more advanced configuration of the FTPClient
, as the following example shows:
public class AdvancedFtpSessionFactory extends DefaultFtpSessionFactory {
protected void postProcessClientBeforeConnect(FTPClient ftpClient) throws IOException {
ftpClient.setActivePortRange(4000, 5000);
}
}
18.2.1. FTPS and Shared SSLSession
When using FTP over SSL or TLS, some servers require the same SSLSession
to be used on the control and data connections.
This is to prevent “stealing” data connections.
See scarybeastsecurity.blogspot.cz/2009/02/vsftpd-210-released.html for more information.
Currently, the Apache FTPSClient does not support this feature. See NET-408.
The following solution, courtesy of Stack Overflow, uses reflection on the sun.security.ssl.SSLSessionContextImpl
, so it may not work on other JVMs.
The stack overflow answer was submitted in 2015, and the solution has been tested by the Spring Integration team recently on JDK 1.8.0_112.
The following example shows how to create an FTPS session:
@Bean
public DefaultFtpsSessionFactory sf() {
DefaultFtpsSessionFactory sf = new DefaultFtpsSessionFactory() {
@Override
protected FTPSClient createClientInstance() {
return new SharedSSLFTPSClient();
}
};
sf.setHost("...");
sf.setPort(21);
sf.setUsername("...");
sf.setPassword("...");
sf.setNeedClientAuth(true);
return sf;
}
private static final class SharedSSLFTPSClient extends FTPSClient {
@Override
protected void _prepareDataSocket_(final Socket socket) throws IOException {
if (socket instanceof SSLSocket) {
// Control socket is SSL
final SSLSession session = ((SSLSocket) _socket_).getSession();
final SSLSessionContext context = session.getSessionContext();
context.setSessionCacheSize(0); // you might want to limit the cache
try {
final Field sessionHostPortCache = context.getClass()
.getDeclaredField("sessionHostPortCache");
sessionHostPortCache.setAccessible(true);
final Object cache = sessionHostPortCache.get(context);
final Method method = cache.getClass().getDeclaredMethod("put", Object.class,
Object.class);
method.setAccessible(true);
String key = String.format("%s:%s", socket.getInetAddress().getHostName(),
String.valueOf(socket.getPort())).toLowerCase(Locale.ROOT);
method.invoke(cache, key, session);
key = String.format("%s:%s", socket.getInetAddress().getHostAddress(),
String.valueOf(socket.getPort())).toLowerCase(Locale.ROOT);
method.invoke(cache, key, session);
}
catch (NoSuchFieldException e) {
// Not running in expected JRE
logger.warn("No field sessionHostPortCache in SSLSessionContext", e);
}
catch (Exception e) {
// Not running in expected JRE
logger.warn(e.getMessage());
}
}
}
}
18.3. Delegating Session Factory
Version 4.2 introduced the DelegatingSessionFactory
, which allows the selection of the actual session factory at runtime.
Prior to invoking the FTP endpoint, call setThreadKey()
on the factory to associate a key with the current thread.
That key is then used to lookup the actual session factory to be used.
You can clear the key by calling clearThreadKey()
after use.
We added convenience methods so that you can easily do use a delegating session factory from a message flow.
The following example shows how to declare a delegating session factory:
<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-ftp:outbound-gateway request-channel="c1" reply-channel="c2" ... />
<int:service-activator input-channel="c2" output-channel="out"
expression="@dsf.clearThreadKey(#root)" />
When you use session caching (see FTP Session Caching), each of the delegates should be cached.
You cannot cache the DelegatingSessionFactory itself.
|
Starting with version 5.0.7, the DelegatingSessionFactory
can be used in conjunction with a RotatingServerAdvice
to poll multiple servers; see Inbound Channel Adapters: Polling Multiple Servers and Directories.
18.4. FTP Inbound Channel Adapter
The FTP inbound channel adapter is a special listener that connects to the FTP server and listens for the remote directory events (for example, new file created) at which point it initiates a file transfer.
The following example shows how to configure an inbound-channel-adapter
:
<int-ftp:inbound-channel-adapter id="ftpInbound"
channel="ftpChannel"
session-factory="ftpSessionFactory"
auto-create-local-directory="true"
delete-remote-files="true"
filename-pattern="*.txt"
remote-directory="some/remote/path"
remote-file-separator="/"
preserve-timestamp="true"
local-filename-generator-expression="#this.toUpperCase() + '.a'"
scanner="myDirScanner"
local-filter="myFilter"
temporary-file-suffix=".writing"
max-fetch-size="-1"
local-directory=".">
<int:poller fixed-rate="1000"/>
</int-ftp:inbound-channel-adapter>
As the preceding configuration shows, you can configure an FTP inbound channel adapter by using the inbound-channel-adapter
element while also providing values for various attributes, such as local-directory
, filename-pattern
(which is based on simple pattern matching, not regular expressions), and the reference to a session-factory
.
By default, the transferred file carries the same name as the original file.
If you want to override this behavior, you can set the local-filename-generator-expression
attribute, which lets you provide a SpEL expression to generate the name of the local file.
Unlike outbound gateways and adapters, where the root object of the SpEL evaluation context is a Message
, this inbound adapter does not yet have the message at the time of evaluation, since that’s what it ultimately generates with the transferred file as its payload.
Consequently, the root object of the SpEL evaluation context is the original name of the remote file (a String
).
The inbound channel adapter first retrieves the File
object for a local directory and then emits each file according to the poller configuration.
Starting with version 5.0, you can now limit the number of files fetched from the FTP server when new file retrievals are needed.
This can be beneficial when the target files are very large or when you run in a clustered system with a persistent file list filter, discussed later.
Use max-fetch-size
for this purpose.
A negative value (the default) means no limit and all matching files are retrieved.
See Inbound Channel Adapters: Controlling Remote File Fetching for more information.
Since version 5.0, you can also provide a custom DirectoryScanner
implementation to the inbound-channel-adapter
by setting the scanner
attribute.
Starting with Spring Integration 3.0, you can specify the preserve-timestamp
attribute (its default is false
).
When true
, the local file’s modified timestamp is set to the value retrieved from the server.
Otherwise, it is set to the current time.
Starting with version 4.2, you can specify remote-directory-expression
instead of remote-directory
, letting you dynamically determine the directory on each poll — for example, remote-directory-expression="@myBean.determineRemoteDir()"
.
Starting with version 4.3, you can omit the remote-directory
and remote-directory-expression
attributes.
They default to null
.
In this case, according to the FTP protocol, the client working directory is used as the default remote directory.
Sometimes, file filtering based on the simple pattern specified with the filename-pattern
attribute might not suffice.
If this is the case, you can use the filename-regex
attribute to specify a regular expression (such as filename-regex=".*\.test$"
).
Also, if you need complete control, you can use the filter
attribute and provide a reference to any custom implementation of the o.s.i.file.filters.FileListFilter
, a strategy interface for filtering a list of files.
This filter determines which remote files are retrieved.
You can also combine a pattern-based filter with other filters (such as an AcceptOnceFileListFilter
to avoid synchronizing files that have previously been fetched) by using a CompositeFileListFilter
.
The AcceptOnceFileListFilter
stores its state in memory.
If you wish the state to survive a system restart, consider using the FtpPersistentAcceptOnceFileListFilter
instead.
This filter stores the accepted file names in an instance of the MetadataStore
strategy (see Metadata Store).
This filter matches on the filename and the remote modified time.
Since version 4.0, this filter requires a ConcurrentMetadataStore
.
When used with a shared data store (such as Redis
with the RedisMetadataStore
), it lets filter keys be shared across multiple application or server instances.
Starting with version 5.0, the FtpPersistentAcceptOnceFileListFilter
with in-memory SimpleMetadataStore
is applied by default for the FtpInboundFileSynchronizer
.
This filter is also applied with the regex
or pattern
option in the XML configuration as well as with FtpInboundChannelAdapterSpec
in the Java DSL.
Any other use cases can be managed with CompositeFileListFilter
(or ChainFileListFilter
).
The preceding discussion refers to filtering the files before retrieving them.
Once the files have been retrieved, an additional filter is applied to the files on the file system.
By default, this is an AcceptOnceFileListFilter
which, as discussed earlier, retains state in memory and does not consider the file’s modified time.
Unless your application removes files after processing, the adapter will re-process the files on disk by default after an application restart.
Also, if you configure the filter
to use a FtpPersistentAcceptOnceFileListFilter
and the remote file timestamp changes (causing it to be re-fetched), the default local filter does not let this new file be processed.
You can use the local-filter
attribute to configure the behavior of the local file system filter.
Starting with version 4.3.8, a FileSystemPersistentAcceptOnceFileListFilter
is configured by default.
This filter stores the accepted file names and modified timestamp in an instance of the MetadataStore
strategy (see Metadata Store) and detects changes to the local file modified time.
The default MetadataStore
is a SimpleMetadataStore
, which stores state in memory.
Since version 4.1.5, these filters have a new property (flushOnUpdate
) that causes them to flush the
metadata store on every update (if the store implements Flushable
).
Further, if you use a distributed MetadataStore (such as Redis or GemFire), you can have multiple instances of the same adapter or application and be sure that each file is processed only once.
|
The actual local filter is a CompositeFileListFilter
that contains the supplied filter and a pattern filter that prevents processing files that are in the process of being downloaded (based on the temporary-file-suffix
).
Files are downloaded with this suffix (the default is .writing
), and the file is renamed to its final name when the transfer is complete, making it 'visible' to the filter.
The remote-file-separator
attribute lets you configure a file separator character to use if the default '/' is not applicable for your particular environment.
See the schema for more details on these attributes.
You should also understand that the FTP inbound channel adapter is a polling consumer.
Therefore, you must configure a poller (by using either a global default or a local sub-element).
Once a file has been transferred, a message with a java.io.File
as its payload is generated and sent to the channel identified by the channel
attribute.
18.4.1. More on File Filtering and Incomplete Files
Sometimes the file that just appeared in the monitored (remote) directory is not complete.
Typically, such a file is written with a temporary extension (such as somefile.txt.writing
) and is then renamed once the writing process finishes.
In most cases, you are only interested in files that are complete and would like to filter for only files that are complete.
To handle these scenarios, you can use the filtering support provided by the filename-pattern
, filename-regex
, and filter
attributes.
The following example uses a custom filter implementation:
<int-ftp:inbound-channel-adapter
channel="ftpChannel"
session-factory="ftpSessionFactory"
filter="customFilter"
local-directory="file:/my_transfers">
remote-directory="some/remote/path"
<int:poller fixed-rate="1000"/>
</int-ftp:inbound-channel-adapter>
<bean id="customFilter" class="org.example.CustomFilter"/>
18.4.2. Poller Configuration Notes for the Inbound FTP Adapter
The job of the inbound FTP adapter consists of two tasks:
-
Communicate with a remote server in order to transfer files from a remote directory to a local directory.
-
For each transferred file, generate a message with that file as a payload and send it to the channel identified by the 'channel' attribute. That is why they are called "'channel adapters'" rather than just "'adapters'". The main job of such an adapter is to generate a message to send to a message channel. Essentially, the second task takes precedence in such a way that, if your local directory already has one or more files, it first generates messages from those. Only when all local files have been processed does it initiate the remote communication to retrieve more files.
Also, when configuring a trigger on the poller, you should pay close attention to the max-messages-per-poll
attribute.
Its default value is 1
for all SourcePollingChannelAdapter
instances (including FTP).
This means that, as soon as one file is processed, it waits for the next execution time as determined by your trigger configuration.
If you happened to have one or more files sitting in the local-directory
, it would process those files before it would initiate communication with the remote FTP server.
Also, if the max-messages-per-poll
is set to 1
(the default), it processes only one file at a time with intervals as defined by your trigger, essentially working as “one-poll === one-file”.
For typical file-transfer use cases, you most likely want the opposite behavior: to process all the files you can for each poll and only then wait for the next poll.
If that is the case, set max-messages-per-poll
to -1.
Then, on each poll, the adapter tries to generate as many messages as it possibly can.
In other words, it processes everything in the local directory, and then it connects to the remote directory to transfer everything that is available there to be processed locally.
Only then is the poll operation considered complete, and the poller waits for the next execution time.
You can alternatively set the 'max-messages-per-poll' value to a positive value that indicates the upward limit of messages to be created from files with each poll.
For example, a value of 10
means that, on each poll, it tries to process no more than ten files.
18.4.3. Recovering from Failures
It is important to understand the architecture of the adapter.
There is a file synchronizer that fetches the files and a FileReadingMessageSource
that emits a message for each
synchronized file.
As discussed earlier, two filters are involved.
The filter
attribute (and patterns) refers to the remote (FTP) file list, to avoid fetching files that have already
been fetched.
The local-filter
is used by the FileReadingMessageSource
to determine which files are to be sent as messages.
The synchronizer lists the remote files and consults its filter.
The files are then transferred.
If an IO error occurs during file transfer, any files that have already been added to the filter are removed so that they
are eligible to be re-fetched on the next poll.
This only applies if the filter implements ReversibleFileListFilter
(such as the AcceptOnceFileListFilter
).
If, after synchronizing the files, an error occurs on the downstream flow processing a file, no automatic rollback of the filter occurs, so the failed file is not reprocessed by default.
If you wish to reprocess such files after a failure, you can use configuration similar to the following to facilitate the removal of the failed file from the filter:
<int-ftp:inbound-channel-adapter id="ftpAdapter"
session-factory="ftpSessionFactory"
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-ftp: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" />
The preceding configuration works for any ResettableFileListFilter
.
Starting with version 5.0, the inbound channel adapter can build sub-directories locally that correspond to the generated local file name.
That can be a remote sub-path as well.
To be able to read a local directory recursively for modification according to the hierarchy support, you can now supply an internal FileReadingMessageSource
with a new RecursiveDirectoryScanner
based on the Files.walk()
algorithm.
See AbstractInboundFileSynchronizingMessageSource.setScanner()
for more information.
Also, you can now switch the AbstractInboundFileSynchronizingMessageSource
to the WatchService
-based DirectoryScanner
by using setUseWatchService()
option.
It is also configured for all the WatchEventType
instances to react to any modifications in local directory.
The reprocessing sample shown earlier is based on the built-in functionality of the FileReadingMessageSource.WatchServiceDirectoryScanner
to perform ResettableFileListFilter.remove()
when the file is deleted (StandardWatchEventKinds.ENTRY_DELETE
) from the local directory.
See WatchServiceDirectoryScanner
for more information.
18.4.4. Configuring with Java Configuration
The following Spring Boot application show an example of how to configure the inbound adapter with Java configuration:
@SpringBootApplication
public class FtpJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(FtpJavaApplication.class)
.web(false)
.run(args);
}
@Bean
public SessionFactory<FTPFile> ftpSessionFactory() {
DefaultFtpSessionFactory sf = new DefaultFtpSessionFactory();
sf.setHost("localhost");
sf.setPort(port);
sf.setUsername("foo");
sf.setPassword("foo");
sf.setTestSession(true);
return new CachingSessionFactory<FTPFile>(sf);
}
@Bean
public FtpInboundFileSynchronizer ftpInboundFileSynchronizer() {
FtpInboundFileSynchronizer fileSynchronizer = new FtpInboundFileSynchronizer(ftpSessionFactory());
fileSynchronizer.setDeleteRemoteFiles(false);
fileSynchronizer.setRemoteDirectory("foo");
fileSynchronizer.setFilter(new FtpSimplePatternFileListFilter("*.xml"));
return fileSynchronizer;
}
@Bean
@InboundChannelAdapter(channel = "ftpChannel", poller = @Poller(fixedDelay = "5000"))
public MessageSource<File> ftpMessageSource() {
FtpInboundFileSynchronizingMessageSource source =
new FtpInboundFileSynchronizingMessageSource(ftpInboundFileSynchronizer());
source.setLocalDirectory(new File("ftp-inbound"));
source.setAutoCreateLocalDirectory(true);
source.setLocalFilter(new AcceptOnceFileListFilter<File>());
source.setMaxFetchSize(1);
return source;
}
@Bean
@ServiceActivator(inputChannel = "ftpChannel")
public MessageHandler handler() {
return new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
System.out.println(message.getPayload());
}
};
}
}
18.4.5. Configuring with the Java DSL
The following Spring Boot application shows an example of how to configure the inbound adapter with the Java DSL:
@SpringBootApplication
public class FtpJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(FtpJavaApplication.class)
.web(false)
.run(args);
}
@Bean
public IntegrationFlow ftpInboundFlow() {
return IntegrationFlows
.from(Fpt.inboundAdapter(this.ftpSessionFactory)
.preserveTimestamp(true)
.remoteDirectory("foo")
.regexFilter(".*\\.txt$")
.localFilename(f -> f.toUpperCase() + ".a")
.localDirectory(new File("d:\\ftp_files")),
e -> e.id("ftpInboundAdapter")
.autoStartup(true)
.poller(Pollers.fixedDelay(5000)))
.handle(m -> System.out.println(m.getPayload()))
.get();
}
}
18.4.6. Dealing With Incomplete Data
The FtpSystemMarkerFilePresentFileListFilter
is provided to filter remote files that do not have a corresponding marker file on the remote system.
See the Javadoc (and browse to the parent classes) for configuration information.
18.5. FTP Streaming Inbound Channel Adapter
Version 4.3 introduced the streaming inbound channel adapter.
This adapter produces message with payloads of type InputStream
, letting files be fetched without writing to the
local file system.
Since the session remains open, the consuming application is responsible for closing the session when the file has been
consumed.
The session is provided in the closeableResource
header (IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE
).
Standard framework components, such as the FileSplitter
and StreamTransformer
, automatically close the session.
See File Splitter and Stream Transformer for more information about these components.
The follwoing example shows how to configure an inbound-streaming-channel-adapter
:
<int-ftp: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-ftp:inbound-streaming-channel-adapter>
Only one of filename-pattern
, filename-regex
, filter
, or filter-expression
is allowed.
Starting with version 5.0, by default, the FtpStreamingMessageSource adapter prevents duplicates for remote files with FtpPersistentAcceptOnceFileListFilter based on the in-memory SimpleMetadataStore .
By default, this filter is also applied with the filename pattern (or regex).
If you need to allow duplicates, you can use AcceptAllFileListFilter .
Any other use cases can be handled by CompositeFileListFilter (or ChainFileListFilter ).
The Java configuration (later in the document) shows one technique to remove the remote file after processing to avoid duplicates.
|
Use the max-fetch-size
attribute to limit the number of files fetched on each poll when a fetch is necessary.
Set it to 1
and use a persistent filter when running in a clustered environment.
See Inbound Channel Adapters: Controlling Remote File Fetching for more information.
The adapter puts the remote directory and file name in the FileHeaders.REMOTE_DIRECTORY
and FileHeaders.REMOTE_FILE
headers, respectively.
Starting with version 5.0, the FileHeaders.REMOTE_FILE_INFO
header provides additional remote file information (represented in JSON by default).
If you set the fileInfoJson
property on the FtpStreamingMessageSource
to false
, the header contains an FtpFileInfo
object.
The FTPFile
object provided by the underlying Apache Net library can be accessed by using the FtpFileInfo.getFileInfo()
method.
The fileInfoJson
property is not available when you use XML configuration, but you can set it by injecting the FtpStreamingMessageSource
into one of your configuration classes.
Starting with version 5.1, the generic type of the comparator
is FTPFile
.
Previously, it was AbstractFileInfo<FTPFile>
.
This is because the sort is now performed earlier in the processing, before filtering and applying maxFetch
.
18.5.1. Configuring with Java Configuration
The following Spring Boot application shows an example of how to configure the inbound adapter with Java configuration:
@SpringBootApplication
public class FtpJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(FtpJavaApplication.class)
.web(false)
.run(args);
}
@Bean
@InboundChannelAdapter(channel = "stream")
public MessageSource<InputStream> ftpMessageSource() {
FtpStreamingMessageSource messageSource = new FtpStreamingMessageSource(template());
messageSource.setRemoteDirectory("ftpSource/");
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 FtpRemoteFileTemplate template() {
return new FtpRemoteFileTemplate(ftpSessionFactory());
}
@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;
}
}
Notice that, in this example, the message handler downstream of the transformer has an advice that removes the remote file after processing.
18.6. Inbound Channel Adapters: Polling Multiple Servers and Directories
Starting with version 5.0.7, the RotatingServerAdvice
is available; when configured as a poller advice, the inbound adapters can poll multiple servers and directories.
Configure the advice and add it to the poller’s advice chain as normal.
A DelegatingSessionFactory
is used to select the server see Delegating Session Factory for more information.
The advice configuration consists of a list of RotatingServerAdvice.KeyDirectory
objects.
@Bean
public RotatingServerAdvice advice() {
List<KeyDirectory> keyDirectories = new ArrayList<>();
keyDirectories.add(new KeyDirectory("one", "foo"));
keyDirectories.add(new KeyDirectory("one", "bar"));
keyDirectories.add(new KeyDirectory("two", "baz"));
keyDirectories.add(new KeyDirectory("two", "qux"));
keyDirectories.add(new KeyDirectory("three", "fiz"));
keyDirectories.add(new KeyDirectory("three", "buz"));
return new RotatingServerAdvice(delegatingSf(), keyDirectories);
}
This advice will poll directory foo
on server one
until no new files exist then move to directory bar
and then directory baz
on server two
, etc.
This default behavior can be modified with the fair
constructor arg:
@Bean
public RotatingServerAdvice advice() {
...
return new RotatingServerAdvice(delegatingSf(), keyDirectories, true);
}
In this case, the advice will move to the next server/directory regardless of whether the previous poll returned a file.
Alternatively, you can provide your own RotatingServerAdvice.RotationPolicy
to reconfigure the message source as needed:
public interface RotationPolicy {
void beforeReceive(MessageSource<?> source);
void afterReceive(boolean messageReceived, MessageSource<?> source);
}
and
@Bean
public RotatingServerAdvice advice() {
return new RotatingServerAdvice(myRotationPolicy());
}
The local-filename-generator-expression
attribute (localFilenameGeneratorExpression
on the synchronizer) can now contain the #remoteDirectory
variable.
This allows files retrieved from different directories to be downloaded to similar directories locally:
@Bean
public IntegrationFlow flow() {
return IntegrationFlows.from(Ftp.inboundAdapter(sf())
.filter(new FtpPersistentAcceptOnceFileListFilter(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();
}
Do not configure a TaskExecutor on the poller when using this advice; see Conditional Pollers for Message Sources for more information.
|
18.7. Inbound Channel Adapters: Controlling Remote File Fetching
There are two properties that you should consider when you configure inbound channel adapters.
max-messages-per-poll
, as with all pollers, can be used to limit the number of messages emitted on each poll (if more than the configured value are ready).
max-fetch-size
(since version 5.0) can limit the number of files retrieved from the remote server at one time.
The following scenarios assume the starting state is an empty local directory:
-
max-messages-per-poll=2
andmax-fetch-size=1
: The adapter fetches one file, emits it, fetches the next file, emits it, and then sleeps until the next poll. -
max-messages-per-poll=2
andmax-fetch-size=2
): The adapter fetches both files and then emits each one. -
max-messages-per-poll=2
andmax-fetch-size=4
: The adapter fetches up to four files (if available) and emits the first two (if there are at least two). The next two files are emitted on the next poll. -
max-messages-per-poll=2
andmax-fetch-size
not specified: The adapter fetches all remote files and emits the first two (if there are at least two). The subsequent files are emitted on subsequent polls (two at a time). When all files are consumed, the remote fetch is attempted again, to pick up any new files.
When you deploy multiple instances of an application, we recommend a small max-fetch-size , to avoid one instance “grabbing” all the files and starving other instances.
|
Another use for max-fetch-size
is if you want to stop fetching remote files but continue to process files that have already been fetched.
Setting the maxFetchSize
property on the MessageSource
(programmatically, with JMX, or with a control bus) effectively stops the adapter from fetching more files but lets the poller continue to emit messages for files that have previously been fetched.
If the poller is active when the property is changed, the change takes effect on the next poll.
Starting with version 5.1, the synchronizer can be provided with a Comparator<FTPFile>
.
This is useful when restricting the number of files fetched with maxFetchSize
.
18.8. FTP Outbound Channel Adapter
The FTP outbound channel adapter relies on a MessageHandler
implementation that connects to the FTP server and initiates an FTP transfer for every file it receives in the payload of incoming messages.
It also supports several representations of a file, so you are not limited only to java.io.File
-typed payloads.
The FTP outbound channel adapter supports the following payloads:
-
java.io.File
: The actual file object -
byte[]
: A byte array that represents the file contents -
java.lang.String
: Text that represents the file contents.
The following example shows how to configure an outbound-channel-adapter
:
<int-ftp:outbound-channel-adapter id="ftpOutbound"
channel="ftpChannel"
session-factory="ftpSessionFactory"
charset="UTF-8"
remote-file-separator="/"
auto-create-directory="true"
remote-directory-expression="headers['remote_dir']"
temporary-remote-directory-expression="headers['temp_remote_dir']"
filename-generator="fileNameGenerator"
use-temporary-filename="true"
mode="REPLACE"/>
The preceding configuration shows how you can configure an FTP outbound channel adapter by using the outbound-channel-adapter
element while also providing values for various attributes, such as filename-generator
(an implementation of the o.s.i.file.FileNameGenerator
strategy interface), a reference to a session-factory
, and other attributes.
You can also see some examples of *expression
attributes that let you use SpEL to configure settings such as remote-directory-expression
, temporary-remote-directory-expression
, and remote-filename-generator-expression
(a SpEL alternative to filename-generator
, shown in the preceding example).
As with any component that allows the usage of SpEL, access to the payload and the message Headers is available through the 'payload' and 'headers' variables.
See the schema for more details on the available attributes.
By default, if no file name generator is specified, Spring Integration uses o.s.i.file.DefaultFileNameGenerator .
DefaultFileNameGenerator determines the file name based on the value of the file_name header (if it exists) in the MessageHeaders , or, if the payload of the Message is already a java.io.File , it uses the original name of that file.
|
Defining certain values (such as remote-directory ) might be platform- or FTP server-dependent.
For example, as was reported on forum.spring.io/showthread.php?p=333478&posted=1#post333478, on some platforms, you must add a slash to the end of the directory definition (for example, remote-directory="/thing1/thing2/" instead of remote-directory="/thing1/thing2" ).
|
Starting with version 4.1, you can specify the mode
when transferring the file.
By default, an existing file is overwritten.
The modes are defined by the FileExistsMode
enumeration, which includes the following values:
-
REPLACE
(default) *APPEND
-
IGNORE
-
FAIL
IGNORE
and FAIL
do not transfer the file.
FAIL
causes an exception to be thrown, while IGNORE
silently ignores the transfer (although a DEBUG
log entry is produced).
18.8.1. Avoiding Partially Written Files
One of the common problems that arises when dealing with file transfers is the possibility of processing a partial file. That is, a file might appear in the file system before its transfer is actually complete.
To deal with this issue, Spring Integration FTP adapters use a common algorithm: Files are transferred under a temporary name and then renamed once they are fully transferred.
By default, every file that is in the process of being transferred appears in the file system with an additional suffix, which, by default, is .writing
.
You can change this suffix by setting the temporary-file-suffix
attribute.
However, there may be situations where you do not want to use this technique (for example, if the server does not permit renaming files).
For situations like this, you can disable this feature by setting use-temporary-file-name
to false
(the default is true
).
When this attribute is false
, the file is written with its final name and the consuming application needs some other mechanism to detect that the file is completely uploaded before accessing it.
18.8.2. Configuring with Java Configuration
The following Spring Boot application shows an example of how to configure the outbound adapter with Java configuration:
@SpringBootApplication
@IntegrationComponentScan
public class FtpJavaApplication {
public static void main(String[] args) {
ConfigurableApplicationContext context =
new SpringApplicationBuilder(FtpJavaApplication.class)
.web(false)
.run(args);
MyGateway gateway = context.getBean(MyGateway.class);
gateway.sendToFtp(new File("/foo/bar.txt"));
}
@Bean
public SessionFactory<FTPFile> ftpSessionFactory() {
DefaultFtpSessionFactory sf = new DefaultFtpSessionFactory();
sf.setHost("localhost");
sf.setPort(port);
sf.setUsername("foo");
sf.setPassword("foo");
sf.setTestSession(true);
return new CachingSessionFactory<FTPFile>(sf);
}
@Bean
@ServiceActivator(inputChannel = "ftpChannel")
public MessageHandler handler() {
FtpMessageHandler handler = new FtpMessageHandler(ftpSessionFactory());
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 = "toFtpChannel")
void sendToFtp(File file);
}
}
18.8.3. Configuring with the Java DSL
The following Spring Boot application shows an example of how to configure the outbound adapter using the Java DSL:
@SpringBootApplication
@IntegrationComponentScan
public class FtpJavaApplication {
public static void main(String[] args) {
ConfigurableApplicationContext context =
new SpringApplicationBuilder(FtpJavaApplication.class)
.web(false)
.run(args);
MyGateway gateway = context.getBean(MyGateway.class);
gateway.sendToFtp(new File("/foo/bar.txt"));
}
@Bean
public SessionFactory<FTPFile> ftpSessionFactory() {
DefaultFtpSessionFactory sf = new DefaultFtpSessionFactory();
sf.setHost("localhost");
sf.setPort(port);
sf.setUsername("foo");
sf.setPassword("foo");
sf.setTestSession(true);
return new CachingSessionFactory<FTPFile>(sf);
}
@Bean
public IntegrationFlow ftpOutboundFlow() {
return IntegrationFlows.from("toFtpChannel")
.handle(Ftp.outboundAdapter(ftpSessionFactory(), FileExistsMode.FAIL)
.useTemporaryFileName(false)
.fileNameExpression("headers['" + FileHeaders.FILENAME + "']")
.remoteDirectory(this.ftpServer.getTargetFtpDirectory().getName())
).get();
}
@MessagingGateway
public interface MyGateway {
@Gateway(requestChannel = "toFtpChannel")
void sendToFtp(File file);
}
}
18.9. FTP Outbound Gateway
The FTP outbound gateway provides a limited set of commands to interact with a remote FTP or FTPS server. The supported commands are:
-
ls
(list files) -
nlst
(list file names) -
get
(retrieve file) -
mget
(retrieve file(s)) -
rm
(remove file(s)) -
mv
(move/rename file) -
put
(send file) -
mput
(send multiple files)
18.9.1. Using the ls
Command
ls
lists remote files and supports the following options:
-
-1
: Retrieve a list of file names. The default is to retrieve a list ofFileInfo
objects. -
-a
: Include all files (including those starting with '.') -
-f
: Do not sort the list -
-dirs
: Include directories (they are excluded by default) -
-links
: Include symbolic links (they are excluded by default) -
-R
: List the remote directory recursively
In addition, filename filtering is provided, in the same manner as the inbound-channel-adapter
.
See FTP Inbound Channel Adapter.
The message payload resulting from an ls
operation is a list of file names or a list of FileInfo
objects.
These objects provide information such as modified time, permissions, and other details.
The remote directory that the ls
command acted on is provided in the file_remoteDirectory
header.
When using the recursive option (-R
), the fileName
includes any subdirectory elements, representing a relative path to the file (relative to the remote directory).
If the -dirs
option is included, each recursive directory is also returned as an element in the list.
In this case, it is recommended that you not use the -1
option, because you would not be able to distinguish files from directories, which you can do with the FileInfo
objects.
Starting with version 4.3, the FtpSession
supports null
for the list()
and listNames()
methods.
Therefore, you can omit the expression
attribute.
For convenience, Java configuration has two constructors that do not have an expression
argument.
or LS
, NLST
, PUT
and MPUT
commands, null
is treated as the client working directory, according to the FTP protocol.
All other commands must be supplied with the expression
to evaluate the remote path against the request message.
You can set the working directory with the FTPClient.changeWorkingDirectory()
function when you extend the DefaultFtpSessionFactory
and implement the postProcessClientAfterConnect()
callback.
18.9.2. Using the nlst
Command
Version 5 introduced support for the nlst
command.
nlst
lists remote file names and supports only one option:
-
-f
: Do not sort the list
The message payload resulting from an nlst
operation is a list of file names.
The remote directory that the nlst
command acted on is provided in the file_remoteDirectory
header.
Unlike the -1
option for the ls
command, which uses the LIST
command, the nlst
command sends an NLST
command to the target FTP server.
This command is useful when the server does not support LIST
(due to security restrictions, for example).
The result of the nlst
operation is the names without other detail.
Therefore, the framework cannot determine if an entity is a directory, to perform filtering or recursive listing, for example.
18.9.3. Using the get
Command
get
retrieves a remote file.
It supports the following option:
-
-P
: Preserve the timestamp of the remote file. -
-stream
: Retrieve the remote file as a stream. -
-D
: Delete the remote file after successful transfer. The remote file is not deleted if the transfer is ignored, because theFileExistsMode
isIGNORE
and the local file already exists.
The file_remoteDirectory
header provides the remote directory name, and the file_remoteFile
header provides the file name.
The message payload resulting from a get
operation is a File
object that represents the retrieved file or an InputStream
when you use the -stream
option.
The -stream
option allows retrieving the file as a stream.
For text files, a common use case is to combine this operation with a file splitter or a stream transformer.
When consuming remote files as streams, you are responsible for closing the Session
after the stream is consumed.
For convenience, the Session
is provided in the closeableResource
header, which you can access with a convenience method on IntegrationMessageHeaderAccessor
The following example shows how to do use the covenience method:
Closeable closeable = new IntegrationMessageHeaderAccessor(message).getCloseableResource();
if (closeable != null) {
closeable.close();
}
Framework components such as the file splitter and the stream transformer automatically close the session after the data is transferred.
The following example shows how to consume a file as a stream:
<int-ftp: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" />
If you consume the input stream in a custom component, you must close the Session .
You can do so either in your custom code or by routing a copy of the message to a service-activator and using SpEL, as the following example shows:
|
<int:service-activator input-channel="closeSession"
expression="headers['closeableResource'].close()" />
18.9.4. Using the mget
Command
mget
retrieves multiple remote files based on a pattern and supports the following options:
-
-P
: Preserve the timestamps of the remote files. -
-R
: Retrieve the entire directory tree recursively. -
-x
: Throw an exception if no files match the pattern (otherwise an empty list is returned). -
-D
: Delete each remote file after successful transfer. The remote file is not deleted if the transfer is ignored, because theFileExistsMode
isIGNORE
and the local file already exists.
The message payload resulting from an mget
operation is a List<File>
object (that is, a List
of File
objects, each representing a retrieved file).
Starting with version 5.0, if the FileExistsMode is IGNORE , the payload of the output message no longer contains files that were not fetched due to the file already existing.
Previously, the list contained all files, including those that already existed.
|
The expression used to determine the remote path should produce a result that ends with - e.g.
somedir/
will fetch the complete tree under somedir
.
Starting with version 5.0, a recursive mget
, combined with the new FileExistsMode.REPLACE_IF_MODIFIED
mode, can be used to periodically synchronize an entire remote directory tree locally.
This mode replaces the local file’s last modified timestamp with the remote timestamp, regardless of the -P
(preserve timestamp) option.
|
Using recursion (
-R )The pattern is ignored, and If a subdirectory is filtered, no additional traversal of that subdirectory is performed. The Typically, you would use the |
Starting with version 5.0, the FtpSimplePatternFileListFilter
and FtpRegexPatternFileListFilter
can be configured to always pass directories by setting the alwaysAcceptDirectories
property to true
.
Doing so allows recursion for a simple pattern, as the following examples show:
<bean id="starDotTxtFilter"
class="org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter">
<constructor-arg value="*.txt" />
<property name="alwaysAcceptDirectories" value="true" />
</bean>
<bean id="dotStarDotTxtFilter"
class="org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter">
<constructor-arg value="^.*\.txt$" />
<property name="alwaysAcceptDirectories" value="true" />
</bean>
Once you have defined filters such as those in the preceding example, you can use one by setting the filter
property on the gateway.
18.9.5. Using the put
Command
The put
commad sends a file to the remote server.
The payload of the message can be a java.io.File
, a byte[]
, or a String
.
A remote-filename-generator
(or expression) is used to name the remote file.
Other available attributes include remote-directory
, temporary-remote-directory
, and their *-expression
equivalents: use-temporary-file-name
and auto-create-directory
.
See the schema documentation for more information.
The message payload resulting from a put
operation is a String
that represents the full path of the file on the server after transfer.
Using the mput
Command
The mput
sends multiple files to the server and supports only one option:
-
-R
: Recursive. Send all files (possibly filtered) in the directory and its subdirectories.
The message payload must be a java.io.File
(or String
) that represents a local directory.
Since version 5.1, a collection of File
or String
is also supported.
This command supports the same attributes as the put
command.
In addition, files in the local directory can be filtered with one of mput-pattern
, mput-regex
, mput-filter
, or mput-filter-expression
.
The filter works with recursion, as long as the subdirectories themselves pass the filter.
Subdirectories that do not pass the filter are not recursed.
The message payload resulting from an mget
operation is a List<String>
object (that is, a List
of remote file paths that result from the transfer).
18.9.6. Using the rm
Command
The rm
command removes files.
The rm
command has no options.
The message payload resulting from an rm
operation is Boolean.TRUE
if the remove was successful or Boolean.FALSE
otherwise.
The file_remoteDirectory
header provides the remote directory, and the file_remoteFile
header provides the file name.
18.9.7. Using the mv
Command
The mv
command moves files
The mv
command has no options.
The expression
attribute defines the “from” path and the rename-expression
attribute defines the “to” path.
By default, the rename-expression
is headers['file_renameTo']
.
This expression must not evaluate to null or an empty String
.
If necessary, any necessary remote directories are created.
The payload of the result message is Boolean.TRUE
.
The file_remoteDirectory
header provides the original remote directory, and file_remoteFile
header provides the file name.
The new path is in the file_renameTo
header.
18.9.8. Additional Information about FTP Outbound Gateway Commands
The get
and mget
commands support the local-filename-generator-expression
attribute.
It defines a SpEL expression to generate the name of local files during the transfer.
The root object of the evaluation context is the request message.
The remoteFileName
variable, which is particularly useful for mget