<bean id="clientFactory"
class="org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory">
<property name="connectionOptions">
<bean class="org.eclipse.paho.client.mqttv3.MqttConnectOptions">
<property name="userName" value="${mqtt.username}"/>
<property name="password" value="${mqtt.password}"/>
</bean>
</property>
</bean>
<int-mqtt:message-driven-channel-adapter id="mqttInbound"
client-id="${mqtt.default.client.id}.src"
url="${mqtt.url}"
topics="sometopic"
client-factory="clientFactory"
channel="output"/>
The following listing shows the available attributes:
<int-mqtt:message-driven-channel-adapter id="oneTopicAdapter"
client-id="foo" (1)
url="tcp://localhost:1883" (2)
topics="bar,baz" (3)
qos="1,2" (4)
converter="myConverter" (5)
client-factory="clientFactory" (6)
send-timeout="123" (7)
error-channel="errors" (8)
recovery-interval="10000" (9)
channel="out" />
| 1 | The client ID. |
| 2 | The broker URL. |
| 3 | A comma-separated list of topics from which this adapter receives messages. |
| 4 | A comma-separated list of QoS values. It can be a single value that is applied to all topics or a value for each topic (in which case, the lists must be the same length). |
| 5 | An MqttMessageConverter (optional).
By default, the default DefaultPahoMessageConverter produces a message with a String payload with the following headers:
|
| 6 | The client factory. |
| 7 | The send timeout.
It applies only if the channel might block (such as a bounded QueueChannel that is currently full). |
| 8 | The error channel.
Downstream exceptions are sent to this channel, if supplied, in an ErrorMessage .
The payload is a MessagingException that contains the failed message and cause. |
| 9 | The recovery interval.
It controls the interval at which the adapter attempts to reconnect after a failure.
It defaults to 10000ms (ten seconds). |
Starting with version 4.1, you can omit the URL.
Instead, you can provide the server URIs in the serverURIs property of the DefaultMqttPahoClientFactory .
Doing so enables, for example, connection to a highly available (HA) cluster.
|
Starting with version 4.2.2, an MqttSubscribedEvent
is published when the adapter successfully subscribes to the topics.
MqttConnectionFailedEvent
events are published when the connection or subscription fails.
These events can be received by a bean that implements ApplicationListener
.
Also, a new property called recoveryInterval
controls the interval at which the adapter attempts to reconnect after a failure.
It defaults to 10000ms
(ten seconds).
|
Prior to version 4.2.3, the client always unsubscribed when the adapter was stopped.
This was incorrect because, if the client QOS is greater than 0, we need to keep the subscription active so that messages arriving
while the adapter is stopped are delivered on the next start.
This also requires setting the Starting with version 4.2.3, the adapter does not unsubscribe (by default) if the This behavior can be overridden by setting the To revert to the pre-4.2.3 behavior, use |
|
Starting with version 5.0, the |
26.1.1. Adding and Removing Topics at Runtime
Starting with version 4.1, you can programmatically change the topics to which the adapter is subscribed.
Spring Integration provides the addTopic()
and removeTopic()
methods.
When adding topics, you can optionally specify the QoS
(default: 1).
You can also modify the topics by sending an appropriate message to a <control-bus/>
with an appropriate payload — for example: "myMqttAdapter.addTopic('foo', 1)"
.
Stopping and starting the adapter has no effect on the topic list (it does not revert to the original settings in the configuration). The changes are not retained beyond the life cycle of the application context. A new application context reverts to the configured settings.
Changing the topics while the adapter is stopped (or disconnected from the broker) takes effect the next time a connection is established.
26.1.2. 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 MqttJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(MqttJavaApplication.class)
.web(false)
.run(args);
}
@Bean
public MessageChannel mqttInputChannel() {
return new DirectChannel();
}
@Bean
public MessageProducer inbound() {
MqttPahoMessageDrivenChannelAdapter adapter =
new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883", "testClient",
"topic1", "topic2");
adapter.setCompletionTimeout(5000);
adapter.setConverter(new DefaultPahoMessageConverter());
adapter.setQos(1);
adapter.setOutputChannel(mqttInputChannel());
return adapter;
}
@Bean
@ServiceActivator(inputChannel = "mqttInputChannel")
public MessageHandler handler() {
return new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
System.out.println(message.getPayload());
}
};
}
}
26.1.3. Configuring with the Java DSL
The following Spring Boot application provides an example of configuring the inbound adapter with the Java DSL:
@SpringBootApplication
public class MqttJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(MqttJavaApplication.class)
.web(false)
.run(args);
}
@Bean
public IntegrationFlow mqttInbound() {
return IntegrationFlows.from(
new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883",
"testClient", "topic1", "topic2");)
.handle(m -> System.out.println(m.getPayload()))
.get();
}
}
26.2. Outbound Channel Adapter
The outbound channel adapter is implemented by the MqttPahoMessageHandler
, which is wrapped in a ConsumerEndpoint
.
For convenience, you can configure it by using the namespace.
Starting with version 4.1, the adapter supports asynchronous send operations, avoiding blocking until the delivery is confirmed. You can emit application events to enable applications to confirm delivery if desired.
The following listing shows the attributes available for an outbound channel adapter:
<int-mqtt:outbound-channel-adapter id="withConverter"
client-id="foo" (1)
url="tcp://localhost:1883" (2)
converter="myConverter" (3)
client-factory="clientFactory" (4)
default-qos="1" (5)
qos-expression="" (6)
default-retained="true" (7)
retained-expression="" (8)
default-topic="bar" (9)
topic-expression="" (10)
async="false" (11)
async-events="false" (12)
channel="target" />
| 1 | The client ID. |
| 2 | The broker URL. |
| 3 | An MqttMessageConverter (optional).
The default DefaultPahoMessageConverter recognizes the following headers:
|
| 4 | The client factory. |
| 5 | The default quality of service.
It is used if no mqtt_qos header is found or the qos-expression returns null .
It is not used if you supply a custom converter . |
| 6 | An expression to evaluate to determine the qos.
The default is headers[mqtt_qos] . |
| 7 | The default value of the retained flag.
It is used if no mqtt_retained header is found.
It is not used if a custom converter is supplied. |
| 8 | An expression to evaluate to determine the retained boolean.
The default is headers[mqtt_retained] . |
| 9 | The default topic to which the message is sent (used if no mqtt_topic header is found). |
| 10 | An expression to evaluate to determine the destination topic.
The default is headers['topic'] . |
| 11 | When true , the caller does not block.
Rather, it waits for delivery confirmation when a message is sent.
The default is false (the send blocks until delivery is confirmed). |
| 12 | When async and async-events are both true , an MqttMessageSentEvent is emitted.
It contains the message, the topic, the messageId generated by the client library, the clientId , and the clientInstance (incremented each time the client is connected).
When the delivery is confirmed by the client library, an MqttMessageDeliveredEvent is emitted.
It contains the the messageId , the clientId , and the clientInstance , enabling delivery to be correlated with the send.
Any ApplicationListener or an event inbound channel adapter can received these events.
Note that it is possible for the MqttMessageDeliveredEvent to be received before the MqttMessageSentEvent .
The default is false . |
Starting with version 4.1, the URL can be omitted.
Instead, the server URIs can be provided in the serverURIs property of the DefaultMqttPahoClientFactory .
This enables, for example, connection to a highly available (HA) cluster.
|
26.2.1. Configuring with Java Configuration
The following Spring Boot application show an example of how to configure the outbound adapter with Java configuration:
@SpringBootApplication
@IntegrationComponentScan
public class MqttJavaApplication {
public static void main(String[] args) {
ConfigurableApplicationContext context =
new SpringApplicationBuilder(MqttJavaApplication.class)
.web(false)
.run(args);
MyGateway gateway = context.getBean(MyGateway.class);
gateway.sendToMqtt("foo");
}
@Bean
public MqttPahoClientFactory mqttClientFactory() {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
MqttConnectOptions options = new MqttConnectOptions();
options.setServerURIs(new String[] { "tcp://host1:1883", "tcp://host2:1883" });
options.setUserName("username");
options.setPassword("password".toCharArray());
factory.setConnectionOptions(options);
return factory;
}
@Bean
@ServiceActivator(inputChannel = "mqttOutboundChannel")
public MessageHandler mqttOutbound() {
MqttPahoMessageHandler messageHandler =
new MqttPahoMessageHandler("testClient", mqttClientFactory());
messageHandler.setAsync(true);
messageHandler.setDefaultTopic("testTopic");
return messageHandler;
}
@Bean
public MessageChannel mqttOutboundChannel() {
return new DirectChannel();
}
@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MyGateway {
void sendToMqtt(String data);
}
}
26.2.2. Configuring with the Java DSL
The following Spring Boot application provides an example of configuring the outbound adapter with the Java DSL:
@SpringBootApplication
public class MqttJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(MqttJavaApplication.class)
.web(false)
.run(args);
}
@Bean
public IntegrationFlow mqttOutboundFlow() {
return f -> f.handle(new MqttPahoMessageHandler("tcp://host1:1883", "someMqttClient"));
}
}
27. Redis Support
Spring Integration 2.1 introduced support for Redis: “an open source advanced key-value store”.
This support comes in the form of a Redis-based MessageStore
as well as publish-subscribe messaging adapters that are supported by Redis through its PUBLISH
, SUBSCRIBE
, and UNSUBSCRIBE
commands.
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-redis</artifactId>
<version>5.1.5.RELEASE</version>
</dependency>
compile "org.springframework.integration:spring-integration-redis:5.1.5.RELEASE"
You also need to include Redis client dependency, e.g. Lettuce.
To download, install, and run Redis, see the Redis documentation.
27.1. Connecting to Redis
To begin interacting with Redis, you first need to connect to it.
Spring Integration uses support provided by another Spring project, Spring Data Redis, which provides typical Spring constructs: ConnectionFactory
and Template
.
Those abstractions simplify integration with several Redis client Java APIs.
Currently Spring Data Redis supports Jedis and Lettuce.
27.1.1. Using RedisConnectionFactory
To connect to Redis, you can use one of the implementations of the RedisConnectionFactory
interface.
The following listing shows the interface definition:
public interface RedisConnectionFactory extends PersistenceExceptionTranslator {
/**
* Provides a suitable connection for interacting with Redis.
* @return connection for interacting with Redis.
*/
RedisConnection getConnection();
}
The following example shows how to create a LettuceConnectionFactory
in Java:
LettuceConnectionFactory cf = new LettuceConnectionFactory();
cf.afterPropertiesSet();
The following example shows how to create a LettuceConnectionFactory
in Spring’s XML configuration:
<bean id="redisConnectionFactory"
class="o.s.data.redis.connection.lettuce.LettuceConnectionFactory">
<property name="port" value="7379" />
</bean>
The implementations of RedisConnectionFactory
provide a set of properties, such as port and host, that you can set if needed.
Once you have an instance of RedisConnectionFactory
, you can create an instance of RedisTemplate
and inject it with the RedisConnectionFactory
.
27.1.2. Using RedisTemplate
As with other template classes in Spring (such as JdbcTemplate
and JmsTemplate
) RedisTemplate
is a helper class that simplifies Redis data access code.
For more information about RedisTemplate
and its variations (such as StringRedisTemplate
) see the Spring Data Redis documentation.
The following example shows how to create an instance of RedisTemplate
in Java:
RedisTemplate rt = new RedisTemplate<String, Object>();
rt.setConnectionFactory(redisConnectionFactory);
The following example shows how to create an instance of RedisTemplate
in Spring’s XML configuration:
<bean id="redisTemplate"
class="org.springframework.data.redis.core.RedisTemplate">
<property name="connectionFactory" ref="redisConnectionFactory"/>
</bean>
27.2. Messaging with Redis
As mentioned in the introduction, Redis provides support for publish-subscribe messaging through its PUBLISH
, SUBSCRIBE
, and UNSUBSCRIBE
commands.
As with JMS and AMQP, Spring Integration provides message channels and adapters for sending and receiving messages through Redis.
27.2.1. Redis Publish/Subscribe channel
Similarly to JMS, there are cases where both the producer and consumer are intended to be part of the same application, running within the same process. You can accomplished this by using a pair of inbound and outbound channel adapters. However, as with Spring Integration’s JMS support, there is a simpler way to address this use case. You can create a publish-subscribe channel, as the following example shows:
<int-redis:publish-subscribe-channel id="redisChannel" topic-name="si.test.topic"/>
A publish-subscribe-channel
behaves much like a normal <publish-subscribe-channel/>
element from the main Spring Integration namespace.
It can be referenced by both the input-channel
and the output-channel
attributes of any endpoint.
The difference is that this channel is backed by a Redis topic name: a String
value specified by the topic-name
attribute.
However, unlike JMS, this topic does not have to be created in advance or even auto-created by Redis.
In Redis, topics are simple String
values that play the role of an address.
The producer and consumer can communicate by using the same String
value as their topic name.
A simple subscription to this channel means that asynchronous publish-subscribe messaging is possible between the producing and consuming endpoints.
However, unlike the asynchronous message channels created by adding a <queue/>
element within a simple Spring Integration <channel/>
element, the messages are not stored in an in-memory queue.
Instead, those messages are passed through Redis, which lets you rely on its support for persistence and clustering as well as its interoperability with other non-Java platforms.
27.2.2. Redis Inbound Channel Adapter
The Redis inbound channel adapter (RedisInboundChannelAdapter
) adapts incoming Redis messages into Spring messages in the same way as other inbound adapters.
It receives platform-specific messages (Redis in this case) and converts them to Spring messages by using a MessageConverter
strategy.
The following example shows how to configure a Redis inbound channel adapter:
<int-redis:inbound-channel-adapter id="redisAdapter"
topics="thing1, thing2"
channel="receiveChannel"
error-channel="testErrorChannel"
message-converter="testConverter" />
<bean id="redisConnectionFactory"
class="o.s.data.redis.connection.lettuce.LettuceConnectionFactory">
<property name="port" value="7379" />
</bean>
<bean id="testConverter" class="things.something.SampleMessageConverter" />
The preceding example shows a simple but complete configuration of a Redis inbound channel adapter.
Note that the preceding configuration relies on the familiar Spring paradigm of auto-discovering certain beans.
In this case, the redisConnectionFactory
is implicitly injected into the adapter.
You can specify it explicitly by using the connection-factory
attribute instead.
Also, note that the preceding configuration injects the adapter with a custom MessageConverter
.
The approach is similar to JMS, where MessageConverter
instances are used to convert between Redis messages and the Spring Integration message payloads.
The default is a SimpleMessageConverter
.
Inbound adapters can subscribe to multiple topic names, hence the comma-separated set of values in the topics
attribute.
Since version 3.0, the inbound adapter, in addition to the existing topics
attribute, now has the topic-patterns
attribute.
This attribute contains a comma-separated set of Redis topic patterns.
For more information regarding Redis publish-subscribe, see Redis Pub/Sub.
Inbound adapters can use a RedisSerializer
to deserialize the body of Redis messages.
The serializer
attribute of the <int-redis:inbound-channel-adapter>
can be set to an empty string, which results in a null
value for the RedisSerializer
property.
In this case, the raw byte[]
bodies of Redis messages are provided as the message payloads.
Since version 5.0, you can provide an Executor
instance to the inbound adapter by using the task-executor
attribute of the <int-redis:inbound-channel-adapter>
.
Also, the received Spring Integration messages now have the RedisHeaders.MESSAGE_SOURCE
header to indicate the source of the published message: topic or pattern.
You can use this downstream for routing logic.
27.2.3. Redis Outbound Channel Adapter
The Redis outbound channel adapter adapts outgoing Spring Integration messages into Redis messages in the same way as other outbound adapters.
It receives Spring Integration messages and converts them to platform-specific messages (Redis in this case) by using a MessageConverter
strategy.
The following example shows how to configure a Redis outbound channel adapter:
<int-redis:outbound-channel-adapter id="outboundAdapter"
channel="sendChannel"
topic="thing1"
message-converter="testConverter"/>
<bean id="redisConnectionFactory"
class="o.s.data.redis.connection.lettuce.LettuceConnectionFactory">
<property name="port" value="7379"/>
</bean>
<bean id="testConverter" class="things.something.SampleMessageConverter" />
The configuration parallels the Redis inbound channel adapter.
The adapter is implicitly injected with a RedisConnectionFactory
, which is defined with redisConnectionFactory
as its bean name.
This example also includes the optional (and custom) MessageConverter
(the testConverter
bean).
Since Spring Integration 3.0, the <int-redis:outbound-channel-adapter>
offers an alternative to the topic
attribute: You can use the topic-expression
attribute to determine the Redis topic for the message at runtime.
These attributes are mutually exclusive.
27.2.4. Redis Queue Inbound Channel Adapter
Spring Integration 3.0 introduced a queue inbound channel adapter to “pop” messages from a Redis list. By default, it uses “right pop”, but you can configure it to use “left pop” instead. The adapter is message-driven. It uses an internal listener thread and does not use a poller.
The following listing shows all the available attributes for queue-inbound-channel-adapter
:
<int-redis:queue-inbound-channel-adapter id="" (1)
channel="" (2)
auto-startup="" (3)
phase="" (4)
connection-factory="" (5)
queue="" (6)
error-channel="" (7)
serializer="" (8)
receive-timeout="" (9)
recovery-interval="" (10)
expect-message="" (11)
task-executor="" (12)
right-pop=""/> (13)
| 1 | The component bean name.
If you do not provide the channel attribute, a DirectChannel is created and registered in the application context with this id attribute as the bean name.
In this case, the endpoint itself is registered with the bean name id plus .adapter .
(If the bean name were thing1 , the endpoint is registered as thing1.adapter .) |
| 2 | The MessageChannel to which to send Message instances from this Endpoint. |
| 3 | A SmartLifecycle attribute to specify whether this endpoint should start automatically after the application context start or not.
It defaults to true . |
| 4 | A SmartLifecycle attribute to specify the phase in which this endpoint is started.
It defaults to 0 . |
| 5 | A reference to a RedisConnectionFactory bean.
It defaults to redisConnectionFactory . |
| 6 | The name of the Redis list on which the queue-based 'pop' operation is performed to get Redis messages. |
| 7 | The MessageChannel to which to send ErrorMessage instances when exceptions are received from the listening task of the endpoint.
By default, the underlying MessagePublishingErrorHandler uses the default errorChannel from the application context. |
| 8 | The RedisSerializer bean reference.
It can be an empty string, which means 'no serializer'.
In this case, the raw byte[] from the inbound Redis message is sent to the channel as the Message payload.
By default it is a JdkSerializationRedisSerializer . |
| 9 | The timeout in milliseconds for 'pop' operation to wait for a Redis message from the queue. The default is 1 second. |
| 10 | The time in milliseconds for which the listener task should sleep after exceptions on the 'pop' operation, before restarting the listener task. |
| 11 | Specifies whether this endpoint expects data from the Redis queue to contain entire Message instances.
If this attribute is set to true , the serializer cannot be an empty string, because messages require some form of deserialization (JDK serialization by default).
Its default is false . |
| 12 | A reference to a Spring TaskExecutor (or standard JDK 1.5+ Executor ) bean.
It is used for the underlying listening task.
It defaults to a SimpleAsyncTaskExecutor . |
| 13 | Specifies whether this endpoint should use “right pop” (when true ) or “left pop” (when false ) to read messages from the Redis list.
If true , the Redis List acts as a FIFO queue when used with a default Redis queue outbound channel adapter.
Set it to false to use with software that writes to the list with “right push” or to achieve a stack-like message order.
Its default is true .
Since version 4.3. |
The task-executor has to be configured with more than one thread for processing; otherwise there is a possible deadlock when the RedisQueueMessageDrivenEndpoint tries to restart the listener task after an error.
The errorChannel can be used to process those errors, to avoid restarts, but it preferable to not expose your application to the possible deadlock situation.
See Spring Framework Reference Manual for possible TaskExecutor implementations.
|
27.2.5. Redis Queue Outbound Channel Adapter
Spring Integration 3.0 introduced a queue outbound channel adapter to “push” to a Redis list from Spring Integration messages.
By default, it uses “left push”, but you can configure it to use “right push” instead.
The following listing shows all the available attributes for a Redis queue-outbound-channel-adapter
:
<int-redis:queue-outbound-channel-adapter id="" (1)
channel="" (2)
connection-factory="" (3)
queue="" (4)
queue-expression="" (5)
serializer="" (6)
extract-payload="" (7)
left-push=""/> (8)
| 1 | The component bean name.
If you do not provide the channel attribute, a DirectChannel is created and registered in the application context with this id attribute as the bean name.
In this case, the endpoint is registered with a bean name of id plus .adapter .
(If the bean name were thing1 , the endpoint is registered as thing1.adapter .) |
| 2 | The MessageChannel from which this endpoint receives Message instances. |
| 3 | A reference to a RedisConnectionFactory bean.
It defaults to redisConnectionFactory . |
| 4 | The name of the Redis list on which the queue-based 'push' operation is performed to send Redis messages.
This attribute is mutually exclusive with queue-expression . |
| 5 | A SpEL Expression to determine the name of the Redis list.
It uses the incoming Message at runtime as the #root variable.
This attribute is mutually exclusive with queue . |
| 6 | A RedisSerializer bean reference.
It defaults to a JdkSerializationRedisSerializer .
However, for String payloads, a StringRedisSerializer is used, if a serializer reference is not provided. |
| 7 | Specifies whether this endpoint should send only the payload or the entire Message to the Redis queue.
It defaults to true . |
| 8 | Specifies whether this endpoint should use “left push” (when true ) or “right push” (when false ) to write messages to the Redis list.
If true , the Redis list acts as a FIFO queue when used with a default Redis queue inbound channel adapter.
Set it to false to use with software that reads from the list with “left pop” or to achieve a stack-like message order.
It defaults to true .
Since version 4.3. |
27.2.6. Redis Application Events
Since Spring Integration 3.0, the Redis module provides an implementation of IntegrationEvent
, which, in turn, is a org.springframework.context.ApplicationEvent
.
The RedisExceptionEvent
encapsulates exceptions from Redis operations (with the endpoint being the “source” of the event).
For example, the <int-redis:queue-inbound-channel-adapter/>
emits those events after catching exceptions from the BoundListOperations.rightPop
operation.
The exception may be any generic org.springframework.data.redis.RedisSystemException
or a org.springframework.data.redis.RedisConnectionFailureException
.
Handling these events with an <int-event:inbound-channel-adapter/>
can be useful to determine problems with background Redis tasks and to take administrative actions.
27.3. Redis Message Store
As described in the Enterprise Integration Patterns (EIP) book, a message store lets you persist messages.
This can be useful when dealing with components that have a capability to buffer messages (aggregator, resequencer, and others) when reliability is a concern.
In Spring Integration, the MessageStore
strategy also provides the foundation for the claim check pattern, which is described in EIP as well.
Spring Integration’s Redis module provides the RedisMessageStore
.
The following example shows how to use it with a aggregator:
<bean id="redisMessageStore" class="o.s.i.redis.store.RedisMessageStore">
<constructor-arg ref="redisConnectionFactory"/>
</bean>
<int:aggregator input-channel="inputChannel" output-channel="outputChannel"
message-store="redisMessageStore"/>
The preceding example is a bean configuration, and it expects a RedisConnectionFactory
as a constructor argument.
By default, the RedisMessageStore
uses Java serialization to serialize the message.
However, if you want to use a different serialization technique (such as JSON), you can provide your own serializer by setting the valueSerializer
property of the RedisMessageStore
.
Starting with version 4.3.10, the Framework provides Jackson serializer and deserializer implementations for Message
instances and MessageHeaders
instances — MessageJacksonDeserializer
and MessageHeadersJacksonSerializer
, respectively.
They have to be configured with the SimpleModule
options for the ObjectMapper
.
In addition, you should set enableDefaultTyping
on the ObjectMapper
to add type information for each serialized complex object.
That type information is then used during deserialization.
The framework provides a utility method called JacksonJsonUtils.messagingAwareMapper()
, which is already supplied with all the previously mentioned properties and serializers.
To manage JSON serialization in the RedisMessageStore
, you must configure it in a fashion similar to the following example:
RedisMessageStore store = new RedisMessageStore(redisConnectionFactory);
ObjectMapper mapper = JacksonJsonUtils.messagingAwareMapper();
RedisSerializer<Object> serializer = new GenericJackson2JsonRedisSerializer(mapper);
store.setValueSerializer(serializer);
Starting with version 4.3.12, RedisMessageStore
supports the prefix
option to allow distinguishing between instances of the store on the same Redis server.
27.3.1. Redis Channel Message Stores
The RedisMessageStore
shown earlier maintains each group as a value under a single key (the group ID).
While you can use this to back a QueueChannel
for persistence, a specialized RedisChannelMessageStore
is provided for that purpose (since version 4.0).
This store uses a LIST
for each channel, LPUSH
when sending messages, and RPOP
when receiving messages.
By default, this store also uses JDK serialization, but you can modify the value serializer, as described earlier.
We recommend using this store backing channels, instead of using the general RedisMessageStore
.
The following example defines a Redis message store and uses it in a channel with a queue:
<bean id="redisMessageStore" class="o.s.i.redis.store.RedisChannelMessageStore">
<constructor-arg ref="redisConnectionFactory"/>
</bean>
<int:channel id="somePersistentQueueChannel">
<int:queue message-store="redisMessageStore"/>
<int:channel>
The keys used to store the data have the form: <storeBeanName>:<channelId>
(in the preceding example, redisMessageStore:somePersistentQueueChannel
).
In addition, a subclass RedisChannelPriorityMessageStore
is also provided.
When you use this with a QueueChannel
, the messages are received in (FIFO) priority order.
It uses the standard IntegrationMessageHeaderAccessor.PRIORITY
header and supports priority values (0 - 9
).
Messages with other priorities (and messages with no priority) are retrieved in FIFO order after any messages with priority.
These stores implement only BasicMessageGroupStore and do not implement MessageGroupStore .
They can be used only for situations such as backing a QueueChannel .
|
27.4. Redis Metadata Store
Spring Integration 3.0 introduced a new Redis-based MetadataStore
(see Metadata Store) implementation.
You can use the RedisMetadataStore
to maintain the state of a MetadataStore
across application restarts.
You can use this new MetadataStore
implementation with adapters such as:
To instruct these adapters to use the new RedisMetadataStore
, declare a Spring bean named metadataStore
.
The Feed inbound channel adapter and the feed inbound channel adapter both automatically pick up and use the declared RedisMetadataStore
.
The following example shows how to declare such a bean:
<bean name="metadataStore" class="o.s.i.redis.store.metadata.RedisMetadataStore">
<constructor-arg name="connectionFactory" ref="redisConnectionFactory"/>
</bean>
The RedisMetadataStore
is backed by RedisProperties
.
Interaction with it uses BoundHashOperations
, which, in turn, requires a key
for the entire Properties
store.
In the case of the MetadataStore
, this key
plays the role of a region, which is useful in a distributed environment, when several applications use the same Redis server.
By default, this key
has a value of MetaData
.
Starting with version 4.0, this store implements ConcurrentMetadataStore
, letting it be reliably shared across multiple application instances where only one instance is allowed to store or modify a key’s value.
You cannot use the RedisMetadataStore.replace() (for example, in the AbstractPersistentAcceptOnceFileListFilter ) with a Redis cluster, since the WATCH command for atomicity is not currently supported.
|
27.5. Redis Store Inbound Channel Adapter
The Redis store inbound channel adapter is a polling consumer that reads data from a Redis collection and sends it as a Message
payload.
The following example shows how to configure a Redis store inbound channel adapter:
<int-redis:store-inbound-channel-adapter id="listAdapter"
connection-factory="redisConnectionFactory"
key="myCollection"
channel="redisChannel"
collection-type="LIST" >
<int:poller fixed-rate="2000" max-messages-per-poll="10"/>
</int-redis:store-inbound-channel-adapter>
The preceding example shows how to configure a Redis store inbound channel adapter by using the store-inbound-channel-adapter
element, providing values for various attributes, such as:
-
key
orkey-expression
: The name of the key for the collection being used. -
collection-type
: An enumeration of the collection types supported by this adapter. The supported Collections areLIST
,SET
,ZSET
,PROPERTIES
, andMAP
. -
connection-factory
: Reference to an instance ofo.s.data.redis.connection.RedisConnectionFactory
. -
redis-template
: Reference to an instance ofo.s.data.redis.core.RedisTemplate
. -
Other attributes that are common across all other inbound adapters (such as 'channel').
You cannot set both redis-template and connection-factory .
|
|
By default, the adapter uses a
Th |
Because it has a literal value for the key
, the preceding example is relatively simple and static.
Sometimes, you may need to change the value of the key at runtime based on some condition.
To do so, use key-expression
instead, where the provided expression can be any valid SpEL expression.
Also, you may wish to perform some post-processing on the successfully processed data that was read from the Redis collection.
For example, you may want to move or remove the value after its been processed.
You can do so by using the transaction synchronization feature that was added with Spring Integration 2.2.
The following example uses key-expression
and transaction synchronization:
<int-redis:store-inbound-channel-adapter id="zsetAdapterWithSingleScoreAndSynchronization"
connection-factory="redisConnectionFactory"
key-expression="'presidents'"
channel="otherRedisChannel"
auto-startup="false"
collection-type="ZSET">
<int:poller fixed-rate="1000" max-messages-per-poll="2">
<int:transactional synchronization-factory="syncFactory"/>
</int:poller>
</int-redis:store-inbound-channel-adapter>
<int:transaction-synchronization-factory id="syncFactory">
<int:after-commit expression="payload.removeByScore(18, 18)"/>
</int:transaction-synchronization-factory>
<bean id="transactionManager" class="o.s.i.transaction.PseudoTransactionManager"/>
You can declare your poller to be transactional by using a transactional
element.
This element can reference a real transaction manager (for example, if some other part of your flow invokes JDBC).
If you do not have a “real” transaction, you can use an o.s.i.transaction.PseudoTransactionManager
, which is an implementation of Spring’s PlatformTransactionManager
and enables the use of the transaction synchronization features of the Redis adapter when there is no actual transaction.
| This does not make the Redis activities themselves transactional. It lets the synchronization of actions be taken before or after success (commit) or after failure (rollback). |
Once your poller is transactional, you can set an instance of the o.s.i.transaction.TransactionSynchronizationFactory
on the transactional
element.
TransactionSynchronizationFactory
creates an instance of the TransactionSynchronization
.

Expressions for before-commit, after-commit, and after-rollback are supported, together with channels (one for each kind of event) where the evaluation result (if any) is sent.
For each child element, you can specify expression
and channel
attributes.
If only the channel
attribute is present, the received message is sent there as part of the particular synchronization scenario.
If only the expression
attribute is present and the result of an expression is a non-null value, a message with the result as the payload is generated and sent to a default channel (NullChannel
) and appears in the logs (at the DEBUG
level).
If you want the evaluation result to go to a specific channel, add a channel
attribute.
If the result of an expression is null or void, no message is generated.
For more information about transaction synchronization, see Transaction Synchronization.
27.6. RedisStore Outbound Channel Adapter
The RedisStore outbound channel adapter lets you write a message payload to a Redis collection, as the following example shows:
<int-redis:store-outbound-channel-adapter id="redisListAdapter"
collection-type="LIST"
channel="requestChannel"
key="myCollection" />
The preceding configuration a Redis store outbound channel adapter by using the store-inbound-channel-adapter
element.
It provides values for various attributes, such as:
-
key
orkey-expression
: The name of the key for the collection being used. -
extract-payload-elements
: If set totrue
(the default) and the payload is an instance of a “multi-value” object (that is, aCollection
or aMap
), it is stored by using “addAll” and “putAll” semantics. Otherwise, if set tofalse
, the payload is stored as a single entry regardless of its type. If the payload is not an instance of a “multi-value” object, the value of this attribute is ignored and the payload is always stored as a single entry. -
collection-type
: An enumeration of theCollection
types supported by this adapter. The supported Collections areLIST
,SET
,ZSET
,PROPERTIES
, andMAP
. -
map-key-expression
: SpEL expression that returns the name of the key for the entry being stored. It applies only if thecollection-type
isMAP
orPROPERTIES
and 'extract-payload-elements' is false. -
connection-factory
: Reference to an instance ofo.s.data.redis.connection.RedisConnectionFactory
. -
redis-template
: Reference to an instance ofo.s.data.redis.core.RedisTemplate
. -
Other attributes that are common across all other inbound adapters (such as 'channel').
You cannot set both redis-template and connection-factory .
|
By default, the adapter uses a StringRedisTemplate .
This uses StringRedisSerializer instances for keys, values, hash keys, and hash values.
However, if extract-payload-elements is set to false , a RedisTemplate that has StringRedisSerializer instances for keys and hash keys and JdkSerializationRedisSerializer instances s for values and hash values will be used.
With the JDK serializer, it is important to understand that Java serialization is used for all values, regardless of whether the value is actually a collection or not.
If you need more control over the serialization of values, consider providing your own RedisTemplate rather than relying upon these defaults.
|
Because it has literal values for the key
and other attributes, the preceding example is relatively simple and static.
Sometimes, you may need to change the values dynamically at runtime based on some condition.
To do so, use their -expression
equivalents (key-expression
, map-key-expression
, and so on), where the provided expression can be any valid SpEL expression.
27.7. Redis Outbound Command Gateway
Spring Integration 4.0 introduced the Redis command gateway to let you perform any standard Redis command by using the generic RedisConnection#execute
method.
The following listing shows the available attributes for the Redis outbound gateway:
<int-redis:outbound-gateway
request-channel="" (1)
reply-channel="" (2)
requires-reply="" (3)
reply-timeout="" (4)
connection-factory="" (5)
redis-template="" (6)
arguments-serializer="" (7)
command-expression="" (8)
argument-expressions="" (9)
use-command-variable="" (10)
arguments-strategy="" /> (11)
| 1 | The MessageChannel from which this endpoint receives Message instances. |
| 2 | The MessageChannel where this endpoint sends reply Message instances. |
| 3 | Specifies whether this outbound gateway must return a non-null value.
It defaults to true .
A ReplyRequiredException is thrown when Redis returns a null value. |
| 4 | The timeout (in milliseconds) to wait until the reply message is sent. It is typically applied for queue-based limited reply-channels. |
| 5 | A reference to a RedisConnectionFactory bean.
It defaults to redisConnectionFactory .
It is mutually exclusive with 'redis-template' attribute. |
| 6 | A reference to a RedisTemplate bean.
It is mutually exclusive with 'connection-factory' attribute. |
| 7 | A reference to an instance of org.springframework.data.redis.serializer.RedisSerializer .
It is used to serialize each command argument to byte[], if necessary. |
| 8 | The SpEL expression that returns the command key.
It defaults to the redis_command message header.
It must not evaluate to null . |
| 9 | Comma-separated SpEL expressions that are evaluated as command arguments.
Mutually exclusive with the arguments-strategy attribute.
If you provide neither attribute, the payload is used as the command arguments.
The argument expressions can evaluate to 'null' to support a variable number of arguments. |
| 10 | A boolean flag to specify whether the evaluated Redis command string is made available as the #cmd variable in the expression evaluation context in the o.s.i.redis.outbound.ExpressionArgumentsStrategy when argument-expressions is configured.
Otherwise this attribute is ignored. |
| 11 | Reference to an instance of o.s.i.redis.outbound.ArgumentsStrategy .
It is mutually exclusive with argument-expressions attribute.
If you provide neither attribute, the payload is used as the command arguments. |
You can use the <int-redis:outbound-gateway>
as a common component to perform any desired Redis operation.
For example, the following examlpe shows how to get incremented values from Redis atomic number:
<int-redis:outbound-gateway request-channel="requestChannel"
reply-channel="replyChannel"
command-expression="'INCR'"/>
The Message
payload should have a name of redisCounter
, which may be provided by org.springframework.data.redis.support.atomic.RedisAtomicInteger
bean definition.
The RedisConnection#execute
method has a generic Object
as its return type.
Real result depends on command type.
For example, MGET
returns a List<byte[]>
.
For more information about commands, their arguments and result type, see Redis Specification.
27.8. Redis Queue Outbound Gateway
Spring Integration introduced the Redis queue outbound gateway to perform request and reply scenarios.
It pushes a conversation UUID
to the provided queue
, pushes the value with that UUID
as its key to a Redis list, and waits for the reply from a Redis list with a key of UUID' plus '.reply
.
A different UUID is used for each interaction.
The following listing shows the available attributes for a Redis outbound gateway:
<int-redis:queue-outbound-gateway
request-channel="" (1)
reply-channel="" (2)
requires-reply="" (3)
reply-timeout="" (4)
connection-factory="" (5)
queue="" (6)
order="" (7)
serializer="" (8)
extract-payload=""/> (9)
| 1 | The MessageChannel from which this endpoint receives Message instances. |
| 2 | The MessageChannel where this endpoint sends reply Message instances. |
| 3 | Specifies whether this outbound gateway must return a non-null value.
This value is false by default.
Otherwise, a ReplyRequiredException is thrown when Redis returns a null value. |
| 4 | The timeout (in milliseconds) to wait until the reply message is sent. It is typically applied for queue-based limited reply-channels. |
| 5 | A reference to a RedisConnectionFactory bean.
It defaults to redisConnectionFactory .
It is mutually exclusive with the 'redis-template' attribute. |
| 6 | The name of the Redis list to which the outbound gateway sends a conversation UUID . |
| 7 | The order of this outbound gateway when multiple gateways are registered. |
| 8 | The RedisSerializer bean reference.
It can be an empty string, which means “no serializer”.
In this case, the raw byte[] from the inbound Redis message is sent to the channel as the Message payload.
By default, it is a JdkSerializationRedisSerializer . |
| 9 | Specifies whether this endpoint expects data from the Redis queue to contain entire Message instances.
If this attribute is set to true , the serializer cannot be an empty string, because messages require some form of deserialization (JDK serialization by default). |
27.9. Redis Queue Inbound Gateway
Spring Integration 4.1 introduced the Redis queue inbound gateway to perform request and reply scenarios.
It pops a conversation UUID
from the provided queue
, pops the value with that UUID
as its key from the Redis list, and pushes the reply to the Redis list with a key of UUID
plus .reply
.
The following listing shows the available attributes for a Redis queue inbound gateway:
<int-redis:queue-inbound-gateway
request-channel="" (1)
reply-channel="" (2)
executor="" (3)
reply-timeout="" (4)
connection-factory="" (5)
queue="" (6)
order="" (7)
serializer="" (8)
receive-timeout="" (9)
expect-message="" (10)
recovery-interval=""/> (11)
| 1 | The MessageChannel from which this endpoint receives Message instances. |
| 2 | The MessageChannel where this endpoint sends reply Message instances. |
| 3 | A reference to a Spring TaskExecutor (or a standard JDK 1.5+ Executor ) bean.
It is used for the underlying listening task.
It defaults to a SimpleAsyncTaskExecutor . |
| 4 | The timeout (in milliseconds) to wait until the reply message is sent. It is typically applied for queue-based limited reply-channels. |
| 5 | A reference to a RedisConnectionFactory bean.
It defaults to redisConnectionFactory .
It is mutually exclusive with 'redis-template' attribute. |
| 6 | The name of the Redis list for the conversation UUID . |
| 7 | The order of this inbound gateway when multiple gateways are registered. |
| 8 | The RedisSerializer bean reference.
It can be an empty string, which means “no serializer”.
In this case, the raw byte[] from the inbound Redis message is sent to the channel as the Message payload.
It default to a JdkSerializationRedisSerializer .
(Note that, in releases before version 4.3, it was a StringRedisSerializer by default.
To restore that behavior, provide a reference to a StringRedisSerializer ). |
| 9 | The timeout (in milliseconds) to wait until the receive message is fetched. It is typically applied for queue-based limited request-channels. |
| 10 | Specifies whether this endpoint expects data from the Redis queue to contain entire Message instances.
If this attribute is set to true , the serializer cannot be an empty string, because messages require some form of deserialization (JDK serialization by default). |
| 11 | The time (in milliseconds) the listener task should sleep after exceptions on the “right pop” operation before restarting the listener task. |
The task-executor has to be configured with more than one thread for processing; otherwise there is a possible deadlock when the RedisQueueMessageDrivenEndpoint tries to restart the listener task after an error.
The errorChannel can be used to process those errors, to avoid restarts, but it preferable to not expose your application to the possible deadlock situation.
See Spring Framework Reference Manual for possible TaskExecutor implementations.
|
27.10. Redis Lock Registry
Spring Integration 4.0 introduced the RedisLockRegistry
.
Certain components (for example, aggregator and resequencer) use a lock obtained from a LockRegistry
instance to ensure that only one thread manipulates a group at a time.
The DefaultLockRegistry
performs this function within a single component.
You can now configure an external lock registry on these components.
When you use it with a shared MessageGroupStore
, you can use the RedisLockRegistry
to provide this functionality across multiple application instances, such that only one instance can manipulate the group at a time.
When a lock is released by a local thread, another local thread can generally acquire the lock immediately. If a lock is released by a thread using a different registry instance, it can take up to 100ms to acquire the lock.
To avoid “hung” locks (when a server fails), the locks in this registry are expired after a default 60 seconds, but you can configure this value on the registry. Locks are normally held for a much smaller time.
| Because the keys can expire, an attempt to unlock an expired lock results in an exception being thrown. However, the resources protected by such a lock may have been compromised, so such exceptions should be considered to be severe. You should set the expiry at a large enough value to prevent this condition, but set it low enough that the lock can be recovered after a server failure in a reasonable amount of time. |
Starting with version 5.0, the RedisLockRegistry
implements ExpirableLockRegistry
, which removes locks last acquired more than age
ago and that are not currently locked.
28. Resource Support
The resource inbound channel adapter builds upon Spring’s Resource
abstraction to support greater flexibility across a variety of actual types of underlying resources, such as a file, a URL, or a class path resource.
Therefore, it is similar to but more generic than the file inbound channel adapter.
28.1. Resource Inbound Channel Adapter
The resource inbound channel adapter is a polling adapter that creates a Message
whose payload is a collection of Resource
objects.
Resource
objects are resolved based on the pattern specified by the pattern
attribute.
The collection of resolved Resource
objects is then sent as a payload within a Message
to the adapter’s channel.
That is one major difference between resource inbound channel adapter and file inbound channel adapter: The latter buffers File
objects and sends a single File
object per Message
.
The following example shows a simple configuration that finds all files that end with the 'properties' extension in the things.thing1
package available on the classpath and sends them as the payload of a Message
to the channel named ‘resultChannel’:
<int:resource-inbound-channel-adapter id="resourceAdapter"
channel="resultChannel"
pattern="classpath:things/thing1/*.properties">
<int:poller fixed-rate="1000"/>
</int:resource-inbound-channel-adapter>
The resource inbound channel adapter relies on the org.springframework.core.io.support.ResourcePatternResolver
strategy interface to resolve the provided pattern.
It defaults to an instance of the current ApplicationContext
.
However, you can provide a reference to an instance of your own implementation of ResourcePatternResolver
by setting the pattern-resolver
attribute, as the following example shows:
<int:resource-inbound-channel-adapter id="resourceAdapter"
channel="resultChannel"
pattern="classpath:things/thing1/*.properties"
pattern-resolver="myPatternResolver">
<int:poller fixed-rate="1000"/>
</int:resource-inbound-channel-adapter>
<bean id="myPatternResolver" class="org.example.MyPatternResolver"/>
You may have a use case where you need to further filter the collection of resources resolved by the ResourcePatternResolver
.
For example, you may want to prevent resources that were already resolved from appearing in a collection of resolved resources ever again.
On the other hand, your resources might be updated rather often and you do want them to be picked up again.
In other words, both defining an additional filter and disabling filtering altogether are valid use cases.
You can provide your own implementation of the org.springframework.integration.util.CollectionFilter
strategy interface, as the following example shows:
public interface CollectionFilter<T> {
Collection<T> filter(Collection<T> unfilteredElements);
}
The CollectionFilter
receives a collection of un-filtered elements (which are Resource
objects in the preceding example), and it returns a collection of filtered elements of that same type.
If you define the adapter with XML but you do not specify a filter reference, the resource inbound channel adapter uses a default implementation of CollectionFilter
.
The implementation class of that default filter is org.springframework.integration.util.AcceptOnceCollectionFilter
.
It remembers the elements passed in the previous invocation in order to avoid returning those elements more than once.
To inject your own implementation of CollectionFilter
instead, use the filter
attribute, as the following example shows:
<int:resource-inbound-channel-adapter id="resourceAdapter"
channel="resultChannel"
pattern="classpath:things/thing1/*.properties"
filter="myFilter">
<int:poller fixed-rate="1000"/>
</int:resource-inbound-channel-adapter>
<bean id="myFilter" class="org.example.MyFilter"/>
If you do not need any filtering and want to disable even the default CollectionFilter
strategy, provide an empty value for the filter attribute (for example, filter=""
)
29. RMI Support
This chapter explains how to use channel adapters that are specific to RMI (Remote Method Invocation) to distribute a system over multiple JVMs. The first section deals with sending messages over RMI. The second section shows how to receive messages over RMI. The last section shows how to define RMI channel adapters by using the namespace support.
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-rmi</artifactId>
<version>5.1.5.RELEASE</version>
</dependency>
compile "org.springframework.integration:spring-integration-rmi:5.1.5.RELEASE"
29.1. Outbound RMI
To send messages from a channel over RMI, you can define an RmiOutboundGateway
.
This gateway uses Spring’s RmiProxyFactoryBean
internally to create a proxy for a remote gateway.
Note that, to invoke a remote interface that does not use Spring Integration, you should use a service activator in combination with Spring’s RmiProxyFactoryBean.
To configure the outbound gateway, you can write a bean definition similar the following:
<bean id="rmiOutGateway" class=org.spf.integration.rmi.RmiOutboundGateway>
<constructor-arg value="rmi://host"/>
<property name="replyChannel" value="replies"/>
</bean>
29.2. Inbound RMI
To receive messages over RMI, you need to use an RmiInboundGateway
.
You can configure the gateway as shown in the following example:
<bean id="rmiInGateway" class=org.spf.integration.rmi.RmiInboundGateway>
<property name="requestChannel" value="requests"/>
</bean>
If you use an errorChannel on an inbound gateway, it the error flow normally returns a result or throws an exception.
This is because it is likely that there is a corresponding outbound gateway waiting for a response of some kind.
Consuming a message on the error flow and not replying results in no reply for the inbound gateway.
Exceptions (on the main flow when there is no errorChannel or on the error flow) propagate to the corresponding inbound gateway.
|
29.3. RMI namespace support
To configure the inbound gateway, you can use the namespace support for it. The following code snippet shows the different configuration options that are supported:
<int-rmi:inbound-gateway id="gatewayWithDefaults" request-channel="testChannel"/>
<int-rmi:inbound-gateway id="gatewayWithCustomProperties" request-channel="testChannel"
expect-reply="false" request-timeout="123" reply-timeout="456"/>
<int-rmi:inbound-gateway id="gatewayWithHost" request-channel="testChannel"
registry-host="localhost"/>
<int-rmi:inbound-gateway id="gatewayWithPort" request-channel="testChannel"
registry-port="1234" error-channel="rmiErrorChannel"/>
<int-rmi:inbound-gateway id="gatewayWithExecutorRef" request-channel="testChannel"
remote-invocation-executor="invocationExecutor"/>
You can also use the namespace support to configure the outbound gateway. The following code snippet shows the different configuration for an outbound RMI gateway:
<int-rmi:outbound-gateway id="gateway"
request-channel="localChannel"
remote-channel="testChannel"
host="localhost"/>
29.4. Configuring with Java Configuration
The following example shows how to configure an inbound gateway and an outbound gateway with Java:
@Bean
public RmiInboundGateway inbound() {
RmiInboundGateway gateway = new RmiInboundGateway();
gateway.setRequestChannel(requestChannel());
gateway.setRegistryHost("host");
gateway.setRegistryPort(port);
return gateway;
}
@Bean
@ServiceActivator(inputChannel="inChannel")
public RmiOutboundGateway outbound() {
RmiOutboundGateway gateway = new RmiOutboundGateway("rmi://host:port/"
+ RmiInboundGateway.SERVICE_NAME_PREFIX + "remoteChannelName");
return gateway;
}
As of version 4.3, the outbound gateway has a second constructor that takes an RmiProxyFactoryBeanConfigurer
instance, along with the service url argument.
It allows further configuration before the proxy is created — for example, to inject a Spring Security ContextPropagatingRemoteInvocationFactory
, as the following example shows:
@Bean
@ServiceActivator(inputChannel="inChannel")
public RmiOutboundGateway outbound() {
RmiOutboundGateway gateway = new RmiOutboundGateway("rmi://host:port/"
+ RmiInboundGateway.SERVICE_NAME_PREFIX + "remoteChannelName",
pfb -> {
pfb.setRemoteInvocationFactory(new ContextPropagatingRemoteInvocationFactory());
});
return gateway;
}
Starting with version 5.0, you can set this with the XML namespace by using the configurer
attribute.
30. SFTP Adapters
Spring Integration provides support for file transfer operations over SFTP.
The Secure File Transfer Protocol (SFTP) is a network protocol that lets you transfer files between two computers on the Internet over any reliable stream.
The SFTP protocol requires a secure channel, such as SSH, and visibility to a client’s identity throughout the SFTP session.
Spring Integration supports sending and receiving files over SFTP by providing three client side endpoints: inbound channel adapter, outbound channel adapter, and outbound gateway. It also provides convenient namespace configuration to define these client components.
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-sftp</artifactId>
<version>5.1.5.RELEASE</version>
</dependency>
compile "org.springframework.integration:spring-integration-sftp:5.1.5.RELEASE"
To include the SFTP namespace in your xml configuration, include the following attributes on the root element:
xmlns:int-sftp="http://www.springframework.org/schema/integration/sftp"
xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp
https://www.springframework.org/schema/integration/sftp/spring-integration-sftp.xsd"
30.1. SFTP Session Factory
| As of version 3.0, sessions are no longer cached by default. See SFTP Session Caching. |
Before configuring SFTP adapters, you must configure an SFTP session factory. You can configure the SFTP session factory with a regular bean definition, as the following example shows:
<beans:bean id="sftpSessionFactory"
class="org.springframework.integration.sftp.session.DefaultSftpSessionFactory">
<beans:property name="host" value="localhost"/>
<beans:property name="privateKey" value="classpath:META-INF/keys/sftpTest"/>
<beans:property name="privateKeyPassphrase" value="springIntegration"/>
<beans:property name="port" value="22"/>
<beans:property name="user" value="kermit"/>
</beans:bean>
Every time an adapter requests a session object from its SessionFactory
, a new SFTP session is created.
Under the covers, the SFTP Session Factory relies on the JSch library to provide the SFTP capabilities.
However, Spring Integration also supports the caching of SFTP sessions. See SFTP Session Caching for more information.
|
JSch supports multiple channels (operations) over a connection to the server.
By default, the Spring Integration session factory uses a separate physical connection for each channel.
Since Spring Integration 3.0, you can configure the session factory (using a boolean constructor arg - default When using this feature, you must wrap the session factory in a caching session factory, as described later, so that the connection is not physically closed when an operation completes. If the cache is reset, the session is disconnected only when the last channel is closed. The connection is refreshed if it is found to be disconnected when a new operation obtains a session. |
If you experience connectivity problems and would like to trace session creation and see which sessions are polled, you may enable tracing by setting the logger to TRACE level (for example, log4j.category.org.springframework.integration.sftp=TRACE ).
See SFTP/JSCH Logging.
|
Now all you need to do is inject this SFTP session factory into your adapters.
| A more practical way to provide values for the SFTP session factory is to use Spring’s property placeholder support. |
30.1.1. Configuration Properties
The following list describes all the properties that are exposed by the DefaultSftpSessionFactory
.
isSharedSession
(constructor argument)::When true
, a single connection is used, and JSch Channels
are multiplexed.
It defaults to false
.
clientVersion
::Lets you set the client version property.
It’s default depends on the underlying JSch version but it will look like:_SSH-2.0-JSCH-0.1.45_
enableDaemonThread
::If true
, all threads are daemon threads.
If set to false
, normal non-daemon threads are used instead.
This property is set on the underlying session.
There, this property defaults to false
.
host
::The URL of the host to which you want to connect.
Required.
hostKeyAlias
::Sets the host key alias, which is used when comparing the host key to the known hosts list.
knownHosts
::Specifies the filename that used for a host key repository.
The file has the same format as OpenSSH’s known_hosts
file and is required and must be pre-populated if allowUnknownKeys
is false.
password
::The password to authenticate against the remote host.
If a password is not provided, then the privateKey
property is required.
It is not allowed if you set userInfo
.
The password is obtained from that object.
port
::The port over which the SFTP connection shall be established.
If not specified, this value defaults to 22
.
If specified, this properties must be a positive number.
privateKey
::Lets you set a resource that represents the location of the private key used for authenticating against the remote host.
If the privateKey
is not provided, then the password
property is required.
privateKeyPassphrase
::The password for the private key.
If you set userInfo
, privateKeyPassphrase
is not allowed .
The passphrase is obtained from that object.
Optional.
proxy
::Allows for specifying a JSch-based proxy.
If set, the proxy object is used to create the connection to the remote host through the proxy.
See Proxy Factory Bean for a convenient way to configure the proxy.
serverAliveCountMax
::Specifies the number of server-alive messages, which are sent without any reply from the server before disconnecting.
If not set, this property defaults to 1
.
serverAliveInterval
::Sets the timeout interval (in milliseconds) before a server-alive message is sent, in case no message is received from the server.
sessionConfig
::By using Properties
, you can set additional configuration setting on the underlying JSch Session.
socketFactory
::Lets you pass in a SocketFactory
.
The socket factory is used to create a socket to the target host.
When a proxy is used, the socket factory is passed to the proxy.
By default, plain TCP sockets are used.
timeout
::The timeout property is used as the socket timeout parameter, as well as the default connection timeout.
Defaults to 0
, which means, that no timeout will occur.
user
::The remote user to use.
Required.
allowUnknownKeys
::Set to true
to allow connections to hosts with unknown (or changed) keys.
Its default is 'false'.
It is applied only if no userInfo
is provided.
If false
, a pre-populated knownHosts
file is required.
userInfo
::Set a custom UserInfo
to be used during authentication.
In particular, promptYesNo()
is invoked when an unknown (or changed) host key is received.
See also allowUnknownKeys
.
When you provide a UserInfo
, the password
and private key passphrase
are obtained from it, and you cannot set discrete
password
and privateKeyPassprase
properties.
30.2. Proxy Factory Bean
Jsch
provides a mechanism to connect to the server over an HTTP or SOCKS proxy.
To use this feature, configure the Proxy
and provide a reference to the DefaultSftpSessionFactory
, as discussed
earlier.
Three implementations are provided by Jsch
: HTTP
, SOCKS4
, and SOCKS5
.
Spring Integration 4.3 introduced a FactoryBean
, easing configuration of these proxies by allowing property
injection, as the following example shows:
<bean id="proxySocks5" class="org.springframework.integration.sftp.session.JschProxyFactoryBean">
<constructor-arg value="SOCKS5" />
<constructor-arg value="${sftp.proxy.address}" />
<constructor-arg value="${sftp.proxy.port}" />
<constructor-arg value="${sftp.proxy.user}" />
<constructor-arg value="${sftp.proxy.pw}" />
</bean>
<bean id="sessionFactory"
class="org.springframework.integration.sftp.session.DefaultSftpSessionFactory" >
...
<property name="proxy" ref="proxySocks5" />
...
</bean>
30.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, you can call setThreadKey()
on the factory to associate a key with the current thread.
That key is then used to look up 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 more easily do so from a message flow, as the following example shows:
<bean id="dsf" class="org.springframework.integration.file.remote.session.DelegatingSessionFactory">
<constructor-arg>
<bean class="o.s.i.file.remote.session.DefaultSessionFactoryLocator">
<!-- delegate factories here -->
</bean>
</constructor-arg>
</bean>
<int:service-activator input-channel="in" output-channel="c1"
expression="@dsf.setThreadKey(#root, headers['factoryToUse'])" />
<int-sftp:outbound-gateway request-channel="c1" reply-channel="c2" ... />
<int:service-activator input-channel="c2" output-channel="out"
expression="@dsf.clearThreadKey(#root)" />
When using session caching (see SFTP 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.
30.4. SFTP Session Caching
Starting with Spring Integration version 3.0, sessions are no longer cached by default.
The cache-sessions attribute is no longer supported on endpoints.
If you wish to cache sessions, you must use a CachingSessionFactory (see the next example).
|
In versions prior to 3.0, the sessions were automatically cached by default.
A cache-sessions
attribute was available for disabling the auto caching, but that solution did not provide a way to configure other session-caching attributes.
For example, you could not limit on the number of sessions created.
To support that requirement and other configuration options, we added a CachingSessionFactory
.
It provides sessionCacheSize
and sessionWaitTimeout
properties.
As its name suggests, the sessionCacheSize
property controls how many active sessions the factory maintains in its cache (the default is unbounded).
If the sessionCacheSize
threshold has been reached, any attempt to acquire another session blocks until either one of the cached sessions becomes available or until the wait time for a session expires (the default wait time is Integer.MAX_VALUE
).
The sessionWaitTimeout
property enables configuration of the wait time.
If you want your sessions to be cached, configure your default session factory (as described earlier) and then wrap it in an instance of CachingSessionFactory
where you may provide those additional properties.
The following example shows how to do so:
<bean id="sftpSessionFactory"
class="org.springframework.integration.sftp.session.DefaultSftpSessionFactory">
<property name="host" value="localhost"/>
</bean>
<bean id="cachingSessionFactory"
class="org.springframework.integration.file.remote.session.CachingSessionFactory">
<constructor-arg ref="sftpSessionFactory"/>
<constructor-arg value="10"/>
<property name="sessionWaitTimeout" value="1000"/>
</bean>
The preceding example creates a CachingSessionFactory
with its sessionCacheSize
set to 10
and its sessionWaitTimeout
set to one second (1000 milliseconds).
Starting with Spring Integration version 3.0, the CachingConnectionFactory
provides a resetCache()
method.
When invoked, all idle sessions are immediately closed and in-use sessions are closed when they are returned to the cache.
When using isSharedSession=true
, the channel is closed and the shared session is closed only when the last channel is closed.
New requests for sessions establish new sessions as necessary.
Starting with version 5.1, the CachingSessionFactory
has a new property testSession
.
When true, the session will be tested by performing a stat(getHome())
command to ensure it is still active; if not, it will be removed from the cache; a new session is created if no active sessions are in the cache.
30.5. Using RemoteFileTemplate
Spring Integration version 3.0 provides a new abstraction over the SftpSession
object.
The template provides methods to send, retrieve (as an InputStream
), remove, and rename files.
In addition, we provide an execute
method to let the caller run multiple operations on the session.
In all cases, the template takes care of reliably closing the session.
For more information, see the Javadoc for RemoteFileTemplate
There is a subclass for SFTP: SftpRemoteFileTemplate
.
We added additional methods in version 4.1, including getClientInstance()
.
It provides access to the underlying ChannelSftp
, which enables access to low-level APIs.
Version 5.0 introduced the RemoteFileOperations.invoke(OperationsCallback<F, T> action)
method.
This method lets several RemoteFileOperations
calls be called in the scope of the same thread-bounded Session
.
This is useful when you need to perform several high-level operations of the RemoteFileTemplate
as one unit of work.
For example, AbstractRemoteFileOutboundGateway
uses it with the mput
command implementation, where we perform a put
operation for each file in the provided directory and recursively for its sub-directories.
See the Javadoc for more information.
30.6. SFTP Inbound Channel Adapter
The SFTP inbound channel adapter is a special listener that connects to the server and listens for the remote directory events (such as a new file being created), at which point it initiates a file transfer. The following example shows how to configure an SFTP inbound channel adapter:
<int-sftp:inbound-channel-adapter id="sftpAdapterAutoCreate"
session-factory="sftpSessionFactory"
channel="requestChannel"
filename-pattern="*.txt"
remote-directory="/foo/bar"
preserve-timestamp="true"
local-directory="file:target/foo"
auto-create-local-directory="true"
local-filename-generator-expression="#this.toUpperCase() + '.a'"
scanner="myDirScanner"
local-filter="myFilter"
temporary-file-suffix=".writing"
max-fetch-size="-1"
delete-remote-files="false">
<int:poller fixed-rate="1000"/>
</int-sftp:inbound-channel-adapter>
The preceding configuration example shows how to provide values for various attributes, including the following:
-
local-directory
: The location to which files are going to be transferred -
remote-directory
: The remote source directory from which files are going to be transferred -
session-factory
: A reference to the bean we configured earlier
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 is 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 to a local directory and then emits each file according to the poller configuration.
Starting with version 5.0, you can 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 large or when running in a clustered system with a persistent file list filter, discussed later in this section.
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 (the 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
, which lets
you dynamically determine the directory on each poll — for example, remote-directory-expression="@myBean.determineRemoteDir()"
.
Sometimes, file filtering based on the simple pattern specified via filename-pattern
attribute might not suffice.
If this is the case, you can use the filename-regex
attribute to specify a regular expression (for example, filename-regex=".*\.test$"
).
If you need complete control, you can use the filter
attribute to provide a reference to a custom implementation of the org.springframework.integration.file.filters.FileListFilter
, which is 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 SftpPersistentAcceptOnceFileListFilter
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
), this lets filter keys be shared across multiple application or server instances.
Starting with version 5.0, the SftpPersistentAcceptOnceFileListFilter
with an in-memory SimpleMetadataStore
is applied by default for the SftpInboundFileSynchronizer
.
This filter is also applied, together with the regex
or pattern
option in the XML configuration, as well as through FtpInboundChannelAdapterSpec
in Java DSL.
You can handle any other use-cases by using CompositeFileListFilter
(or ChainFileListFilter
).
The above 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 in this section, retains state in memory and does not consider the file’s modified time. Unless your application removes files after processing, the adapter re-processes the files on disk by default after an application restart.
Also, if you configure the filter
to use a SftpPersistentAcceptOnceFileListFilter
and the remote file timestamp changes (causing it to be re-fetched), the default local filter does not allow this new file to be processed.
For more information about this filter, and how it is used, see [remote-persistent-flf].
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
that stores state in memory.
Since version 4.1.5, these filters have a new property called flushOnUpdate
, which 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 Metadata Store or Gemfire Metadata Store), you can have multiple instances of the same adapter or application and be sure that one and only one instance processes a file.
|
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 files are renamed to their final names when the transfer is complete, making them 'visible' to the filter.
See the schema for more detail on these attributes.
SFTP inbound channel adapter is a polling consumer.
Therefore, you must configure a poller (either a global default or a local element).
Once the file has been transferred to a local directory, a message with java.io.File
as its payload type is generated and sent to the channel identified by the channel
attribute.
30.6.1. More on File Filtering and Large Files
Sometimes, a file that just appeared in the monitored (remote) directory is not complete.
Typically such a file is written with some temporary extension (such as .writing
on a file named something.txt.writing
) and then renamed after the writing process completes.
In most cases, developers are interested only in files that are complete and would like to filter only those files.
To handle these scenarios, you can use the filtering support provided by the filename-pattern
, filename-regex
, and filter
attributes.
If you need a custom filter implementation, you can include a reference in your adapter by setting the filter
attribute.
The following example shows how to do so:
<int-sftp:inbound-channel-adapter id="sftpInbondAdapter"
channel="receiveChannel"
session-factory="sftpSessionFactory"
filter="customFilter"
local-directory="file:/local-test-dir"
remote-directory="/remote-test-dir">
<int:poller fixed-rate="1000" max-messages-per-poll="10" task-executor="executor"/>
</int-sftp:inbound-channel-adapter>
<bean id="customFilter" class="org.foo.CustomFilter"/>
30.6.2. Recovering from Failures
You should understand the architecture of the adapter.
A file synchronizer fetches the files, and a FileReadingMessageSource
emits a message for each synchronized file.
As discussed earlier, two filters are involved.
The filter
attribute (and patterns) refers to the remote (SFTP) file list, to avoid fetching files that have already
been fetched.
the FileReadingMessageSource
uses the local-filter
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 applies only 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 a configuration similar to the following to facilitate the removal of the failed file from the filter:
<int-sftp:inbound-channel-adapter id="sftpAdapter"
session-factory="sftpSessionFactory"
channel="requestChannel"
remote-directory-expression="'/sftpSource'"
local-directory="file:myLocalDir"
auto-create-local-directory="true"
filename-pattern="*.txt">
<int:poller fixed-rate="1000">
<int:transactional synchronization-factory="syncFactory" />
</int:poller>
</int-sftp:inbound-channel-adapter>
<bean id="acceptOnceFilter"
class="org.springframework.integration.file.filters.AcceptOnceFileListFilter" />
<int:transaction-synchronization-factory id="syncFactory">
<int:after-rollback expression="payload.delete()" />
</int:transaction-synchronization-factory>
<bean id="transactionManager"
class="org.springframework.integration.transaction.PseudoTransactionManager" />
The preceding configuration works for any ResettableFileListFilter
.
Starting with version 5.0, the inbound channel adapter can build sub-directories locally, according 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 for any modifications in local directory.
The reprocessing sample shown earlier is based on the built-in functionality of the FileReadingMessageSource.WatchServiceDirectoryScanner
, which uses ResettableFileListFilter.remove()
when the file is deleted (StandardWatchEventKinds.ENTRY_DELETE
) from the local directory.
See WatchServiceDirectoryScanner
for more information.
30.6.3. Configuring with Java Configuration
The following Spring Boot application shows an example of how to configure the inbound adapter with Java:
@SpringBootApplication
public class SftpJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(SftpJavaApplication.class)
.web(false)
.run(args);
}
@Bean
public SessionFactory<LsEntry> sftpSessionFactory() {
DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
factory.setHost("localhost");
factory.setPort(port);
factory.setUser("foo");
factory.setPassword("foo");
factory.setAllowUnknownKeys(true);
factory.setTestSession(true);
return new CachingSessionFactory<LsEntry>(factory);
}
@Bean
public SftpInboundFileSynchronizer sftpInboundFileSynchronizer() {
SftpInboundFileSynchronizer fileSynchronizer = new SftpInboundFileSynchronizer(sftpSessionFactory());
fileSynchronizer.setDeleteRemoteFiles(false);
fileSynchronizer.setRemoteDirectory("foo");
fileSynchronizer.setFilter(new SftpSimplePatternFileListFilter("*.xml"));
return fileSynchronizer;
}
@Bean
@InboundChannelAdapter(channel = "sftpChannel", poller = @Poller(fixedDelay = "5000"))
public MessageSource<File> sftpMessageSource() {
SftpInboundFileSynchronizingMessageSource source =
new SftpInboundFileSynchronizingMessageSource(sftpInboundFileSynchronizer());
source.setLocalDirectory(new File("sftp-inbound"));
source.setAutoCreateLocalDirectory(true);
source.setLocalFilter(new AcceptOnceFileListFilter<File>());
source.setMaxFetchSize(1);
return source;
}
@Bean
@ServiceActivator(inputChannel = "sftpChannel")
public MessageHandler handler() {
return new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
System.out.println(message.getPayload());
}
};
}
}
30.6.4. 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 SftpJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(SftpJavaApplication.class)
.web(false)
.run(args);
}
@Bean
public IntegrationFlow sftpInboundFlow() {
return IntegrationFlows
.from(Sftp.inboundAdapter(this.sftpSessionFactory)
.preserveTimestamp(true)
.remoteDirectory("foo")
.regexFilter(".*\\.txt$")
.localFilenameExpression("#this.toUpperCase() + '.a'")
.localDirectory(new File("sftp-inbound")),
e -> e.id("sftpInboundAdapter")
.autoStartup(true)
.poller(Pollers.fixedDelay(5000)))
.handle(m -> System.out.println(m.getPayload()))
.get();
}
}
30.6.5. Dealing With Incomplete Data
The SftpSystemMarkerFilePresentFileListFilter
is provided to filter remote files that don’t have the corresponding marker file on the remote system.
See the Javadoc for configuration information.
30.7. SFTP Streaming Inbound Channel Adapter
Version 4.3 introduced the streaming inbound channel adapter.
This adapter produces message with payloads of type InputStream
, letting you fetch files 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 following example shows how to configure an SFTP streaming inbound channel adapter:
<int-sftp:inbound-streaming-channel-adapter id="ftpInbound"
channel="ftpChannel"
session-factory="sessionFactory"
filename-pattern="*.txt"
filename-regex=".*\.txt"
filter="filter"
filter-expression="@myFilterBean.check(#root)"
remote-file-separator="/"
comparator="comparator"
max-fetch-size="1"
remote-directory-expression="'foo/bar'">
<int:poller fixed-rate="1000" />
</int-sftp:inbound-streaming-channel-adapter>
You can use only one of filename-pattern
, filename-regex
, filter
, or filter-expression
.
Starting with version 5.0, by default, the SftpStreamingMessageSource adapter prevents duplicates for remote files by using SftpPersistentAcceptOnceFileListFilter based on the in-memory SimpleMetadataStore .
By default, this filter is also applied together with the filename pattern (or regex) as well.
If you need to allow duplicates, you can use the AcceptAllFileListFilter .
You can handle any other use cases by using CompositeFileListFilter (or ChainFileListFilter ).
The Java configuration shown later shows one technique to remove the remote file after processing, avoiding duplicates.
|
For more information about the SftpPersistentAcceptOnceFileListFilter
, and how it is used, see [remote-persistent-flf].
You can 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 the file name in headers (FileHeaders.REMOTE_DIRECTORY
and FileHeaders.REMOTE_FILE
, respectively).
Starting with version 5.0, the FileHeaders.REMOTE_FILE_INFO
header provides additional remote file information (in JSON).
If you set the fileInfoJson
property on the SftpStreamingMessageSource
to false
, the header contains an SftpFileInfo
object.
You can access the LsEntry
object provided by the underlying Jsch library by using the SftpFileInfo.getFileInfo()
method.
The fileInfoJson
property is not available when you use XML configuration, but you can set it by injecting the SftpStreamingMessageSource
into one of your configuration classes.
Starting with version 5.1, the generic type of the comparator
is LsEntry
.
Previously, it was AbstractFileInfo<LsEntry>
.
This is because the sort is now performed earlier in the processing, before filtering and applying maxFetch
.
30.7.1. Configuring with Java Configuration
The following Spring Boot application shows an example of how to configure the inbound adapter with Java:
@SpringBootApplication
public class SftpJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(SftpJavaApplication.class)
.web(false)
.run(args);
}
@Bean
@InboundChannelAdapter(channel = "stream")
public MessageSource<InputStream> ftpMessageSource() {
SftpStreamingMessageSource messageSource = new SftpStreamingMessageSource(template());
messageSource.setRemoteDirectory("sftpSource/");
messageSource.setFilter(new AcceptAllFileListFilter<>());
messageSource.setMaxFetchSize(1);
return messageSource;
}
@Bean
@Transformer(inputChannel = "stream", outputChannel = "data")
public org.springframework.integration.transformer.Transformer transformer() {
return new StreamTransformer("UTF-8");
}
@Bean
public SftpRemoteFileTemplate template() {
return new SftpRemoteFileTemplate(sftpSessionFactory());
}
@ServiceActivator(inputChannel = "data", adviceChain = "after")
@Bean
public MessageHandler handle() {
return System.out::println;
}
@Bean
public ExpressionEvaluatingRequestHandlerAdvice after() {
ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
advice.setOnSuccessExpression(
"@template.remove(headers['file_remoteDirectory'] + headers['file_remoteFile'])");
advice.setPropagateEvaluationFailures(true);
return advice;
}
}
Notice that, in this example, the message handler downstream of the transformer has an advice that removes the remote file after processing.
30.8. 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.
|
30.9. Inbound Channel Adapters: Controlling Remote File Fetching
You should consider two properties when configuring 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 a 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, and emit it. Then it sleeps until the next poll. -
max-messages-per-poll=2
andmax-fetch-size=2
): The adapter fetch both files and then emits each one. -
max-messages-per-poll=2
andmax-fetch-size=4
: The adapter fetches up to 4 files (if available) and emits the first two (if there are at least two). The next two files will be 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 are consumed, the remote fetch is attempted again, to pick up any new files.
When you deploy multiple instances of an application, we recommend setting a small max-fetch-size , to avoid one instance “grabbing” all the files and starving other instances.
|
Another use for max-fetch-size
is when 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, via JMX, or via 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<LsEntry>
.
This is useful when restricting the number of files fetched with maxFetchSize
.
30.10. SFTP Outbound Channel Adapter
The SFTP outbound channel adapter is a special MessageHandler
that connects to the remote directory and initiates a file transfer for every file it receives as the payload of an incoming Message
.
It also supports several representations of the file so that you are not limited to the File
object.
Similar to the FTP outbound adapter, the SFTP 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 SFTP outbound channel adapter:
<int-sftp:outbound-channel-adapter id="sftpOutboundAdapter"
session-factory="sftpSessionFactory"
channel="inputChannel"
charset="UTF-8"
remote-file-separator="/"
remote-directory="foo/bar"
remote-filename-generator-expression="payload.getName() + '-mysuffix'"
filename-generator="fileNameGenerator"
use-temporary-filename="true"
chmod="600"
mode="REPLACE"/>
See the schema for more detail on these attributes.
30.10.1. SpEL and the SFTP Outbound Adapter
As with many other components in Spring Integration, you can use the Spring Expression Language (SpEL) when you configure an SFTP outbound channel adapter by specifying two attributes: remote-directory-expression
and remote-filename-generator-expression
(described earlier).
The expression evaluation context has the message as its root object, which lets you use expressions that can dynamically compute the file name or the existing directory path based on the data in the message (from either the 'payload' or the 'headers').
In the preceding example, we define the remote-filename-generator-expression
attribute with an expression value that computes the file name based on its original name while also appending a suffix: '-mysuffix'.
Starting with version 4.1, you can specify the mode
when you transferring the file.
By default, an existing file is overwritten.
The modes are defined by the FileExistsMode
enumeration, which has the following values: REPLACE
(default), APPEND
, IGNORE
, and FAIL
.
With IGNORE
and FAIL
, the file is not transferred.
FAIL
causes an exception to be thrown, while IGNORE
silently ignores the transfer (although a DEBUG
log entry is produced).
30.10.2. Avoiding Partially Written Files
One of the common problems when dealing with file transfers is the possibility of processing a partial file. A file might appear in the file system before its transfer is actually complete.
To deal with this issue, Spring Integration SFTP adapters use a common algorithm in which files are transferred under a temporary name and than renamed once they are fully transferred.
By default, every file that is in the process of being transferred appear in the file system with an additional suffix, which, by default, is .writing
.
You can change 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.
Version 4.3 introduced the chmod
attribute, which you can use to change the remote file permissions after upload.
You can use the conventional Unix octal format (for example, 600
allows read-write for the file owner only).
When configuring the adapter using java, you can use setChmodOctal("600")
or setChmodDecimal(384)
.
30.10.3. Configuring with Java Configuration
The following Spring Boot application shows an example of how to configure the outbound adapter with Java:
@SpringBootApplication
@IntegrationComponentScan
public class SftpJavaApplication {
public static void main(String[] args) {
ConfigurableApplicationContext context =
new SpringApplicationBuilder(SftpJavaApplication.class)
.web(false)
.run(args);
MyGateway gateway = context.getBean(MyGateway.class);
gateway.sendToSftp(new File("/foo/bar.txt"));
}
@Bean
public SessionFactory<LsEntry> sftpSessionFactory() {
DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
factory.setHost("localhost");
factory.setPort(port);
factory.setUser("foo");
factory.setPassword("foo");
factory.setAllowUnknownKeys(true);
factory.setTestSession(true);
return new CachingSessionFactory<LsEntry>(factory);
}
@Bean
@ServiceActivator(inputChannel = "toSftpChannel")
public MessageHandler handler() {
SftpMessageHandler handler = new SftpMessageHandler(sftpSessionFactory());
handler.setRemoteDirectoryExpressionString("headers['remote-target-dir']");
handler.setFileNameGenerator(new FileNameGenerator() {
@Override
public String generateFileName(Message<?> message) {
return "handlerContent.test";
}
});
return handler;
}
@MessagingGateway
public interface MyGateway {
@Gateway(requestChannel = "toSftpChannel")
void sendToSftp(File file);
}
}
30.10.4. Configuring with the Java DSL
The following Spring Boot application shows an example of how to configure the outbound adapter with the Java DSL:
@SpringBootApplication
public class SftpJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(SftpJavaApplication.class)
.web(false)
.run(args);
}
@Bean
public IntegrationFlow sftpOutboundFlow() {
return IntegrationFlows.from("toSftpChannel")
.handle(Sftp.outboundAdapter(this.sftpSessionFactory, FileExistsMode.FAIL)
.useTemporaryFileName(false)
.remoteDirectory("/foo")
).get();
}
}
30.11. SFTP Outbound Gateway
The SFTP outbound gateway provides a limited set of commands that let you interact with a remote SFTP server:
-
ls
(list files) -
nlst
(list file names) -
get
(retrieve a file) -
mget
(retrieve multiple files) -
rm
(remove file(s)) -
mv
(move and rename file) -
put
(send a file) -
mput
(send multiple files)
30.11.1. Using the ls
Command
ls
lists remote files and supports the following options:
-
-1
: Retrieve a list of filenames. 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 (excluded by default) -
-links
: Include symbolic links (excluded by default) -
-R
: List the remote directory recursively
In addition, filename filtering is provided in the same manner as the inbound-channel-adapter
.
The message payload resulting from an ls
operation is a list of file names or a list of FileInfo
objects (depending on whether you usr the -1
switch).
These objects provide information such as modified time, permissions, and others.
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 and represents the relative path to the file (relative to the remote directory).
If you use the -dirs
option, each recursive directory is also returned as an element in the list.
In this case, we recommend that you not use the -1
option, because you would not be able to distinguish files from directories, which you can do when you use FileInfo
objects.
Using 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 file_remoteDirectory
header holds the remote directory on which the nlst
command acted.
The SFTP protocol does not provide the ability to list names.
This command is the equivalent of the ls
command with the -1
option and is added here for convenience.
30.11.2. Using the get
Command
get
retrieves a remote file and supports the following options:
-
-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 holds the remote directory, and the file_remoteFile
header holds the filename.
The message payload resulting from a get
operation is a File
object representing the retrieved file.
If you use the -stream
option, the payload is an InputStream
rather than a File
.
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, and IntegrationMessageHeaderAccessor
offers convenience method:
Closeable closeable = new IntegrationMessageHeaderAccessor(message).getCloseableResource();
if (closeable != null) {
closeable.close();
}
Framework components, such as the File Splitter and Stream Transformer, automatically close the session after the data is transferred.
The following example shows how to consume a file as a stream:
<int-sftp:outbound-gateway session-factory="ftpSessionFactory"
request-channel="inboundGetStream"
command="get"
command-options="-stream"
expression="payload"
remote-directory="ftpTarget"
reply-channel="stream" />
<int-file:splitter input-channel="stream" output-channel="lines" />
If you consume the input stream in a custom component, you must close the Session .
You can either do that in your custom code or route a copy of the message to a service-activator and use SpEL, as the following example shows:
|
<int:service-activator input-channel="closeSession"
expression="headers['closeableResource'].close()" />
30.11.3. 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. If the transfer is ignored, the remote file is not deleted, 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 contain files that were not fetched due to the file already existing.
Previously, the array contained all files, including those that already existed.
|
The expression you use determine the remote path should produce a result that ends with for example
myfiles/
fetches the complete tree under myfiles
.
Starting with version 5.0, you can use a recursive MGET
, combined with the FileExistsMode.REPLACE_IF_MODIFIED
mode, to periodically synchronize an entire remote directory tree locally.
This mode sets the local file’s last modified timestamp to the remote file’s timestamp, regardless of the -P
(preserve timestamp) option.
|
Notes for when using recursion (
-R )The pattern is ignored and If you filter a subdirectory, no additional traversal of that subdirectory is performed. The Typically, you would use the |
Starting with version 5.0, you can configure the SftpSimplePatternFileListFilter
and SftpRegexPatternFileListFilter
to always pass directories by setting the alwaysAcceptDirectorties
to true
.
Doing so allows recursion for a simple pattern, as the following examples show:
<bean id="starDotTxtFilter"
class="org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter">
<constructor-arg value="*.txt" />
<property name="alwaysAcceptDirectories" value="true" />
</bean>
<bean id="dotStarDotTxtFilter"
class="org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter">
<constructor-arg value="^.*\.txt$" />
<property name="alwaysAcceptDirectories" value="true" />
</bean>
You can provide one of these filters by using the filter
property on the gateway.
30.11.4. Using the put
Command
put
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 contains the full path of the file on the server after transfer.
Version 4.3 introduced the chmod
attribute, which changes the remote file permissions after upload.
You can use the conventional Unix octal format (for example, 600
allows read-write for the file owner only).
When configuring the adapter using java, you can use setChmod(0600)
.
30.11.5. Using the mput
Command
mput
sends multiple files to the server and supports the following option:
-
-R
: Recursive — send all files (possibly filtered) in the directory and 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.
The same attributes as the put
command are supported.
In addition, you can filter files in the local directory 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 resulting from the transfer).
Version 4.3 introduced the chmod
attribute, which lets you change the remote file permissions after upload.
You can use the conventional Unix octal format (for example, 600
allows read-write for the file owner only).
When configuring the adapter with Java, you can use setChmodOctal("600")
or setChmodDecimal(384)
.
Using the rm
Command
The rm
command has no options.
If the remove operation was successful, the resulting message payload is Boolean.TRUE
.
Otherwise, the message payload is Boolean.FALSE
.
The file_remoteDirectory
header holds the remote directory, and the file_remoteFile
header holds the file name.
30.11.6. Using the mv
Command
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 remote directories needed are created.
The payload of the result message is Boolean.TRUE
.
The the file_remoteDirectory
header holds the original remote directory, and the file_remoteFile
header holds the filename.
The file_renameTo
header holds the new path.
30.11.7. Additional Command Information
The get
and mget
commands support the local-filename-generator-expression
attribute.
It defines a SpEL expression to generate the names of local files during the transfer.
The root object of the evaluation context is the request message.
The remoteFileName
variable is also available.
It is particularly useful for mget
(for example: local-filename-generator-expression="#remoteFileName.toUpperCase() + headers.foo"
).
The get
and mget
commands support the local-directory-expression
attribute.
It defines a SpEL expression to generate the names of local directories during the transfer.
The root object of the evaluation context is the request message.
The remoteDirectory
variable is also available.
It is particularly useful for mget (for example: local-directory-expression="'/tmp/local/' + #remoteDirectory.toUpperCase() + headers.myheader"
).
This attribute is mutually exclusive with the local-directory
attribute.
For all commands, the 'expression' property of the gateway holds the path on which the command acts.
For the mget
command, the expression might evaluate to '', meaning to retrieve all files, 'somedirectory/', and other values that end with *
.
The following example shows a gateway configured for an ls
command:
<int-ftp:outbound-gateway id="gateway1"
session-factory="ftpSessionFactory"
request-channel="inbound1"
command="ls"
command-options="-1"
expression="payload"
reply-channel="toSplitter"/>
The payload of the message sent to the toSplitter
channel is a list of String
objects, each of which contains the name of a file.
If you omitted command-options="-1"
, the payload would be a list of FileInfo
objects.
You can provide options as a space-delimited list (for example, command-options="-1 -dirs -links"
).
Starting with version 4.2, the GET
, MGET
, PUT
, and MPUT
commands support a FileExistsMode
property (mode
when using the namespace support).
This affects the behavior when the local file exists (GET
and MGET
) or the remote file exists (PUT
and MPUT
).
The supported modes are REPLACE
, APPEND
, FAIL
, and IGNORE
.
For backwards compatibility, the default mode for PUT
and MPUT
operations is REPLACE
.
For GET
and MGET
operations, the default is FAIL
.
30.11.8. Configuring with Java Configuration
The following Spring Boot application shows an example of how to configure the outbound gateway with Java:
@SpringBootApplication
public class SftpJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(SftpJavaApplication.class)
.web(false)
.run(args);
}
@Bean
@ServiceActivator(inputChannel = "sftpChannel")
public MessageHandler handler() {
return new SftpOutboundGateway(ftpSessionFactory(), "ls", "'my_remote_dir/'");
}
}
30.11.9. Configuring with the Java DSL
The following Spring Boot application shows an example of how to configure the outbound gateway with the Java DSL:
@SpringBootApplication
public class SftpJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(SftpJavaApplication.class)
.web(false)
.run(args);
}
@Bean
public SessionFactory<LsEntry> sftpSessionFactory() {
DefaultFtpSessionFactory sf = new DefaultFtpSessionFactory();
sf.setHost("localhost");
sf.setPort(port);
sf.setUsername("foo");
sf.setPassword("foo");
factory.setTestSession(true);
return new CachingSessionFactory<LsEntry>(sf);
}
@Bean
public QueueChannelSpec remoteFileOutputChannel() {
return MessageChannels.queue();
}
@Bean
public IntegrationFlow sftpMGetFlow() {
return IntegrationFlows.from("sftpMgetInputChannel")
.handle(Sftp.outboundGateway(sftpSessionFactory(),
AbstractRemoteFileOutboundGateway.Command.MGET, "payload")
.options(AbstractRemoteFileOutboundGateway.Option.RECURSIVE)
.regexFileNameFilter("(subSftpSource|.*1.txt)")
.localDirectoryExpression("'myDir/' + #remoteDirectory")
.localFilenameExpression("#remoteFileName.replaceFirst('sftpSource', 'localTarget')"))
.channel("remoteFileOutputChannel")
.get();
}
}
30.11.10. Outbound Gateway Partial Success (mget
and mput
)
When performing operations on multiple files (by using mget
and mput
) an exception can occur some time after one or more files have been transferred.
In this case (starting with version 4.2), a PartialSuccessException
is thrown.
As well as the usual MessagingException
properties (failedMessage
and cause
), this exception has two additional
properties:
-
partialResults
: The successful transfer results. -
derivedInput
: The list of files generated from the request message (such as local files to transfer for anmput
).
These attributes let you determine which files were successfully transferred and which were not.
In the case of a recursive mput
, the PartialSuccessException
may have nested PartialSuccessException
instances.
Consider the following directory structure:
root/
|- file1.txt
|- subdir/
| - file2.txt
| - file3.txt
|- zoo.txt
If the exception occurs on file3.txt
, the PartialSuccessException
thrown by the gateway has derivedInput
of file1.txt
, subdir
, and zoo.txt
and partialResults
of file1.txt
.
Its cause
is another PartialSuccessException
with derivedInput
of file2.txt
and file3.txt
and partialResults
of file2.txt
.
30.12. SFTP/JSCH Logging
Since we use JSch libraries (www.jcraft.com/jsch/) to provide SFTP support, you may at times require more information from the JSch API itself, especially if something is not working properly (such as authentication exceptions).
Unfortunately JSch does not use commons-logging
but instead relies on custom implementations of their com.jcraft.jsch.Logger
interface.
As of Spring Integration 2.0.1, we have implemented this interface.
So now, to enable JSch logging, you can configure your logger the way you usually do.
For example, the following example is valid configuration of a logger that uses Log4J:
log4j.category.com.jcraft.jsch=DEBUG
30.13. MessageSessionCallback
Starting with Spring Integration version 4.2, you can use a MessageSessionCallback<F, T>
implementation with the <int-sftp:outbound-gateway/>
(SftpOutboundGateway
) to perform any operation on the Session<LsEntry>
with the requestMessage
context.
You can use it for any non-standard or low-level FTP operation (or several), such as allowing access from an integration flow definition, or functional interface (lambda) implementation injection.
The following example uses a lambda:
@Bean
@ServiceActivator(inputChannel = "sftpChannel")
public MessageHandler sftpOutboundGateway(SessionFactory<ChannelSftp.LsEntry> sessionFactory) {
return new SftpOutboundGateway(sessionFactory,
(session, requestMessage) -> session.list(requestMessage.getPayload()));
}
Another example might be to pre- or post-process the file data being sent or retrieved.
When using XML configuration, the <int-sftp:outbound-gateway/>
provides a session-callback
attribute that lets you specify the MessageSessionCallback
bean name.
The session-callback is mutually exclusive with the command and expression attributes.
When configuring with Java, the SftpOutboundGateway class offers different constructors.
|
31. STOMP Support
Spring Integration version 4.2 introduced STOMP (Simple Text Orientated Messaging Protocol) client support.
It is based on the architecture, infrastructure, and API from the Spring Framework’s messaging module, stomp package.
Spring Integration uses many of Spring STOMP components (such as StompSession
and StompClientSupport
).
For more information, see the Spring Framework STOMP Support chapter in the Spring Framework reference manual.
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-stomp</artifactId>
<version>5.1.5.RELEASE</version>
</dependency>
compile "org.springframework.integration:spring-integration-stomp:5.1.5.RELEASE"
For server side components you need to add a org.springframework:spring-websocket
and/or io.projectreactor.netty:reactor-netty
dependencies.
31.1. Overview
To configure STOMP, you should start with the STOMP client object. The Spring Framework provides the following implementations:
-
WebSocketStompClient
: Built on the Spring WebSocket API with support for standard JSR-356 WebSocket, Jetty 9, and SockJS for HTTP-based WebSocket emulation with SockJS Client. -
ReactorNettyTcpStompClient
: Built onReactorNettyTcpClient
from thereactor-netty
project.
You can provide any other StompClientSupport
implementation.
See the Javadoc of those classes.
The StompClientSupport
class is designed as a factory to produce a StompSession
for the provided StompSessionHandler
and all the remaining work is done through the callbacks to that StompSessionHandler
and StompSession
abstraction.
With the Spring Integration adapter abstraction, we need to provide some managed shared object to represent our application as a STOMP client with its unique session.
For this purpose, Spring Integration provides the StompSessionManager
abstraction to manage the single StompSession
between any provided StompSessionHandler
.
This allows the use of inbound or outbound channel adapters (or both) for the particular STOMP Broker.
See StompSessionManager
(and its implementations) JavaDocs for more information.
31.2. STOMP Inbound Channel Adapter
The StompInboundChannelAdapter
is a one-stop MessageProducer
component that subscribes your Spring Integration application to the provided STOMP destinations and receives messages from them (converted from the STOMP frames by using the provided MessageConverter
on the connected StompSession
).
You can change the destinations (and therefore STOMP subscriptions) at runtime by using appropriate @ManagedOperation
annotations on the StompInboundChannelAdapter
.
For more configuration options, see STOMP Namespace Support and the StompInboundChannelAdapter
Javadoc.
31.3. STOMP Outbound Channel Adapter
The StompMessageHandler
is the MessageHandler
for the <int-stomp:outbound-channel-adapter>
and is used to send the outgoing Message<?>
instances to the STOMP destination
(pre-configured or determined at runtime with a SpEL expression) through the StompSession
(which is provided by the shared StompSessionManager
).
For more configuration options see STOMP Namespace Support and the StompMessageHandler
Javadoc.
31.4. STOMP Headers Mapping
The STOMP protocol provides headers as part of its frame. The entire structure of the STOMP frame has the following format:
.... COMMAND header1:value1 header2:value2 Body^@ ....
Spring Framework provides StompHeaders
to represent these headers.
See the Javadoc for more details.
STOMP frames are converted to and from Message<?>
instances and these headers are mapped to and from MessageHeaders
instances.
Spring Integration provides a default HeaderMapper
implementation for the STOMP adapters.
The implementation is StompHeaderMapper
.
It provides fromHeaders()
and toHeaders()
operations for the inbound and outbound adapters, respectively.
As with many other Spring Integration modules, the IntegrationStompHeaders
class has been introduced to map standard STOMP headers to MessageHeaders
, with stomp_
as the header name prefix.
In addition, all MessageHeaders
instances with that prefix are mapped to the StompHeaders
when sending to a destination.
For more information, see the Javadoc for those classes and the mapped-headers
attribute description in the STOMP Namespace Support.
31.5. STOMP Integration Events
Many STOMP operations are asynchronous, including error handling.
For example, STOMP has a RECEIPT
server frame that it returns when a client frame has requested one by adding the RECEIPT
header.
To provide access to these asynchronous events, Spring Integration emits StompIntegrationEvent
instances, which you can obtain by implementing an ApplicationListener
or by using an <int-event:inbound-channel-adapter>
(see Receiving Spring Application Events).
Specifically, a StompExceptionEvent
is emitted from the AbstractStompSessionManager
when a stompSessionListenableFuture
receives onFailure()
due to failure to connect to STOMP broker.
Another example is the StompMessageHandler
.
It processes ERROR
STOMP frames, which are server responses to improper (unaccepted) messages sent by this StompMessageHandler
.
The StompMessageHandler
emits StompReceiptEvent
as a part of StompSession.Receiptable
callbacks in the asynchronous answers for the messages sent to the StompSession
.
The StompReceiptEvent
can be positive or negative, depending on whether or not the RECEIPT
frame was received from the server within the receiptTimeLimit
period, which you can configure on the StompClientSupport
instance.
It defaults to 15 * 1000
(in milliseconds, so 15 seconds).
The StompSession.Receiptable callbacks are added only if the RECEIPT STOMP header of the message to send is not null .
You can enable automatic RECEIPT header generation on the StompSession through its autoReceipt option and on the StompSessionManager respectively.
|
See STOMP Adapters Java Configuration for more information how to configure Spring Integration to accept those ApplicationEvent
instances.
31.6. STOMP Adapters Java Configuration
The following example shows a comprehensive Java configuration for STOMP adapters:
@Configuration
@EnableIntegration
public class StompConfiguration {
@Bean
public ReactorNettyTcpStompClient stompClient() {
ReactorNettyTcpStompClient stompClient = new ReactorNettyTcpStompClient("<IP_ADDRESS>", 61613);
stompClient.setMessageConverter(new PassThruMessageConverter());
ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
taskScheduler.afterPropertiesSet();
stompClient.setTaskScheduler(taskScheduler);
stompClient.setReceiptTimeLimit(5000);
return stompClient;
}
@Bean
public StompSessionManager stompSessionManager() {
ReactorNettyTcpStompSessionManager stompSessionManager = new ReactorNettyTcpStompSessionManager(stompClient());
stompSessionManager.setAutoReceipt(true);
return stompSessionManager;
}
@Bean
public PollableChannel stompInputChannel() {
return new QueueChannel();
}
@Bean
public StompInboundChannelAdapter stompInboundChannelAdapter() {
StompInboundChannelAdapter adapter =
new StompInboundChannelAdapter(stompSessionManager(), "/topic/myTopic");
adapter.setOutputChannel(stompInputChannel());
return adapter;
}
@Bean
@ServiceActivator(inputChannel = "stompOutputChannel")
public MessageHandler stompMessageHandler() {
StompMessageHandler handler = new StompMessageHandler(stompSessionManager());
handler.setDestination("/topic/myTopic");
return handler;
}
@Bean
public PollableChannel stompEvents() {
return new QueueChannel();
}
@Bean
public ApplicationListener<ApplicationEvent> stompEventListener() {
ApplicationEventListeningMessageProducer producer = new ApplicationEventListeningMessageProducer();
producer.setEventTypes(StompIntegrationEvent.class);
producer.setOutputChannel(stompEvents());
return producer;
}
}
31.7. STOMP Namespace Support
The Spring Integration STOMP namespace implements the inbound and outbound channel adapter components. To include it in your configuration, provide the following namespace declaration in your application context configuration file:
<?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-stomp="http://www.springframework.org/schema/integration/stomp"
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/stomp
https://www.springframework.org/schema/integration/stomp/spring-integration-stomp.xsd">
...
</beans>
31.7.1. Understanding the <int-stomp:outbound-channel-adapter>
Element
The following listing shows the available attributes for the STOMP outbound channel adapter:
<int-stomp:outbound-channel-adapter
id="" (1)
channel="" (2)
stomp-session-manager="" (3)
header-mapper="" (4)
mapped-headers="" (5)
destination="" (6)
destination-expression="" (7)
auto-startup="" (8)
phase=""/> (9)
| 1 | The component bean name.
The MessageHandler is registered with a bean alias of id plus .handler .
If you do not set the channel attribute, a DirectChannel is created and registered in the application context with the value of this id attribute as the bean name.
In this case, the endpoint is registered with a bean name id plus .adapter . |
| 2 | Identifies the channel attached to this adapter if id is present.
See id .
Optional. |
| 3 | Reference to a StompSessionManager bean, which encapsulates the low-level connection and StompSession handling operations.
Required. |
| 4 | Reference to a bean that implements HeaderMapper<StompHeaders> , which maps Spring Integration MessageHeaders to and from
STOMP frame headers.
It is mutually exclusive with mapped-headers .
It defaults to StompHeaderMapper . |
| 5 | Comma-separated list of names of STOMP Headers to be mapped to the STOMP frame headers.
It can be provided only if the header-mapper reference is not set.
The values in this list can also be simple patterns to be matched against the header names (such as myheader* or *myheader ).
A special token (STOMP_OUTBOUND_HEADERS ) represents all the standard STOMP headers (content-length, receipt, heart-beat, and so on).
They are included by default.
If you want to add your own headers and want the standard headers to also be mapped, you must include this token or provide your own HeaderMapper implementation by using header-mapper . |
| 6 | Name of the destination to which STOMP Messages are sent.
It is mutually exclusive with the destination-expression . |
| 7 | A SpEL expression to be evaluated at runtime against each Spring Integration Message as the root object.
It is mutually exclusive with the destination . |
| 8 | Boolean value indicating whether this endpoint should start automatically.
It defaults to true . |
| 9 | The lifecycle phase within which this endpoint should start and stop.
The lower the value, the earlier this endpoint starts and the later it stops.
The default is Integer.MIN_VALUE .
Values can be negative.
See SmartLifeCycle . |
31.7.2. Understanding the <int-stomp:inbound-channel-adapter>
Element
The following listing shows the available attributes for the STOMP inbound channel adapter:
<int-stomp:inbound-channel-adapter
id="" (1)
channel="" (2)
error-channel="" (3)
stomp-session-manager="" (4)
header-mapper="" (5)
mapped-headers="" (6)
destinations="" (7)
send-timeout="" (8)
payload-type="" (9)
auto-startup="" (10)
phase=""/> (11)
| 1 | The component bean name.