From 93bcfdbc55622f2d9606dc193b30fd5784409f73 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 6 Nov 2013 11:52:31 +0200 Subject: [PATCH 1/5] INT-3198: Add 'recovery-interval' for redis-q-c-a JIRA: https://jira.springsource.org/browse/INT-3198 * Expose `RedisQueueMessageDrivenEndpoint.recoveryInterval` to Namespace support * Add parser's tests * Rename parser test class * Polishing for concurrency of`RedisQueueMessageDrivenEndpointTests#testInt3196Recovery` test * Add documentation for 'recovery-interval' and Redis Events implementation Doc Polishing Add more delay because of unsolvable race condition in the Redis Subscription - see DATAREDIS-242 --- ...RedisQueueInboundChannelAdapterParser.java | 1 + .../config/spring-integration-redis-3.0.xsd | 9 +++++++ ...oundChannelAdapterParserTests-context.xml} | 1 + ...ueueInboundChannelAdapterParserTests.java} | 4 ++- .../RedisQueueMessageDrivenEndpointTests.java | 22 ++++++++++++--- .../redis/rules/RedisAvailableTests.java | 2 ++ src/reference/docbook/redis.xml | 27 ++++++++++++++++++- 7 files changed, 61 insertions(+), 5 deletions(-) rename spring-integration-redis/src/test/java/org/springframework/integration/redis/config/{RedisMessageDrivenEndpointParserTests-context.xml => RedisQueueInboundChannelAdapterParserTests-context.xml} (98%) rename spring-integration-redis/src/test/java/org/springframework/integration/redis/config/{RedisMessageDrivenEndpointParserTests.java => RedisQueueInboundChannelAdapterParserTests.java} (94%) diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParser.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParser.java index 18996384da..401b413dea 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParser.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParser.java @@ -50,6 +50,7 @@ public class RedisQueueInboundChannelAdapterParser extends AbstractChannelAdapte IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "expect-message"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "receive-timeout"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "recovery-interval"); builder.addPropertyReference("outputChannel", channelName); return builder.getBeanDefinition(); diff --git a/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-3.0.xsd b/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-3.0.xsd index 450916967e..b6ac1e088d 100644 --- a/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-3.0.xsd +++ b/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-3.0.xsd @@ -392,6 +392,15 @@ + + + + Specify the time in milliseconds for which the listener task should sleep after catching + an Exception on a Redis operation, before restarting the listener task. + Default is 5 seconds. + + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisMessageDrivenEndpointParserTests-context.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests-context.xml similarity index 98% rename from spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisMessageDrivenEndpointParserTests-context.xml rename to spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests-context.xml index 6c7558adf3..5700eee8ef 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisMessageDrivenEndpointParserTests-context.xml +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests-context.xml @@ -28,6 +28,7 @@ serializer="serializer" error-channel="errorChannel" receive-timeout="2000" + recovery-interval="3000" task-executor="executor" auto-startup="false" phase="100"/> diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisMessageDrivenEndpointParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests.java similarity index 94% rename from spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisMessageDrivenEndpointParserTests.java rename to spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests.java index 56ba70bba6..31cad85045 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisMessageDrivenEndpointParserTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests.java @@ -46,7 +46,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; */ @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) -public class RedisMessageDrivenEndpointParserTests { +public class RedisQueueInboundChannelAdapterParserTests { @Autowired @Qualifier("redisConnectionFactory") @@ -90,6 +90,7 @@ public class RedisMessageDrivenEndpointParserTests { assertEquals("si.test.Int3017.Inbound1", TestUtils.getPropertyValue(this.defaultAdapter, "boundListOperations.key")); assertFalse(TestUtils.getPropertyValue(this.defaultAdapter, "expectMessage", Boolean.class)); assertEquals(new Long(1000), TestUtils.getPropertyValue(this.defaultAdapter, "receiveTimeout", Long.class)); + assertEquals(new Long(5000), TestUtils.getPropertyValue(this.defaultAdapter, "recoveryInterval", Long.class)); assertNull(TestUtils.getPropertyValue(this.defaultAdapter, "errorChannel")); assertThat(TestUtils.getPropertyValue(this.defaultAdapter, "taskExecutor"), Matchers.instanceOf(ErrorHandlingTaskExecutor.class)); assertThat(TestUtils.getPropertyValue(this.defaultAdapter, "serializer"), Matchers.instanceOf(JdkSerializationRedisSerializer.class)); @@ -103,6 +104,7 @@ public class RedisMessageDrivenEndpointParserTests { assertEquals("si.test.Int3017.Inbound2", TestUtils.getPropertyValue(this.customAdapter, "boundListOperations.key")); assertTrue(TestUtils.getPropertyValue(this.customAdapter, "expectMessage", Boolean.class)); assertEquals(new Long(2000), TestUtils.getPropertyValue(this.customAdapter, "receiveTimeout", Long.class)); + assertEquals(new Long(3000), TestUtils.getPropertyValue(this.customAdapter, "recoveryInterval", Long.class)); assertSame(this.errorChannel, TestUtils.getPropertyValue(this.customAdapter, "errorChannel")); assertSame(this.taskExecutor, TestUtils.getPropertyValue(this.customAdapter, "taskExecutor")); assertSame(this.serializer, TestUtils.getPropertyValue(this.customAdapter, "serializer")); diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java index ec28d11af0..638067ce99 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java @@ -20,12 +20,15 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; import java.util.ArrayList; import java.util.Arrays; import java.util.Date; import java.util.List; import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import org.hamcrest.Matchers; import org.junit.Test; @@ -213,6 +216,8 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { final List exceptionEvents = new ArrayList(); + final CountDownLatch exceptionsLatch = new CountDownLatch(2); + RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName, this.connectionFactory); endpoint.setBeanFactory(Mockito.mock(BeanFactory.class)); endpoint.setApplicationEventPublisher(new ApplicationEventPublisher() { @@ -220,6 +225,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { @Override public void publishEvent(ApplicationEvent event) { exceptionEvents.add(event); + exceptionsLatch.countDown(); } }); endpoint.setOutputChannel(channel); @@ -228,16 +234,26 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { endpoint.afterPropertiesSet(); endpoint.start(); + int n = 0; + do { + n++; + if (n == 100) { + break; + } + Thread.sleep(100); + } while (!endpoint.isListening()); + + assertTrue(n < 100); + ((DisposableBean) this.connectionFactory).destroy(); - Thread.sleep(300); + assertTrue(exceptionsLatch.await(10, TimeUnit.SECONDS)); - assertThat(exceptionEvents.size(), Matchers.greaterThan(0)); for (ApplicationEvent exceptionEvent : exceptionEvents) { assertThat(exceptionEvent, Matchers.instanceOf(RedisExceptionEvent.class)); assertSame(endpoint, exceptionEvent.getSource()); assertThat(((IntegrationEvent) exceptionEvent).getCause().getClass(), - Matchers.isIn(Arrays.> asList(RedisSystemException.class, RedisConnectionFailureException.class))); + Matchers.isIn(Arrays.>asList(RedisSystemException.class, RedisConnectionFailureException.class))); } ((InitializingBean) this.connectionFactory).afterPropertiesSet(); diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java index 4adb288a48..b476dc7780 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java @@ -68,6 +68,8 @@ public class RedisAvailableTests { while (n++ < 100 && !connection.isSubscribed()) { Thread.sleep(100); } + // TODO: remove this additional delay when/if https://jira.springsource.org/browse/DATAREDIS-242 is resolved + Thread.sleep(250); assertTrue("RedisMessageListenerContainer Failed to Subscribe", n < 100); } diff --git a/src/reference/docbook/redis.xml b/src/reference/docbook/redis.xml index 1bd7c2023a..ff6c822d9e 100644 --- a/src/reference/docbook/redis.xml +++ b/src/reference/docbook/redis.xml @@ -205,6 +205,7 @@ rt.setConnectionFactory(redisConnectionFactory);]]> error-channel="" ]]> ]]> @@ -247,7 +248,9 @@ rt.setConnectionFactory(redisConnectionFactory);]]> The MessageChannel to which to send ErrorMessages with - Exceptions from the listening task of the Endpoint. + Exceptions from the listening task of the Endpoint. By default + the underlying MessagePublishingErrorHandler uses the + default errorChannel from the application context. @@ -262,6 +265,12 @@ rt.setConnectionFactory(redisConnectionFactory);]]> The timeout in milliseconds for 'right pop' operation to wait for a Redis message from the queue. Default is 1 second. + + + The time in milliseconds for which the listener task should sleep after exceptions on the 'right pop' operation, + before restarting the listener task. + + Specify if this Endpoint expects data from the Redis queue to contain entire Messages. @@ -344,6 +353,22 @@ rt.setConnectionFactory(redisConnectionFactory);]]> +
+ 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 an 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 using an <int-event:inbound-channel-adapter/> can be useful to determine + problems with background Redis tasks and to take administrative actions. + +
From b1b78409c89b9d2362747c95844bcb9aa6c44c11 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 6 Nov 2013 18:11:18 +0200 Subject: [PATCH 2/5] INT-2946: Add redis:i-c-a `topic-patterns` Attr. JIRA: https://jira.springsource.org/browse/INT-2946 Polishing --- .../RedisInboundChannelAdapterParser.java | 1 + .../inbound/RedisInboundChannelAdapter.java | 36 ++++++++++++++++--- .../config/spring-integration-redis-3.0.xsd | 7 ++++ ...boundChannelAdapterParserTests-context.xml | 8 ++--- ...RedisInboundChannelAdapterParserTests.java | 20 ++++++++--- .../redis/rules/RedisAvailableTests.java | 15 ++++++++ src/reference/docbook/redis.xml | 11 ++++-- src/reference/docbook/whats-new.xml | 4 +++ 8 files changed, 87 insertions(+), 15 deletions(-) diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParser.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParser.java index 13aab72337..2a22976734 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParser.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParser.java @@ -45,6 +45,7 @@ public class RedisInboundChannelAdapterParser extends AbstractChannelAdapterPars builder.addConstructorArgReference(connectionFactory); builder.addPropertyReference("outputChannel", channelName); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "topics"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "topic-patterns"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "serializer", true); diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java index d1c7d17dc2..e0f65a29c3 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java @@ -21,7 +21,9 @@ import java.util.List; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.listener.ChannelTopic; +import org.springframework.data.redis.listener.PatternTopic; import org.springframework.data.redis.listener.RedisMessageListenerContainer; +import org.springframework.data.redis.listener.Topic; import org.springframework.data.redis.listener.adapter.MessageListenerAdapter; import org.springframework.data.redis.serializer.RedisSerializer; import org.springframework.data.redis.serializer.StringRedisSerializer; @@ -35,6 +37,7 @@ import org.springframework.util.Assert; * @author Mark Fisher * @author Oleg Zhurakousky * @author Gary Russell + * @author Artem Bilan * @since 2.1 */ public class RedisInboundChannelAdapter extends MessageProducerSupport { @@ -45,6 +48,8 @@ public class RedisInboundChannelAdapter extends MessageProducerSupport { private volatile String[] topics; + private volatile String[] topicPatterns; + private volatile RedisSerializer serializer = new StringRedisSerializer(); public RedisInboundChannelAdapter(RedisConnectionFactory connectionFactory) { @@ -60,6 +65,10 @@ public class RedisInboundChannelAdapter extends MessageProducerSupport { this.topics = topics; } + public void setTopicPatterns(String... topicPatterns) { + this.topicPatterns = topicPatterns; + } + public void setMessageConverter(MessageConverter messageConverter) { Assert.notNull(messageConverter, "messageConverter must not be null"); this.messageConverter = messageConverter; @@ -73,13 +82,32 @@ public class RedisInboundChannelAdapter extends MessageProducerSupport { @Override protected void onInit() { super.onInit(); - Assert.notEmpty(this.topics, "at least one topis is required for subscription"); + boolean hasTopics = false; + if (this.topics != null) { + Assert.noNullElements(this.topics, "'topics' may not contain null elements."); + hasTopics = true; + } + boolean hasPatterns = false; + if (this.topicPatterns != null) { + Assert.noNullElements(this.topicPatterns, "'topicPatterns' may not contain null elements."); + hasPatterns = true; + + } + Assert.state(hasTopics || hasPatterns, "at least one topic or topic pattern is required for subscription."); + MessageListenerDelegate delegate = new MessageListenerDelegate(); MessageListenerAdapter adapter = new MessageListenerAdapter(delegate); adapter.setSerializer(this.serializer); - List topicList = new ArrayList(); - for (String topic : this.topics) { - topicList.add(new ChannelTopic(topic)); + List topicList = new ArrayList(); + if (hasTopics) { + for (String topic : this.topics) { + topicList.add(new ChannelTopic(topic)); + } + } + if (hasPatterns) { + for (String pattern : this.topicPatterns) { + topicList.add(new PatternTopic(pattern)); + } } adapter.afterPropertiesSet(); this.container.addMessageListener(adapter, topicList); diff --git a/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-3.0.xsd b/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-3.0.xsd index b6ac1e088d..0e62464ad0 100644 --- a/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-3.0.xsd +++ b/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-3.0.xsd @@ -147,6 +147,13 @@ + + + + Redis topic patterns as a comma-delimited list of Strings. + + + @@ -25,14 +25,14 @@ class="org.springframework.integration.redis.config.RedisInboundChannelAdapterParserTests$TestMessageConverter" /> + id="autoChannel" topics="foo1, bar1" error-channel="testErrorChannel" + message-converter="testConverter" auto-startup="false"/> - + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java index ed0e51b9ed..00229199f3 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java @@ -20,14 +20,19 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; +import org.hamcrest.Matchers; import org.junit.Test; import org.junit.runner.RunWith; + import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.ApplicationContext; import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.listener.RedisMessageListenerContainer; +import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.redis.inbound.RedisInboundChannelAdapter; @@ -58,7 +63,6 @@ public class RedisInboundChannelAdapterParserTests extends RedisAvailableTests { private RedisInboundChannelAdapter autoChannelAdapter; @Test - @RedisAvailable public void validateConfiguration() { RedisInboundChannelAdapter adapter = context.getBean("adapter", RedisInboundChannelAdapter.class); assertEquals("adapter", adapter.getComponentName()); @@ -78,18 +82,24 @@ public class RedisInboundChannelAdapterParserTests extends RedisAvailableTests { @Test @RedisAvailable public void testInboundChannelAdapterMessaging() throws Exception { + RedisInboundChannelAdapter adapter = context.getBean("adapter", RedisInboundChannelAdapter.class); + this.awaitContainerSubscribedWithPatterns(TestUtils.getPropertyValue(adapter, "container", RedisMessageListenerContainer.class)); + RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest(); connectionFactory.getConnection().publish("foo".getBytes(), "Hello Redis from foo".getBytes()); + connectionFactory.getConnection().publish("bar".getBytes(), "Hello Redis from bar".getBytes()); QueueChannel receiveChannel = context.getBean("receiveChannel", QueueChannel.class); - assertEquals("Hello Redis from foo", receiveChannel.receive(2000).getPayload()); - connectionFactory.getConnection().publish("bar".getBytes(), "Hello Redis from bar".getBytes()); - assertEquals("Hello Redis from bar", receiveChannel.receive(2000).getPayload()); + for (int i = 0; i < 3; i++) { + Message receive = receiveChannel.receive(2000); + assertNotNull(receive); + assertThat(receive.getPayload(), Matchers. isOneOf("Hello Redis from foo", "Hello Redis from bar")); + } + } @Test - @RedisAvailable public void testAutoChannel() { assertSame(autoChannel, TestUtils.getPropertyValue(autoChannelAdapter, "outputChannel")); } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java index b476dc7780..b1d089a889 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java @@ -73,6 +73,20 @@ public class RedisAvailableTests { assertTrue("RedisMessageListenerContainer Failed to Subscribe", n < 100); } + protected void awaitContainerSubscribedWithPatterns(RedisMessageListenerContainer container) throws Exception { + this.awaitContainerSubscribed(container); + RedisConnection connection = TestUtils.getPropertyValue(container, "subscriptionTask.connection", + RedisConnection.class); + + int n = 0; + while (n++ < 100 && connection.getSubscription().getPatterns().size() == 0) { + Thread.sleep(100); + } + // TODO: remove this additional delay when/if https://jira.springsource.org/browse/DATAREDIS-242 is resolved + Thread.sleep(250); + assertTrue("RedisMessageListenerContainer Failed to Subscribe with patterns", n < 100); + } + protected void prepareList(RedisConnectionFactory connectionFactory){ StringRedisTemplate redisTemplate = new StringRedisTemplate(); @@ -121,4 +135,5 @@ public class RedisAvailableTests { ops.add("Abraham Lincoln", 19); ops.add("George Washington", 18); } + } diff --git a/src/reference/docbook/redis.xml b/src/reference/docbook/redis.xml index ff6c822d9e..e3c71c2be7 100644 --- a/src/reference/docbook/redis.xml +++ b/src/reference/docbook/redis.xml @@ -150,8 +150,15 @@ rt.setConnectionFactory(redisConnectionFactory);]]> Redis Messages and the Spring Integration Message payloads. The default is a SimpleMessageConverter. - Inbound adapters can subscribe to multiple topic names hence the comma-delimited set of values in the - topics attribute. + + Inbound adapters can subscribe to multiple topic names hence the comma-delimited set of values in the + topics attribute. + + + Since Spring Integration 3.0, the Inbound Adapter, in addition to the existing topics attribute, + now has the topic-patterns attribute. This attribute contains a comma-delimited 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 diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 3a6935d225..95186b1deb 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -584,6 +584,10 @@ The Redis Outbound Channel Adapter now has the topic-expression property to determine the Redis topic against the Message at runtime. + + The Redis Inbound Channel Adapter, in addition to the existing topics attribute, + now has the topic-patterns attribute. + From f11f00b5ab863765f3246265c9d02253f941696c Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 6 Nov 2013 17:07:08 -0500 Subject: [PATCH 3/5] INT-3100 Support Remote File Streaming (Inbound) - Add readRaw and finalizeRaw methods to Session - Add tests for SFTP and FTP Allows retrieval of a remote file as a stream. JIRA: https://jira.springsource.org/browse/INT-3100 --- .../remote/session/CachingSessionFactory.java | 26 ++ .../file/remote/session/Session.java | 27 +- .../RemoteFileOutboundGatewayTests.java | 294 ++++++++++++++---- .../integration/ftp/session/FtpSession.java | 63 +++- .../integration/ftp/TesFtpServer.java | 10 + .../ftp/outbound/FtpServerOutboundTests.java | 21 ++ .../integration/sftp/session/SftpSession.java | 48 ++- .../outbound/SftpServerOutboundTests.java | 31 +- 8 files changed, 435 insertions(+), 85 deletions(-) diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java index 39077b51e5..77c1363c8f 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java @@ -65,14 +65,17 @@ public class CachingSessionFactory implements SessionFactory, DisposableBe public CachingSessionFactory(SessionFactory sessionFactory, int sessionCacheSize) { this.sessionFactory = sessionFactory; this.pool = new SimplePool>(sessionCacheSize, new SimplePool.PoolItemCallback>() { + @Override public Session createForPool() { return CachingSessionFactory.this.sessionFactory.getSession(); } + @Override public boolean isStale(Session session) { return !session.isOpen(); } + @Override public void removedFromPool(Session session) { session.close(); } @@ -100,6 +103,7 @@ public class CachingSessionFactory implements SessionFactory, DisposableBe /** * Get a session from the pool (or block if none available). */ + @Override public Session getSession() { return new CachedSession(this.pool.getItem()); } @@ -107,6 +111,7 @@ public class CachingSessionFactory implements SessionFactory, DisposableBe /** * Remove (close) any unused sessions in the pool. */ + @Override public void destroy() { this.pool.removeAllIdleItems(); } @@ -122,6 +127,7 @@ public class CachingSessionFactory implements SessionFactory, DisposableBe this.targetSession = targetSession; } + @Override public synchronized void close() { if (released) { if (logger.isDebugEnabled()){ @@ -137,41 +143,61 @@ public class CachingSessionFactory implements SessionFactory, DisposableBe } } + @Override public boolean remove(String path) throws IOException{ return this.targetSession.remove(path); } + @Override public F[] list(String path) throws IOException{ return this.targetSession.list(path); } + @Override public void read(String source, OutputStream os) throws IOException{ this.targetSession.read(source, os); } + @Override public void write(InputStream inputStream, String destination) throws IOException{ this.targetSession.write(inputStream, destination); } + @Override public boolean isOpen() { return this.targetSession.isOpen(); } + @Override public void rename(String pathFrom, String pathTo) throws IOException { this.targetSession.rename(pathFrom, pathTo); } + @Override public boolean mkdir(String directory) throws IOException { return this.targetSession.mkdir(directory); } + @Override public boolean exists(String path) throws IOException{ return this.targetSession.exists(path); } + @Override public String[] listNames(String path) throws IOException { return this.targetSession.listNames(path); } + + @Override + public InputStream readRaw(String source) throws IOException { + return this.targetSession.readRaw(source); + } + + @Override + public boolean finalizeRaw() throws IOException { + return this.targetSession.finalizeRaw(); + } + } } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/Session.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/Session.java index 4a5c325002..2b9469c74d 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/Session.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/Session.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -35,21 +35,36 @@ public interface Session { boolean remove(String path) throws IOException; T[] list(String path) throws IOException; - + void read(String source, OutputStream outputStream) throws IOException; void write(InputStream inputStream, String destination) throws IOException; - + boolean mkdir(String directory) throws IOException; - + void rename(String pathFrom, String pathTo) throws IOException; void close(); - + boolean isOpen(); - + boolean exists(String path) throws IOException; String[] listNames(String path) throws IOException; + /** + * Retrieve a remote file as a raw {@link InputStream}. + * @param source The path of the remote file + * @return The raw inputStream. + */ + InputStream readRaw(String source) throws IOException; + + /** + * Invoke after closing the InputStream from {@link #readRaw(String)}. + * Required by some session providers. + * @return true if successful. + * @throws IOException + */ + boolean finalizeRaw() throws IOException; + } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java index 16a6171622..e460b71359 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java @@ -55,7 +55,6 @@ import org.springframework.integration.support.MessageBuilder; /** * @author Gary Russell * @since 2.1 - * */ @SuppressWarnings("rawtypes") public class RemoteFileOutboundGatewayTests { @@ -63,11 +62,11 @@ public class RemoteFileOutboundGatewayTests { private final String tmpDir = System.getProperty("java.io.tmpdir"); - @Test(expected=IllegalArgumentException.class) + @Test(expected = IllegalArgumentException.class) public void testBad() throws Exception { SessionFactory sessionFactory = mock(SessionFactory.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "bad", "payload"); + (sessionFactory, "bad", "payload"); gw.afterPropertiesSet(); } @@ -106,7 +105,7 @@ public class RemoteFileOutboundGatewayTests { SessionFactory sessionFactory = mock(SessionFactory.class); Session session = mock(Session.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "ls", "payload"); + (sessionFactory, "ls", "payload"); gw.afterPropertiesSet(); when(sessionFactory.getSession()).thenReturn(session); TestLsEntry[] files = fileList(); @@ -128,6 +127,7 @@ public class RemoteFileOutboundGatewayTests { /** * Test a wildcard mget where the full path is returned for each file + * * @throws Exception */ @Test @@ -138,19 +138,25 @@ public class RemoteFileOutboundGatewayTests { private void testMGetWildGuts(final String path1, final String path2) { SessionFactory sessionFactory = mock(SessionFactory.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "mget", "payload"); - gw.setLocalDirectory(new File(this.tmpDir )); + (sessionFactory, "mget", "payload"); + gw.setLocalDirectory(new File(this.tmpDir)); gw.afterPropertiesSet(); new File(this.tmpDir + "/f1").delete(); new File(this.tmpDir + "/f2").delete(); when(sessionFactory.getSession()).thenReturn(new Session() { int n; + + @Override public boolean remove(String path) throws IOException { return false; } + + @Override public Object[] list(String path) throws IOException { return null; } + + @Override public void read(String source, OutputStream outputStream) throws IOException { if (n++ == 0) { @@ -161,25 +167,49 @@ public class RemoteFileOutboundGatewayTests { } outputStream.write("testData".getBytes()); } + + @Override public void write(InputStream inputStream, String destination) throws IOException { } + + @Override public boolean mkdir(String directory) throws IOException { return false; } + + @Override public void rename(String pathFrom, String pathTo) throws IOException { } + + @Override public void close() { } + + @Override public boolean isOpen() { return false; } + + @Override public boolean exists(String path) throws IOException { return false; } + + @Override public String[] listNames(String path) throws IOException { - return new String[] {path1, path2}; + return new String[]{path1, path2}; + } + + @Override + public InputStream readRaw(String source) throws IOException { + return null; + } + + @Override + public boolean finalizeRaw() throws IOException { + return false; } }); @SuppressWarnings("unchecked") @@ -196,40 +226,69 @@ public class RemoteFileOutboundGatewayTests { public void testMGetSingle() throws Exception { SessionFactory sessionFactory = mock(SessionFactory.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "mget", "payload"); - gw.setLocalDirectory(new File(this.tmpDir )); + (sessionFactory, "mget", "payload"); + gw.setLocalDirectory(new File(this.tmpDir)); gw.afterPropertiesSet(); new File(this.tmpDir + "/f1").delete(); when(sessionFactory.getSession()).thenReturn(new Session() { + @Override public boolean remove(String path) throws IOException { return false; } + + @Override public Object[] list(String path) throws IOException { return null; } + + @Override public void read(String source, OutputStream outputStream) throws IOException { outputStream.write("testData".getBytes()); } + + @Override public void write(InputStream inputStream, String destination) throws IOException { } + + @Override public boolean mkdir(String directory) throws IOException { return false; } + + @Override public void rename(String pathFrom, String pathTo) throws IOException { } + + @Override public void close() { } + + @Override public boolean isOpen() { return false; } + + @Override public boolean exists(String path) throws IOException { return false; } + + @Override public String[] listNames(String path) throws IOException { - return new String[] {"f1"}; + return new String[]{"f1"}; + } + + @Override + public InputStream readRaw(String source) throws IOException { + return null; + } + + @Override + public boolean finalizeRaw() throws IOException { + return false; } }); @SuppressWarnings("unchecked") @@ -241,47 +300,76 @@ public class RemoteFileOutboundGatewayTests { out.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)); } - @Test(expected=MessagingException.class) + @Test(expected = MessagingException.class) public void testMGetEmpty() throws Exception { SessionFactory sessionFactory = mock(SessionFactory.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "mget", "payload"); - gw.setLocalDirectory(new File(this.tmpDir )); + (sessionFactory, "mget", "payload"); + gw.setLocalDirectory(new File(this.tmpDir)); gw.setOptions(" -x "); gw.afterPropertiesSet(); new File(this.tmpDir + "/f1").delete(); new File(this.tmpDir + "/f2").delete(); when(sessionFactory.getSession()).thenReturn(new Session() { + @Override public boolean remove(String path) throws IOException { return false; } + + @Override public Object[] list(String path) throws IOException { return null; } + + @Override public void read(String source, OutputStream outputStream) throws IOException { outputStream.write("testData".getBytes()); } + + @Override public void write(InputStream inputStream, String destination) throws IOException { } + + @Override public boolean mkdir(String directory) throws IOException { return false; } + + @Override public void rename(String pathFrom, String pathTo) throws IOException { } + + @Override public void close() { } + + @Override public boolean isOpen() { return false; } + + @Override public boolean exists(String path) throws IOException { return false; } + + @Override public String[] listNames(String path) throws IOException { return new String[0]; } + + @Override + public InputStream readRaw(String source) throws IOException { + return null; + } + + @Override + public boolean finalizeRaw() throws IOException { + return false; + } }); gw.handleRequestMessage(new GenericMessage("testremote/*")); } @@ -290,7 +378,7 @@ public class RemoteFileOutboundGatewayTests { public void testMove() throws Exception { SessionFactory sessionFactory = mock(SessionFactory.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "mv", "payload"); + (sessionFactory, "mv", "payload"); gw.afterPropertiesSet(); Session session = mock(Session.class); final AtomicReference args = new AtomicReference(); @@ -302,7 +390,7 @@ public class RemoteFileOutboundGatewayTests { return null; } }).when(session).rename(anyString(), anyString()); - when (sessionFactory.getSession()).thenReturn(session); + when(sessionFactory.getSession()).thenReturn(session); Message requestMessage = MessageBuilder.withPayload("foo") .setHeader(FileHeaders.RENAME_TO, "bar") .build(); @@ -316,7 +404,7 @@ public class RemoteFileOutboundGatewayTests { public void testMoveWithExpression() throws Exception { SessionFactory sessionFactory = mock(SessionFactory.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "mv", "payload"); + (sessionFactory, "mv", "payload"); gw.setRenameExpression("payload.substring(1)"); gw.afterPropertiesSet(); Session session = mock(Session.class); @@ -329,7 +417,7 @@ public class RemoteFileOutboundGatewayTests { return null; } }).when(session).rename(anyString(), anyString()); - when (sessionFactory.getSession()).thenReturn(session); + when(sessionFactory.getSession()).thenReturn(session); Message out = (Message) gw.handleRequestMessage(new GenericMessage("foo")); assertEquals("oo", out.getHeaders().get(FileHeaders.RENAME_TO)); assertEquals("foo", out.getHeaders().get(FileHeaders.REMOTE_FILE)); @@ -341,7 +429,7 @@ public class RemoteFileOutboundGatewayTests { public void testMoveWithMkDirs() throws Exception { SessionFactory sessionFactory = mock(SessionFactory.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "mv", "payload"); + (sessionFactory, "mv", "payload"); gw.setRenameExpression("'foo/bar/baz'"); gw.afterPropertiesSet(); Session session = mock(Session.class); @@ -356,12 +444,13 @@ public class RemoteFileOutboundGatewayTests { }).when(session).rename(anyString(), anyString()); final List madeDirs = new ArrayList(); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { madeDirs.add((String) invocation.getArguments()[0]); return null; } }).when(session).mkdir(anyString()); - when (sessionFactory.getSession()).thenReturn(session); + when(sessionFactory.getSession()).thenReturn(session); Message requestMessage = MessageBuilder.withPayload("foo") .setHeader(FileHeaders.RENAME_TO, "bar") .build(); @@ -390,7 +479,7 @@ public class RemoteFileOutboundGatewayTests { SessionFactory sessionFactory = mock(SessionFactory.class); Session session = mock(Session.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "ls", "payload"); + (sessionFactory, "ls", "payload"); gw.setOptions("-f"); gw.afterPropertiesSet(); when(sessionFactory.getSession()).thenReturn(session); @@ -407,23 +496,23 @@ public class RemoteFileOutboundGatewayTests { } public TestLsEntry[] level1List() { - return new TestLsEntry[] { - new TestLsEntry("f1", 123, false, false, 1234, "-r--r--r--"), - new TestLsEntry("d1", 0, true, false, 12345, "drw-r--r--"), - new TestLsEntry("f2", 12345, false, false, 123456, "-rw-r--r--") + return new TestLsEntry[]{ + new TestLsEntry("f1", 123, false, false, 1234, "-r--r--r--"), + new TestLsEntry("d1", 0, true, false, 12345, "drw-r--r--"), + new TestLsEntry("f2", 12345, false, false, 123456, "-rw-r--r--") }; } public TestLsEntry[] level2List() { - return new TestLsEntry[] { - new TestLsEntry("d2", 0, true, false, 12345, "drw-r--r--"), - new TestLsEntry("f3", 12345, false, false, 123456, "-rw-r--r--") + return new TestLsEntry[]{ + new TestLsEntry("d2", 0, true, false, 12345, "drw-r--r--"), + new TestLsEntry("f3", 12345, false, false, 123456, "-rw-r--r--") }; } public TestLsEntry[] level3List() { - return new TestLsEntry[] { - new TestLsEntry("f4", 12345, false, false, 123456, "-rw-r--r--") + return new TestLsEntry[]{ + new TestLsEntry("f4", 12345, false, false, 123456, "-rw-r--r--") }; } @@ -432,7 +521,7 @@ public class RemoteFileOutboundGatewayTests { SessionFactory sessionFactory = mock(SessionFactory.class); Session session = mock(Session.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "ls", "payload"); + (sessionFactory, "ls", "payload"); gw.setOptions("-f -R"); gw.afterPropertiesSet(); when(sessionFactory.getSession()).thenReturn(session); @@ -459,7 +548,7 @@ public class RemoteFileOutboundGatewayTests { SessionFactory sessionFactory = mock(SessionFactory.class); Session session = mock(Session.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "ls", "payload"); + (sessionFactory, "ls", "payload"); gw.setOptions("-f -R -dirs"); gw.afterPropertiesSet(); when(sessionFactory.getSession()).thenReturn(session); @@ -488,7 +577,7 @@ public class RemoteFileOutboundGatewayTests { SessionFactory sessionFactory = mock(SessionFactory.class); Session session = mock(Session.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "ls", "payload"); + (sessionFactory, "ls", "payload"); gw.afterPropertiesSet(); when(sessionFactory.getSession()).thenReturn(session); TestLsEntry[] files = new TestLsEntry[0]; @@ -504,7 +593,7 @@ public class RemoteFileOutboundGatewayTests { SessionFactory sessionFactory = mock(SessionFactory.class); Session session = mock(Session.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "ls", "payload"); + (sessionFactory, "ls", "payload"); gw.setOptions("-1"); gw.afterPropertiesSet(); when(sessionFactory.getSession()).thenReturn(session); @@ -523,7 +612,7 @@ public class RemoteFileOutboundGatewayTests { SessionFactory sessionFactory = mock(SessionFactory.class); Session session = mock(Session.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "ls", "payload"); + (sessionFactory, "ls", "payload"); gw.setOptions("-1 -f"); gw.afterPropertiesSet(); when(sessionFactory.getSession()).thenReturn(session); @@ -542,7 +631,7 @@ public class RemoteFileOutboundGatewayTests { SessionFactory sessionFactory = mock(SessionFactory.class); Session session = mock(Session.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "ls", "payload"); + (sessionFactory, "ls", "payload"); gw.setOptions("-1 -dirs"); gw.afterPropertiesSet(); when(sessionFactory.getSession()).thenReturn(session); @@ -562,7 +651,7 @@ public class RemoteFileOutboundGatewayTests { SessionFactory sessionFactory = mock(SessionFactory.class); Session session = mock(Session.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "ls", "payload"); + (sessionFactory, "ls", "payload"); gw.setOptions("-1 -dirs -links"); gw.afterPropertiesSet(); when(sessionFactory.getSession()).thenReturn(session); @@ -583,7 +672,7 @@ public class RemoteFileOutboundGatewayTests { SessionFactory sessionFactory = mock(SessionFactory.class); Session session = mock(Session.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "ls", "payload"); + (sessionFactory, "ls", "payload"); gw.setOptions("-1 -a -f -dirs -links"); gw.afterPropertiesSet(); when(sessionFactory.getSession()).thenReturn(session); @@ -606,7 +695,7 @@ public class RemoteFileOutboundGatewayTests { SessionFactory sessionFactory = mock(SessionFactory.class); Session session = mock(Session.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "ls", "payload"); + (sessionFactory, "ls", "payload"); gw.setOptions("-1 -a -f -dirs -links"); gw.setFilter(new TestPatternFilter("*4")); gw.afterPropertiesSet(); @@ -624,45 +713,75 @@ public class RemoteFileOutboundGatewayTests { public void testGet() throws Exception { SessionFactory sessionFactory = mock(SessionFactory.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "get", "payload"); - gw.setLocalDirectory(new File(this.tmpDir )); + (sessionFactory, "get", "payload"); + gw.setLocalDirectory(new File(this.tmpDir)); gw.afterPropertiesSet(); new File(this.tmpDir + "/f1").delete(); - when(sessionFactory.getSession()).thenReturn(new Session(){ + when(sessionFactory.getSession()).thenReturn(new Session() { private boolean open = true; + + @Override public boolean remove(String path) throws IOException { return false; } + + @Override public TestLsEntry[] list(String path) throws IOException { - return new TestLsEntry[] { + return new TestLsEntry[]{ new TestLsEntry("f1", 1234, false, false, 12345, "-rw-r--r--") }; } + + @Override public void read(String source, OutputStream outputStream) throws IOException { outputStream.write("testfile".getBytes()); } + + @Override public void write(InputStream inputStream, String destination) throws IOException { } + + @Override public boolean mkdir(String directory) throws IOException { return true; } + + @Override public void rename(String pathFrom, String pathTo) throws IOException { } + + @Override public void close() { open = false; } + + @Override public boolean isOpen() { return open; } + + @Override public boolean exists(String path) throws IOException { return true; } + + @Override public String[] listNames(String path) throws IOException { return null; } + + @Override + public InputStream readRaw(String source) throws IOException { + return null; + } + + @Override + public boolean finalizeRaw() throws IOException { + return false; + } }); @SuppressWarnings("unchecked") Message out = (Message) gw.handleRequestMessage(new GenericMessage("f1")); @@ -680,7 +799,7 @@ public class RemoteFileOutboundGatewayTests { public void testGet_P() throws Exception { SessionFactory sessionFactory = mock(SessionFactory.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "get", "payload"); + (sessionFactory, "get", "payload"); gw.setLocalDirectory(new File(this.tmpDir)); gw.setOptions("-P"); gw.afterPropertiesSet(); @@ -688,41 +807,71 @@ public class RemoteFileOutboundGatewayTests { Calendar cal = Calendar.getInstance(); cal.add(Calendar.MONTH, -1); final Date modified = new Date(cal.getTime().getTime() / 1000 * 1000); - when(sessionFactory.getSession()).thenReturn(new Session(){ + when(sessionFactory.getSession()).thenReturn(new Session() { private boolean open = true; + + @Override public boolean remove(String path) throws IOException { return false; } + + @Override public TestLsEntry[] list(String path) throws IOException { - return new TestLsEntry[] { + return new TestLsEntry[]{ new TestLsEntry("f1", 1234, false, false, modified.getTime(), "-rw-r--r--") }; } + + @Override public void read(String source, OutputStream outputStream) throws IOException { outputStream.write("testfile".getBytes()); } + + @Override public void write(InputStream inputStream, String destination) throws IOException { } + + @Override public boolean mkdir(String directory) throws IOException { return true; } + + @Override public void rename(String pathFrom, String pathTo) throws IOException { } + + @Override public void close() { open = false; } + + @Override public boolean isOpen() { return open; } + + @Override public boolean exists(String path) throws IOException { return true; } + + @Override public String[] listNames(String path) throws IOException { return null; } + + @Override + public InputStream readRaw(String source) throws IOException { + return null; + } + + @Override + public boolean finalizeRaw() throws IOException { + return false; + } }); @SuppressWarnings("unchecked") Message out = (Message) gw.handleRequestMessage(new GenericMessage("x/f1")); @@ -743,44 +892,74 @@ public class RemoteFileOutboundGatewayTests { new File(this.tmpDir + "/x").delete(); SessionFactory sessionFactory = mock(SessionFactory.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "get", "payload"); + (sessionFactory, "get", "payload"); gw.setLocalDirectory(new File(this.tmpDir + "/x")); gw.afterPropertiesSet(); - when(sessionFactory.getSession()).thenReturn(new Session(){ + when(sessionFactory.getSession()).thenReturn(new Session() { private boolean open = true; + + @Override public boolean remove(String path) throws IOException { return false; } + + @Override public TestLsEntry[] list(String path) throws IOException { - return new TestLsEntry[] { + return new TestLsEntry[]{ new TestLsEntry("f1", 1234, false, false, 12345, "-rw-r--r--") }; } + + @Override public void read(String source, OutputStream outputStream) throws IOException { outputStream.write("testfile".getBytes()); } + + @Override public void write(InputStream inputStream, String destination) throws IOException { } + + @Override public boolean mkdir(String directory) throws IOException { return true; } + + @Override public void rename(String pathFrom, String pathTo) throws IOException { } + + @Override public void close() { open = false; } + + @Override public boolean isOpen() { return open; } + + @Override public boolean exists(String path) throws IOException { return true; } + + @Override public String[] listNames(String path) throws IOException { return null; } + + @Override + public InputStream readRaw(String source) throws IOException { + return null; + } + + @Override + public boolean finalizeRaw() throws IOException { + return false; + } }); gw.handleRequestMessage(new GenericMessage("f1")); File out = new File(this.tmpDir + "/x/f1"); @@ -793,7 +972,7 @@ public class RemoteFileOutboundGatewayTests { SessionFactory sessionFactory = mock(SessionFactory.class); Session session = mock(Session.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "rm", "payload"); + (sessionFactory, "rm", "payload"); gw.afterPropertiesSet(); when(sessionFactory.getSession()).thenReturn(session); when(session.remove("testremote/x/f1")).thenReturn(Boolean.TRUE); @@ -812,9 +991,9 @@ public class RemoteFileOutboundGatewayTests { class TestRemoteFileOutboundGateway extends AbstractRemoteFileOutboundGateway { - @SuppressWarnings({ "rawtypes", "unchecked" }) + @SuppressWarnings({"rawtypes", "unchecked"}) public TestRemoteFileOutboundGateway(SessionFactory sessionFactory, - String command, String expression) { + String command, String expression) { super(sessionFactory, Command.toCommand(command), expression); this.setBeanFactory(mock(BeanFactory.class)); } @@ -868,7 +1047,7 @@ class TestLsEntry extends AbstractFileInfo { private final String permissions; public TestLsEntry(String filename, long size, boolean dir, boolean link, - long modified, String permissions) { + long modified, String permissions) { this.filename = filename; this.size = size; this.dir = dir; @@ -877,30 +1056,37 @@ class TestLsEntry extends AbstractFileInfo { this.permissions = permissions; } + @Override public boolean isDirectory() { return this.dir; } + @Override public long getModified() { return this.modified; } + @Override public String getFilename() { return this.filename; } + @Override public boolean isLink() { return this.link; } + @Override public long getSize() { return this.size; } + @Override public String getPermissions() { return this.permissions; } + @Override public TestLsEntry getFileInfo() { return this; } @@ -911,7 +1097,7 @@ class TestLsEntry extends AbstractFileInfo { } -class TestPatternFilter extends AbstractSimplePatternFileListFilter{ +class TestPatternFilter extends AbstractSimplePatternFileListFilter { public TestPatternFilter(String path) { super(path); @@ -922,4 +1108,4 @@ class TestPatternFilter extends AbstractSimplePatternFileListFilter return file.getFilename(); } -} \ No newline at end of file +} diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java index 470d1116ca..24718c8f5e 100644 --- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -19,17 +19,20 @@ package org.springframework.integration.ftp.session; import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; +import java.util.concurrent.atomic.AtomicBoolean; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.commons.net.ftp.FTPClient; import org.apache.commons.net.ftp.FTPFile; +import org.apache.commons.net.ftp.FTPReply; + import org.springframework.integration.file.remote.session.Session; import org.springframework.util.Assert; /** * Implementation of {@link Session} for FTP. - * + * * @author Mark Fisher * @author Oleg Zhurakousky * @author Gary Russell @@ -41,6 +44,7 @@ public class FtpSession implements Session { private final FTPClient client; + private final AtomicBoolean readingRaw = new AtomicBoolean(); public FtpSession(FTPClient client) { Assert.notNull(client, "client must not be null"); @@ -48,6 +52,7 @@ public class FtpSession implements Session { } + @Override public boolean remove(String path) throws IOException { Assert.hasText(path, "path must not be null"); boolean completed = this.client.deleteFile(path); @@ -56,17 +61,20 @@ public class FtpSession implements Session { } return completed; } - + + @Override public FTPFile[] list(String path) throws IOException { Assert.hasText(path, "path must not be null"); return this.client.listFiles(path); } + @Override public String[] listNames(String path) throws IOException { Assert.hasText(path, "path must not be null"); return this.client.listNames(path); } + @Override public void read(String path, OutputStream fos) throws IOException { Assert.hasText(path, "path must not be null"); Assert.notNull(fos, "outputStream must not be null"); @@ -78,12 +86,40 @@ public class FtpSession implements Session { logger.info("File has been successfully transfered from: " + path); } + @Override + public InputStream readRaw(String source) throws IOException { + if (!this.readingRaw.compareAndSet(false, true)) { + throw new IOException("Previous raw read was not finalized"); + } + InputStream inputStream = this.client.retrieveFileStream(source); + if (inputStream == null) { + throw new IOException("Failed to obtain InputStream for remote file " + source + ": " + this.client.getReplyCode()); + } + return inputStream; + } + + @Override + public boolean finalizeRaw() throws IOException { + if (!this.readingRaw.compareAndSet(true, false)) { + throw new IOException("Raw read is not in process"); + } + if (this.client.completePendingCommand()) { + int replyCode = this.client.getReplyCode(); + if (logger.isDebugEnabled()) { + logger.debug(this + " finalizeRaw - reply code: " + replyCode); + } + return FTPReply.isPositiveCompletion(replyCode); + } + throw new IOException("completePendingCommandFailed"); + } + + @Override public void write(InputStream inputStream, String path) throws IOException { Assert.notNull(inputStream, "inputStream must not be null"); Assert.hasText(path, "path must not be null"); boolean completed = this.client.storeFile(path, inputStream); if (!completed) { - throw new IOException("Failed to write to '" + path + throw new IOException("Failed to write to '" + path + "'. Server replied with: " + this.client.getReplyString()); } if (logger.isInfoEnabled()) { @@ -91,6 +127,7 @@ public class FtpSession implements Session { } } + @Override public void close() { try { this.client.disconnect(); @@ -102,6 +139,7 @@ public class FtpSession implements Session { } } + @Override public boolean isOpen() { try { this.client.noop(); @@ -112,38 +150,41 @@ public class FtpSession implements Session { return true; } + @Override public void rename(String pathFrom, String pathTo) throws IOException{ this.client.deleteFile(pathTo); boolean completed = this.client.rename(pathFrom, pathTo); if (!completed) { - throw new IOException("Failed to rename '" + pathFrom + + throw new IOException("Failed to rename '" + pathFrom + "' to " + pathTo + "'. Server replied with: " + this.client.getReplyString()); } if (logger.isInfoEnabled()) { logger.info("File has been successfully renamed from: " + pathFrom + " to " + pathTo); } } - + + @Override public boolean mkdir(String remoteDirectory) throws IOException { return this.client.makeDirectory(remoteDirectory); } - + + @Override public boolean exists(String path) throws IOException{ Assert.hasText(path, "'path' must not be empty"); - + String currentWorkingPath = this.client.printWorkingDirectory(); Assert.state(currentWorkingPath != null, "working directory cannot be determined, therefore exists check can not be completed"); boolean exists = false; try { - if (this.client.changeWorkingDirectory(path)){ + if (this.client.changeWorkingDirectory(path)) { exists = true; } - } + } finally { this.client.changeWorkingDirectory(currentWorkingPath); } - + return exists; } } diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/TesFtpServer.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/TesFtpServer.java index 9d059f3f10..e0a62d2d3d 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/TesFtpServer.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/TesFtpServer.java @@ -17,6 +17,7 @@ package org.springframework.integration.ftp; import java.io.File; +import java.io.FileOutputStream; import java.io.IOException; import java.util.Arrays; @@ -82,13 +83,22 @@ public class TesFtpServer { sourceFtpDirectory.mkdir(); File file = new File(sourceFtpDirectory, "ftpSource1.txt"); file.createNewFile(); + FileOutputStream fos = new FileOutputStream(file); + fos.write("source1".getBytes()); + fos.close(); file = new File(sourceFtpDirectory, "ftpSource2.txt"); file.createNewFile(); + fos = new FileOutputStream(file); + fos.write("source2".getBytes()); + fos.close(); File subSourceFtpDirectory = new File(sourceFtpDirectory, "subFtpSource"); subSourceFtpDirectory.mkdir(); file = new File(subSourceFtpDirectory, "subFtpSource1.txt"); file.createNewFile(); + fos = new FileOutputStream(file); + fos.write("subSource1".getBytes()); + fos.close(); targetFtpDirectory = new File(ftpRootFolder, "ftpTarget"); targetFtpDirectory.mkdirs(); diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java index a5c9275949..04d1f849e0 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java @@ -19,8 +19,10 @@ package org.springframework.integration.ftp.outbound; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import java.io.ByteArrayOutputStream; import java.io.File; import java.util.List; @@ -33,10 +35,12 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.Message; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.core.PollableChannel; +import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.ftp.TesFtpServer; import org.springframework.integration.message.GenericMessage; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.util.FileCopyUtils; /** * @author Artem Bilan @@ -170,4 +174,21 @@ public class FtpServerOutboundTests { } + @Test + public void testInt3100RawGET() throws Exception { + Session session = this.ftpServer.ftpSessionFactory().getSession(); + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + FileCopyUtils.copy(session.readRaw("ftpSource/ftpSource1.txt"), baos); + assertTrue(session.finalizeRaw()); + assertEquals("source1", new String(baos.toByteArray())); + + baos = new ByteArrayOutputStream(); + FileCopyUtils.copy(session.readRaw("ftpSource/ftpSource2.txt"), baos); + assertTrue(session.finalizeRaw()); + assertEquals("source2", new String(baos.toByteArray())); + + session.close(); + } + + } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java index 96cdb9bf12..a3d47250af 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -25,6 +25,7 @@ import java.util.Vector; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; + import org.springframework.core.NestedIOException; import org.springframework.integration.file.remote.session.Session; import org.springframework.util.Assert; @@ -61,6 +62,7 @@ class SftpSession implements Session { } + @Override public boolean remove(String path) throws IOException { Assert.state(this.channel != null, "session is not connected"); try { @@ -72,6 +74,7 @@ class SftpSession implements Session { } } + @Override public LsEntry[] list(String path) throws IOException { Assert.state(this.channel != null, "session is not connected"); try { @@ -92,6 +95,7 @@ class SftpSession implements Session { return new LsEntry[0]; } + @Override public String[] listNames(String path) throws IOException { LsEntry[] entries = this.list(path); List names = new ArrayList(); @@ -107,6 +111,7 @@ class SftpSession implements Session { } + @Override public void read(String source, OutputStream os) throws IOException { Assert.state(this.channel != null, "session is not connected"); try { @@ -114,10 +119,26 @@ class SftpSession implements Session { FileCopyUtils.copy(is, os); } catch (SftpException e) { - throw new NestedIOException("failed to read file", e); + throw new NestedIOException("failed to read file " + source, e); } } + @Override + public InputStream readRaw(String source) throws IOException { + try { + return this.channel.get(source); + } + catch (SftpException e) { + throw new NestedIOException("failed to read file " + source, e); + } + } + + @Override + public boolean finalizeRaw() throws IOException { + return true; + } + + @Override public void write(InputStream inputStream, String destination) throws IOException { Assert.state(this.channel != null, "session is not connected"); try { @@ -128,38 +149,41 @@ class SftpSession implements Session { } } + @Override public void close() { if (this.jschSession.isConnected()) { this.jschSession.disconnect(); } } + @Override public boolean isOpen() { return this.jschSession.isConnected(); } + @Override public void rename(String pathFrom, String pathTo) throws IOException { - try { + try { this.channel.rename(pathFrom, pathTo); - } + } catch (SftpException sftpex) { if (logger.isDebugEnabled()){ - logger.debug("Initial File rename failed, possibly because file already exists. Will attempt to delete file: " + logger.debug("Initial File rename failed, possibly because file already exists. Will attempt to delete file: " + pathTo + " and execute rename again."); } - try { + try { this.remove(pathTo); if (logger.isDebugEnabled()) { logger.debug("Delete file: " + pathTo + " succeeded. Will attempt rename again"); - } - } + } + } catch (IOException ioex) { throw new NestedIOException("Failed to delete file " + pathTo, ioex); } try { // attempt to rename again this.channel.rename(pathFrom, pathTo); - } + } catch (SftpException sftpex2) { throw new NestedIOException("failed to rename from " + pathFrom + " to " + pathTo, sftpex2); } @@ -169,8 +193,9 @@ class SftpSession implements Session { } } + @Override public boolean mkdir(String remoteDirectory) throws IOException { - try { + try { this.channel.mkdir(remoteDirectory); } catch (SftpException e) { @@ -179,6 +204,7 @@ class SftpSession implements Session { return true; } + @Override public boolean exists(String path) { try { this.channel.lstat(path); @@ -189,7 +215,7 @@ class SftpSession implements Session { } return false; } - + void connect() { try { if (!this.jschSession.isConnected()) { diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java index dcccb049c5..3e5d978499 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java @@ -19,10 +19,12 @@ package org.springframework.integration.sftp.outbound; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; +import java.io.ByteArrayOutputStream; import java.io.File; import java.io.IOException; import java.util.List; @@ -44,6 +46,7 @@ import org.springframework.integration.message.GenericMessage; import org.springframework.integration.sftp.session.SftpFileInfo; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.util.FileCopyUtils; import com.jcraft.jsch.ChannelSftp.LsEntry; import com.jcraft.jsch.SftpATTRS; @@ -56,10 +59,10 @@ import com.jcraft.jsch.SftpATTRS; * 
  *  $ tree sftpSource/
  *  sftpSource/
- *  ├── sftpSource1.txt
- *  ├── sftpSource2.txt
+ *  ├── sftpSource1.txt - contains 'source1'
+ *  ├── sftpSource2.txt - contains 'source2'
  *  └── subSftpSource
- *      └── subSftpSource1.txt
+ *      └── subSftpSource1.txt - contains 'subSource1'
  *
* * @author Artem Bilan @@ -252,4 +255,26 @@ public class SftpServerOutboundTests { } + /** + * Only runs with a real server (see class javadocs). + */ + @Test + public void testInt3100RawGET() throws Exception { + if (!sessionFactory.toString().startsWith("Mock for")) { + Session session = this.sessionFactory.getSession(); + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + FileCopyUtils.copy(session.readRaw("sftpSource/sftpSource1.txt"), baos); + assertTrue(session.finalizeRaw()); + assertEquals("source1", new String(baos.toByteArray())); + + baos = new ByteArrayOutputStream(); + FileCopyUtils.copy(session.readRaw("sftpSource/sftpSource2.txt"), baos); + assertTrue(session.finalizeRaw()); + assertEquals("source2", new String(baos.toByteArray())); + + session.close(); + } + } + + } From e128c80c19fda7a246dfd551e51277f0a8d2aee3 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 7 Nov 2013 20:39:11 +0200 Subject: [PATCH 4/5] INT-2421 Support 'container-class' on JMS MDCA Convenience for configuring a custom listener container - avoids having to transfer many attributes from the adapter to the bean when switching to a custom implementation. JIRA: https://jira.springsource.org/browse/INT-2421 INT-2421 Polishing - PR Comments Also auto-create adapter channel if necessary. --- .../config/HttpInboundEndpointParser.java | 3 +- .../JmsMessageDrivenEndpointParser.java | 38 +++++++++++- .../jms/config/spring-integration-jms-3.0.xsd | 21 +++++++ ...essageDrivenChannelAdapterParserTests.java | 62 ++++++++++++------- .../config/jmsInboundWithContainerClass.xml | 29 +++++++++ src/reference/docbook/jms.xml | 13 +++- 6 files changed, 139 insertions(+), 27 deletions(-) create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/config/jmsInboundWithContainerClass.xml diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpInboundEndpointParser.java b/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpInboundEndpointParser.java index d60a447358..ef8685dd4d 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpInboundEndpointParser.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpInboundEndpointParser.java @@ -17,6 +17,7 @@ package org.springframework.integration.http.config; import java.util.List; + import org.w3c.dom.Element; import org.springframework.beans.factory.BeanDefinitionStoreException; @@ -74,7 +75,7 @@ public class HttpInboundEndpointParser extends AbstractSingleBeanDefinitionParse throws BeanDefinitionStoreException { String id = super.resolveId(element, definition, parserContext); - if (!element.hasAttribute(getInputChannelAttributeName())) { + if (!this.expectReply && !element.hasAttribute("channel")) { // the created channel will get the 'id', so the adapter's bean name includes a suffix id = id + ".adapter"; } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java index 3276096978..54abca2327 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java @@ -18,14 +18,17 @@ package org.springframework.integration.jms.config; import org.w3c.dom.Element; +import org.springframework.beans.factory.BeanDefinitionStoreException; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.parsing.BeanComponentDefinition; +import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; import org.springframework.integration.jms.JmsMessageDrivenEndpoint; +import org.springframework.jms.listener.DefaultMessageListenerContainer; import org.springframework.util.StringUtils; /** @@ -79,6 +82,22 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition return JmsMessageDrivenEndpoint.class.getName(); } + @Override + protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext) + throws BeanDefinitionStoreException { + String id = super.resolveId(element, definition, parserContext); + + if (!this.expectReply && !element.hasAttribute("channel")) { + // the created channel will get the 'id', so the adapter's bean name includes a suffix + id = id + ".adapter"; + } + if (!StringUtils.hasText(id)) { + id = BeanDefinitionReaderUtils.generateBeanName(definition, parserContext.getRegistry()); + } + + return id; + } + @Override protected boolean shouldGenerateId() { return false; @@ -100,7 +119,11 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition } private String parseMessageListenerContainer(Element element, ParserContext parserContext) { + String containerClass = element.getAttribute("container-class"); if (element.hasAttribute("container")) { + if (StringUtils.hasText(containerClass)) { + parserContext.getReaderContext().error("Cannot have both 'container' and 'container-class'", element); + } for (String containerAttribute : containerAttributes) { if (element.hasAttribute(containerAttribute)) { parserContext.getReaderContext().error("The '" + containerAttribute + @@ -110,8 +133,13 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition return element.getAttribute("container"); } // otherwise, we build a DefaultMessageListenerContainer instance - BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition( - "org.springframework.jms.listener.DefaultMessageListenerContainer"); + BeanDefinitionBuilder builder; + if (StringUtils.hasText(containerClass)) { + builder = BeanDefinitionBuilder.genericBeanDefinition(containerClass); + } + else { + builder = BeanDefinitionBuilder.genericBeanDefinition(DefaultMessageListenerContainer.class); + } String destinationAttribute = this.expectReply ? "request-destination" : "destination"; String destinationNameAttribute = this.expectReply ? "request-destination-name" : "destination-name"; String pubSubDomainAttribute = this.expectReply ? "request-pub-sub-domain" : "pub-sub-domain"; @@ -198,7 +226,11 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel"); } else { - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "channel", "requestChannel"); + String channelName = element.getAttribute("channel"); + if (!StringUtils.hasText(channelName)) { + channelName = IntegrationNamespaceUtils.createDirectChannel(element, parserContext); + } + builder.addPropertyReference("requestChannel", channelName); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout", "requestTimeout"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-payload", "extractRequestPayload"); } diff --git a/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-3.0.xsd b/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-3.0.xsd index a3aedd4afa..6e967bed3a 100644 --- a/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-3.0.xsd +++ b/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-3.0.xsd @@ -1201,6 +1201,11 @@ + @@ -1208,6 +1213,22 @@ + + + + + + + + + + + diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsMessageDrivenChannelAdapterParserTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsMessageDrivenChannelAdapterParserTests.java index 6c4d094a33..5f1475b3c6 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsMessageDrivenChannelAdapterParserTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsMessageDrivenChannelAdapterParserTests.java @@ -27,6 +27,7 @@ import org.junit.Test; import org.springframework.beans.DirectFieldAccessor; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.Message; +import org.springframework.integration.MessageChannel; import org.springframework.integration.core.PollableChannel; import org.springframework.integration.history.MessageHistory; import org.springframework.integration.jms.JmsMessageDrivenEndpoint; @@ -38,6 +39,7 @@ import org.springframework.jms.support.destination.JmsDestinationAccessor; /** * @author Mark Fisher * @author Michael Bannister + * @author Gary Russell */ public class JmsMessageDrivenChannelAdapterParserTests { @@ -87,7 +89,7 @@ public class JmsMessageDrivenChannelAdapterParserTests { public void adapterWithTaskExecutor() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "jmsInboundWithTaskExecutor.xml", this.getClass()); - JmsMessageDrivenEndpoint endpoint = context.getBean("messageDrivenAdapter", JmsMessageDrivenEndpoint.class); + JmsMessageDrivenEndpoint endpoint = context.getBean("messageDrivenAdapter.adapter", JmsMessageDrivenEndpoint.class); DefaultMessageListenerContainer container = TestUtils.getPropertyValue(endpoint, "listenerContainer", DefaultMessageListenerContainer.class); assertSame(context.getBean("exec"), TestUtils.getPropertyValue(container, "taskExecutor")); @@ -95,52 +97,70 @@ public class JmsMessageDrivenChannelAdapterParserTests { } @Test - public void testGatewayWithReceiveTimeout() { + public void testAdapterWithReceiveTimeout() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "jmsInboundWithContainerSettings.xml", this.getClass()); - JmsMessageDrivenEndpoint gateway = (JmsMessageDrivenEndpoint) context.getBean("adapterWithReceiveTimeout"); - gateway.start(); + JmsMessageDrivenEndpoint adapter = (JmsMessageDrivenEndpoint) context.getBean("adapterWithReceiveTimeout.adapter"); + adapter.start(); AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) - new DirectFieldAccessor(gateway).getPropertyValue("listenerContainer"); + new DirectFieldAccessor(adapter).getPropertyValue("listenerContainer"); assertEquals(1111L, new DirectFieldAccessor(container).getPropertyValue("receiveTimeout")); - gateway.stop(); + adapter.stop(); } @Test - public void testGatewayWithRecoveryInterval() { + public void testAdapterWithRecoveryInterval() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "jmsInboundWithContainerSettings.xml", this.getClass()); - JmsMessageDrivenEndpoint gateway = (JmsMessageDrivenEndpoint) context.getBean("adapterWithRecoveryInterval"); - gateway.start(); + JmsMessageDrivenEndpoint adapter = (JmsMessageDrivenEndpoint) context.getBean("adapterWithRecoveryInterval.adapter"); + adapter.start(); AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) - new DirectFieldAccessor(gateway).getPropertyValue("listenerContainer"); + new DirectFieldAccessor(adapter).getPropertyValue("listenerContainer"); assertEquals(2222L, new DirectFieldAccessor(container).getPropertyValue("recoveryInterval")); - gateway.stop(); + adapter.stop(); } @Test - public void testGatewayWithIdleTaskExecutionLimit() { + public void testAdapterWithIdleTaskExecutionLimit() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "jmsInboundWithContainerSettings.xml", this.getClass()); - JmsMessageDrivenEndpoint gateway = (JmsMessageDrivenEndpoint) context.getBean("adapterWithIdleTaskExecutionLimit"); - gateway.start(); + JmsMessageDrivenEndpoint adapter = (JmsMessageDrivenEndpoint) context.getBean("adapterWithIdleTaskExecutionLimit.adapter"); + adapter.start(); AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) - new DirectFieldAccessor(gateway).getPropertyValue("listenerContainer"); + new DirectFieldAccessor(adapter).getPropertyValue("listenerContainer"); assertEquals(7, new DirectFieldAccessor(container).getPropertyValue("idleTaskExecutionLimit")); - gateway.stop(); + adapter.stop(); } @Test - public void testGatewayWithIdleConsumerLimit() { + public void testAdapterWithIdleConsumerLimit() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "jmsInboundWithContainerSettings.xml", this.getClass()); - JmsMessageDrivenEndpoint gateway = (JmsMessageDrivenEndpoint) context.getBean("adapterWithIdleConsumerLimit"); - gateway.start(); + JmsMessageDrivenEndpoint adapter = (JmsMessageDrivenEndpoint) context.getBean("adapterWithIdleConsumerLimit.adapter"); + adapter.start(); AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) - new DirectFieldAccessor(gateway).getPropertyValue("listenerContainer"); + new DirectFieldAccessor(adapter).getPropertyValue("listenerContainer"); assertEquals(33, new DirectFieldAccessor(container).getPropertyValue("idleConsumerLimit")); assertEquals(3, new DirectFieldAccessor(container).getPropertyValue("cacheLevel")); - gateway.stop(); + adapter.stop(); + } + + @Test + public void testAdapterWithContainerClass() { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "jmsInboundWithContainerClass.xml", this.getClass()); + JmsMessageDrivenEndpoint adapter = context.getBean("adapterWithIdleConsumerLimit.adapter", JmsMessageDrivenEndpoint.class); + MessageChannel channel = context.getBean("adapterWithIdleConsumerLimit", MessageChannel.class); + assertSame(channel, TestUtils.getPropertyValue(adapter, "listener.gatewayDelegate.requestChannel")); + adapter.start(); + FooContainer container = TestUtils.getPropertyValue(adapter, "listenerContainer", FooContainer.class); + assertEquals(33, new DirectFieldAccessor(container).getPropertyValue("idleConsumerLimit")); + assertEquals(3, new DirectFieldAccessor(container).getPropertyValue("cacheLevel")); + adapter.stop(); + } + + public static final class FooContainer extends DefaultMessageListenerContainer { + } } diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/jmsInboundWithContainerClass.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/jmsInboundWithContainerClass.xml new file mode 100644 index 0000000000..34bba45821 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/jmsInboundWithContainerClass.xml @@ -0,0 +1,29 @@ + + + + + + + + + + + + + + diff --git a/src/reference/docbook/jms.xml b/src/reference/docbook/jms.xml index 9e482ffe38..f8b8c5169d 100644 --- a/src/reference/docbook/jms.xml +++ b/src/reference/docbook/jms.xml @@ -87,12 +87,21 @@ message-driven Channel Adapter with a Destination reference. ]]> + The Message-Driven adapter also accepts several properties that pertain to the MessageListener container. - These values are only considered if you do not provide an actual 'container' reference. In that case, + These values are only considered if you do not provide a container reference. In that case, an instance of DefaultMessageListenerContainer will be created and configured based on these properties. For example, you can specify the "transaction-manager" reference, the "concurrent-consumers" value, and several other property references and values. Refer to the JavaDoc and Spring Integration's JMS Schema - (spring-integration-jms.xsd) for more detail. + (spring-integration-jms.xsd) for more details. + + + If you have a custom listener container implementation (usually a subclass of + DefaultMessageListenerContainer), you can either provide a reference to an instance + of it using the container attribute, or simply provide its fully qualified class name using + the container-class attribute. In that case, the attributes on the adapter + are transferred to an instance of your custom container. + From 7bc4fe2d0b2985b0e775f843b7687546825c75d9 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 7 Nov 2013 21:09:33 +0200 Subject: [PATCH 5/5] INT-3047 SFTP Multiplex Over Single Connection Enable the use of multiple JSch channels over a single connection. JIRA: https://jira.springsource.org/browse/INT-3047 INT-3047: Polishing locks logic INT-3047 SFTP Reset Cache; Quiesce Shared Sessions Support resetCache() on the CachingSessionFactory - reset a shared JSch session so it is reestablished on next use - immediately close idle sessions - close in-use sessions as they are returned - close the channel when an SftpSession is closed and a shared JSch is being used - physically close a shared JSch session only when the last channel is closed Polishing Forgot to save the ftp.xml. Change CachedSession.epoch to created. INT-3047 Polishing - PR Comments - Also add missing docs for `preserve-timestamp` INT-3047 Polish - PR Comments Also change epoch to nanoseconds and use a simple != comparison with the epoch in which a session was created. --- .../integration/util/SimplePool.java | 5 +- .../remote/session/CachingSessionFactory.java | 52 ++++++- .../remote/session/SharedSessionCapable.java | 39 +++++ .../session/CachingSessionFactoryTests.java | 147 ++++++++++++++++++ .../session/DefaultSftpSessionFactory.java | 108 ++++++++++++- .../integration/sftp/session/SftpSession.java | 27 +++- .../sftp/outbound/SftpOutboundTests.java | 129 +++++++++++++++ .../SftpServerOutboundTests-context.xml | 15 ++ .../outbound/SftpServerOutboundTests.java | 77 +++++++++ src/reference/docbook/ftp.xml | 16 +- src/reference/docbook/sftp.xml | 43 ++++- src/reference/docbook/whats-new.xml | 10 ++ 12 files changed, 649 insertions(+), 19 deletions(-) create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/SharedSessionCapable.java create mode 100644 spring-integration-file/src/test/java/org/springframework/integration/file/remote/session/CachingSessionFactoryTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java b/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java index 2645c21eda..5814efb0bc 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java @@ -192,7 +192,7 @@ public class SimplePool implements Pool { } else if (this.callback.isStale(item)) { if (logger.isDebugEnabled()) { - logger.debug("Received a stale item, will attempt to get a new one."); + logger.debug("Received a stale item " + item + ", will attempt to get a new one."); } doRemoveItem(item); item = doGetItem(); @@ -241,6 +241,9 @@ public class SimplePool implements Pool { } private void doRemoveItem(T item) { + if (logger.isDebugEnabled()){ + logger.debug("Removing " + item + " from the pool"); + } this.allocated.remove(item); this.inUse.remove(item); this.callback.removedFromPool(item); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java index 77c1363c8f..d28e8762c8 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java @@ -45,6 +45,10 @@ public class CachingSessionFactory implements SessionFactory, DisposableBe private final SimplePool> pool; + private final boolean isSharedSessionCapable; + + private volatile long sharedSessionEpoch; + /** * Create a CachingSessionFactory with an unlimited number of sessions. * @param sessionFactory the underlying session factory. @@ -80,6 +84,7 @@ public class CachingSessionFactory implements SessionFactory, DisposableBe session.close(); } }); + this.isSharedSessionCapable = sessionFactory instanceof SharedSessionCapable; } @@ -105,7 +110,7 @@ public class CachingSessionFactory implements SessionFactory, DisposableBe */ @Override public Session getSession() { - return new CachedSession(this.pool.getItem()); + return new CachedSession(this.pool.getItem(), this.sharedSessionEpoch); } /** @@ -116,27 +121,64 @@ public class CachingSessionFactory implements SessionFactory, DisposableBe this.pool.removeAllIdleItems(); } + /** + * Clear the cache of sessions; also any in-use sessions will be closed when + * returned to the cache. + */ + public synchronized void resetCache() { + if (logger.isDebugEnabled()) { + logger.debug("Cache reset; idle sessions will be removed, in-use sessions will be closed when returned"); + } + if (this.isSharedSessionCapable && ((SharedSessionCapable) this.sessionFactory).isSharedSession()) { + ((SharedSessionCapable) this.sessionFactory).resetSharedSession(); + } + long sharedSessionEpoch = System.nanoTime(); + /* + * Spin until we get a new value - nano precision but may be lower resolution. + * We reset the epoch AFTER resetting the shared session so there is no possibility + * of an "old" session being created in the new epoch. There is a slight possibility + * that a "new" session might appear in the old epoch and thus be closed when returned to + * the cache. + */ + while (sharedSessionEpoch == this.sharedSessionEpoch) { + sharedSessionEpoch = System.nanoTime(); + } + this.sharedSessionEpoch = sharedSessionEpoch; + this.pool.removeAllIdleItems(); + } private class CachedSession implements Session { private final Session targetSession; - private boolean released; + private volatile boolean released; - private CachedSession(Session targetSession) { + /** + * The epoch in which this session was created. + */ + private final long sharedSessionEpoch; + + private CachedSession(Session targetSession, long sharedSessionEpoch) { this.targetSession = targetSession; + this.sharedSessionEpoch = sharedSessionEpoch; } @Override public synchronized void close() { if (released) { if (logger.isDebugEnabled()){ - logger.debug("Session already released."); + logger.debug("Session " + targetSession + " already released."); } } else { if (logger.isDebugEnabled()){ - logger.debug("Releasing Session back to the pool."); + logger.debug("Releasing Session " + targetSession + " back to the pool."); + } + if (this.sharedSessionEpoch != CachingSessionFactory.this.sharedSessionEpoch) { + if (logger.isDebugEnabled()){ + logger.debug("Closing session " + targetSession + " after reset."); + } + this.targetSession.close(); } pool.releaseItem(targetSession); released = true; diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/SharedSessionCapable.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/SharedSessionCapable.java new file mode 100644 index 0000000000..f9db8f99bd --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/SharedSessionCapable.java @@ -0,0 +1,39 @@ +/* + * Copyright 2013 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.file.remote.session; + + +/** + * A {@link SessionFactory} that implements this interface is capable of supporting a shared session. + * + * @author Gary Russell + * @since 3.0 + * + */ +public interface SharedSessionCapable { + + /** + * @return true if this factory uses a shared session. + */ + public abstract boolean isSharedSession(); + + /** + * Resets the shared session so the next {@code #getSession()} will return a session + * using a new connection. + */ + public abstract void resetSharedSession(); + +} diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/session/CachingSessionFactoryTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/session/CachingSessionFactoryTests.java new file mode 100644 index 0000000000..f0cca26600 --- /dev/null +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/session/CachingSessionFactoryTests.java @@ -0,0 +1,147 @@ +/* + * Copyright 2013 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.file.remote.session; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; + +import org.junit.Test; + +import org.springframework.integration.test.util.TestUtils; + +/** + * @author Gary Russell + * @since 3.0 + * + */ +public class CachingSessionFactoryTests { + + @Test + public void testCacheAndReset() { + TestSessionFactory factory = new TestSessionFactory(); + CachingSessionFactory cache = new CachingSessionFactory(factory); + Session sess1 = cache.getSession(); + assertEquals("session:1", TestUtils.getPropertyValue(sess1, "targetSession.id")); + Session sess2 = cache.getSession(); + assertEquals("session:2", TestUtils.getPropertyValue(sess2, "targetSession.id")); + sess1.close(); + // session back to pool; should be open and reused. + assertTrue(sess1.isOpen()); + sess1 = cache.getSession(); + assertEquals("session:1", TestUtils.getPropertyValue(sess1, "targetSession.id")); + sess1.close(); + assertTrue(sess1.isOpen()); + // reset the cache; should close idle (sess1); sess2 should closed later + cache.resetCache(); + assertFalse(sess1.isOpen()); + sess1 = cache.getSession(); + assertEquals("session:3", TestUtils.getPropertyValue(sess1, "targetSession.id")); + sess1.close(); + assertTrue(sess1.isOpen()); + // session from previous epoch is closed on return + sess2.close(); + assertFalse(sess2.isOpen()); + cache.resetCache(); + assertFalse(sess1.isOpen()); + } + + private class TestSessionFactory implements SessionFactory { + + private int n; + + @Override + public Session getSession() { + return new TestSession("session:" + ++n); + } + + } + + private class TestSession implements Session { + + @SuppressWarnings("unused") + private final String id; + + private volatile boolean open = true; + + private TestSession(String id) { + this.id = id; + } + + @Override + public boolean remove(String path) throws IOException { + return false; + } + + @Override + public String[] list(String path) throws IOException { + return null; + } + + @Override + public void read(String source, OutputStream outputStream) throws IOException { + } + + @Override + public void write(InputStream inputStream, String destination) throws IOException { + } + + @Override + public boolean mkdir(String directory) throws IOException { + return false; + } + + @Override + public void rename(String pathFrom, String pathTo) throws IOException { + } + + @Override + public void close() { + this.open = false; + } + + @Override + public boolean isOpen() { + return this.open; + } + + @Override + public boolean exists(String path) throws IOException { + return false; + } + + @Override + public String[] listNames(String path) throws IOException { + return null; + } + + @Override + public InputStream readRaw(String source) throws IOException { + return null; + } + + @Override + public boolean finalizeRaw() throws IOException { + return false; + } + + } + +} diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java index 3f33c12e32..9bad196505 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -17,11 +17,14 @@ package org.springframework.integration.sftp.session; import java.util.Properties; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.ReentrantReadWriteLock; import org.springframework.beans.factory.BeanCreationException; import org.springframework.core.io.Resource; import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.file.remote.session.SharedSessionCapable; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -39,10 +42,11 @@ import com.jcraft.jsch.UserInfo; * @author Mario Gray * @author Oleg Zhurakousky * @author Gunnar Hillert + * @author Gary Russell * * @since 2.0 */ -public class DefaultSftpSessionFactory implements SessionFactory { +public class DefaultSftpSessionFactory implements SessionFactory, SharedSessionCapable { private volatile String host; @@ -76,9 +80,35 @@ public class DefaultSftpSessionFactory implements SessionFactory { private volatile Boolean enableDaemonThread; + private final JSch jsch; - private final JSch jsch = new JSch(); + private final boolean isSharedSession; + private volatile JSchSessionWrapper sharedJschSession; + + private final ReentrantReadWriteLock sharedSessionLock = new ReentrantReadWriteLock(); + + + public DefaultSftpSessionFactory() { + this(false); + } + + /** + * @param isSharedSession + */ + public DefaultSftpSessionFactory(boolean isSharedSession) { + this(new JSch(), isSharedSession); + } + + /** + * Intended for use in tests so the jsch can be mocked. + * @param jsch + * @param isSharedSession + */ + public DefaultSftpSessionFactory(JSch jsch, boolean isSharedSession) { + this.jsch = jsch; + this.isSharedSession = isSharedSession; + } /** * The url of the host you want connect to. This is a mandatory property. @@ -257,9 +287,35 @@ public class DefaultSftpSessionFactory implements SessionFactory { Assert.isTrue(StringUtils.hasText(this.password) || this.privateKey != null, "either a password or a private key is required"); try { - com.jcraft.jsch.Session jschSession = this.initJschSession(); + JSchSessionWrapper jschSession; + if (this.isSharedSession) { + this.sharedSessionLock.readLock().lock(); + try { + if (this.sharedJschSession == null || !this.sharedJschSession.isConnected()) { + this.sharedSessionLock.readLock().unlock(); + this.sharedSessionLock.writeLock().lock(); + try { + if (this.sharedJschSession == null || !this.sharedJschSession.isConnected()) { + this.sharedJschSession = new JSchSessionWrapper(initJschSession()); + } + } + finally { + this.sharedSessionLock.readLock().lock(); + this.sharedSessionLock.writeLock().unlock(); + } + } + } + finally { + this.sharedSessionLock.readLock().unlock(); + } + jschSession = this.sharedJschSession; + } + else { + jschSession = new JSchSessionWrapper(initJschSession()); + } SftpSession sftpSession = new SftpSession(jschSession); sftpSession.connect(); + jschSession.addChannel(); return sftpSession; } catch (Exception e) { @@ -327,6 +383,16 @@ public class DefaultSftpSessionFactory implements SessionFactory { return jschSession; } + @Override + public final boolean isSharedSession() { + return this.isSharedSession; + } + + @Override + public void resetSharedSession() { + Assert.state(this.isSharedSession, "Shared sessions are not being used"); + this.sharedJschSession = null; + } /** * this is a simple, optimistic implementation of the UserInfo interface. @@ -370,4 +436,38 @@ public class DefaultSftpSessionFactory implements SessionFactory { } } + /** + * A wrapper for a JSch session that maintains a channel count and + * physically disconnects when the last channel is closed. + * + */ + public class JSchSessionWrapper { + + private final com.jcraft.jsch.Session session; + + private final AtomicInteger channels = new AtomicInteger(); + + JSchSessionWrapper(com.jcraft.jsch.Session session) { + this.session = session; + } + + public void addChannel() { + this.channels.incrementAndGet(); + } + + public void close() { + if (channels.decrementAndGet() <= 0) { + this.session.disconnect(); + } + } + + public final com.jcraft.jsch.Session getSession() { + return session; + } + + public boolean isConnected() { + return session.isConnected(); + } + + } } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java index a3d47250af..aadd108ae6 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java @@ -28,6 +28,7 @@ import org.apache.commons.logging.LogFactory; import org.springframework.core.NestedIOException; import org.springframework.integration.file.remote.session.Session; +import org.springframework.integration.sftp.session.DefaultSftpSessionFactory.JSchSessionWrapper; import org.springframework.util.Assert; import org.springframework.util.FileCopyUtils; @@ -53,14 +54,24 @@ class SftpSession implements Session { private final com.jcraft.jsch.Session jschSession; + private final JSchSessionWrapper wrapper; + private volatile ChannelSftp channel; + private volatile boolean closed; + public SftpSession(com.jcraft.jsch.Session jschSession) { Assert.notNull(jschSession, "jschSession must not be null"); this.jschSession = jschSession; + this.wrapper = null; } + public SftpSession(DefaultSftpSessionFactory.JSchSessionWrapper wrapper) { + Assert.notNull(wrapper, "wrapper must not be null"); + this.jschSession = wrapper.getSession(); + this.wrapper = wrapper; + } @Override public boolean remove(String path) throws IOException { @@ -151,14 +162,21 @@ class SftpSession implements Session { @Override public void close() { - if (this.jschSession.isConnected()) { - this.jschSession.disconnect(); + this.closed = true; + if (this.wrapper != null) { + this.channel.disconnect(); + this.wrapper.close(); + } + else { + if (this.jschSession.isConnected()) { + this.jschSession.disconnect(); + } } } @Override public boolean isOpen() { - return this.jschSession.isConnected(); + return !this.closed && this.jschSession.isConnected(); } @Override @@ -220,8 +238,8 @@ class SftpSession implements Session { try { if (!this.jschSession.isConnected()) { this.jschSession.connect(); - this.channel = (ChannelSftp) this.jschSession.openChannel("sftp"); } + this.channel = (ChannelSftp) this.jschSession.openChannel("sftp"); if (this.channel != null && !this.channel.isConnected()) { this.channel.connect(); } @@ -230,4 +248,5 @@ class SftpSession implements Session { throw new IllegalStateException("failed to connect", e); } } + } diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpOutboundTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpOutboundTests.java index aa3f584584..76950c77d3 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpOutboundTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpOutboundTests.java @@ -17,15 +17,21 @@ package org.springframework.integration.sftp.outbound; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import static org.mockito.Matchers.anyString; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import java.io.File; import java.io.FileOutputStream; import java.io.InputStream; +import java.lang.reflect.Constructor; import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -36,6 +42,7 @@ import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanFactory; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; @@ -46,16 +53,20 @@ import org.springframework.integration.core.PollableChannel; import org.springframework.integration.file.DefaultFileNameGenerator; import org.springframework.integration.file.remote.FileInfo; import org.springframework.integration.file.remote.handler.FileTransferringMessageHandler; +import org.springframework.integration.file.remote.session.CachingSessionFactory; import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.sftp.session.DefaultSftpSessionFactory; import org.springframework.integration.sftp.session.SftpTestSessionFactory; import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.test.util.TestUtils; import org.springframework.util.FileCopyUtils; import com.jcraft.jsch.ChannelSftp; import com.jcraft.jsch.ChannelSftp.LsEntry; +import com.jcraft.jsch.JSch; +import com.jcraft.jsch.JSchException; import com.jcraft.jsch.SftpATTRS; /** @@ -194,6 +205,7 @@ public class SftpOutboundTests { handler.afterPropertiesSet(); final List madeDirs = new ArrayList(); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { madeDirs.add((String) invocation.getArguments()[0]); return null; @@ -206,6 +218,121 @@ public class SftpOutboundTests { assertEquals("/foo/bar/baz", madeDirs.get(2)); } + @Test + public void testSharedSession() throws Exception { + JSch jsch = spy(new JSch()); + Constructor ctor = com.jcraft.jsch.Session.class.getDeclaredConstructor(JSch.class); + ctor.setAccessible(true); + com.jcraft.jsch.Session jschSession1 = spy(ctor.newInstance(jsch)); + com.jcraft.jsch.Session jschSession2 = spy(ctor.newInstance(jsch)); + new DirectFieldAccessor(jschSession1).setPropertyValue("isConnected", true); + new DirectFieldAccessor(jschSession2).setPropertyValue("isConnected", true); + when(jsch.getSession("foo", "host", 22)).thenReturn(jschSession1, jschSession2); + ChannelSftp channel1 = spy(new ChannelSftp()); + ChannelSftp channel2 = spy(new ChannelSftp()); + new DirectFieldAccessor(channel1).setPropertyValue("session", jschSession1); + new DirectFieldAccessor(channel2).setPropertyValue("session", jschSession1); + when(jschSession1.openChannel("sftp")).thenReturn(channel1, channel2); + DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(jsch, true); + factory.setHost("host"); + factory.setUser("foo"); + factory.setPassword("bar"); + noopConnect(channel1); + noopConnect(channel2); + Session s1 = factory.getSession(); + Session s2 = factory.getSession(); + assertSame(TestUtils.getPropertyValue(s1, "jschSession"), TestUtils.getPropertyValue(s2, "jschSession")); + } + + @Test + public void testNotSharedSession() throws Exception { + JSch jsch = spy(new JSch()); + Constructor ctor = com.jcraft.jsch.Session.class.getDeclaredConstructor(JSch.class); + ctor.setAccessible(true); + com.jcraft.jsch.Session jschSession1 = spy(ctor.newInstance(jsch)); + com.jcraft.jsch.Session jschSession2 = spy(ctor.newInstance(jsch)); + new DirectFieldAccessor(jschSession1).setPropertyValue("isConnected", true); + new DirectFieldAccessor(jschSession2).setPropertyValue("isConnected", true); + when(jsch.getSession("foo", "host", 22)).thenReturn(jschSession1, jschSession2); + ChannelSftp channel1 = spy(new ChannelSftp()); + ChannelSftp channel2 = spy(new ChannelSftp()); + new DirectFieldAccessor(channel1).setPropertyValue("session", jschSession1); + new DirectFieldAccessor(channel2).setPropertyValue("session", jschSession1); + when(jschSession1.openChannel("sftp")).thenReturn(channel1); + when(jschSession2.openChannel("sftp")).thenReturn(channel2); + DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(jsch, false); + factory.setHost("host"); + factory.setUser("foo"); + factory.setPassword("bar"); + noopConnect(channel1); + noopConnect(channel2); + Session s1 = factory.getSession(); + Session s2 = factory.getSession(); + assertNotSame(TestUtils.getPropertyValue(s1, "jschSession"), TestUtils.getPropertyValue(s2, "jschSession")); + } + + @Test + public void testSharedSessionCachedReset() throws Exception { + JSch jsch = spy(new JSch()); + Constructor ctor = com.jcraft.jsch.Session.class.getDeclaredConstructor(JSch.class); + ctor.setAccessible(true); + com.jcraft.jsch.Session jschSession1 = spy(ctor.newInstance(jsch)); + com.jcraft.jsch.Session jschSession2 = spy(ctor.newInstance(jsch)); + new DirectFieldAccessor(jschSession1).setPropertyValue("isConnected", true); + new DirectFieldAccessor(jschSession2).setPropertyValue("isConnected", true); + when(jsch.getSession("foo", "host", 22)).thenReturn(jschSession1, jschSession2); + ChannelSftp channel1 = spy(new ChannelSftp()); + ChannelSftp channel2 = spy(new ChannelSftp()); + ChannelSftp channel3 = spy(new ChannelSftp()); + ChannelSftp channel4 = spy(new ChannelSftp()); + new DirectFieldAccessor(channel1).setPropertyValue("session", jschSession1); + new DirectFieldAccessor(channel2).setPropertyValue("session", jschSession1); + when(jschSession1.openChannel("sftp")).thenReturn(channel1, channel2); + when(jschSession2.openChannel("sftp")).thenReturn(channel3, channel4); + DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(jsch, true); + factory.setHost("host"); + factory.setUser("foo"); + factory.setPassword("bar"); + CachingSessionFactory cachedFactory = new CachingSessionFactory(factory); + noopConnect(channel1); + noopConnect(channel2); + noopConnect(channel3); + noopConnect(channel4); + Session s1 = cachedFactory.getSession(); + Session s2 = cachedFactory.getSession(); + assertSame(jschSession1, TestUtils.getPropertyValue(s2, "targetSession.jschSession")); + assertSame(TestUtils.getPropertyValue(s1, "targetSession.jschSession"), TestUtils.getPropertyValue(s2, "targetSession.jschSession")); + s1.close(); + Session s3 = cachedFactory.getSession(); + assertSame(TestUtils.getPropertyValue(s1, "targetSession"), TestUtils.getPropertyValue(s3, "targetSession")); + s3.close(); + cachedFactory.resetCache(); + verify(jschSession1, never()).disconnect(); + s3 = cachedFactory.getSession(); + assertSame(jschSession2, TestUtils.getPropertyValue(s3, "targetSession.jschSession")); + assertNotSame(TestUtils.getPropertyValue(s1, "targetSession"), TestUtils.getPropertyValue(s3, "targetSession")); + s2.close(); + verify(jschSession1).disconnect(); + s2 = cachedFactory.getSession(); + assertSame(jschSession2, TestUtils.getPropertyValue(s2, "targetSession.jschSession")); + assertNotSame(TestUtils.getPropertyValue(s3, "targetSession"), TestUtils.getPropertyValue(s2, "targetSession")); + s2.close(); + s3.close(); + verify(jschSession2, never()).disconnect(); + cachedFactory.resetCache(); + verify(jschSession2).disconnect(); + } + + private void noopConnect(ChannelSftp channel1) throws JSchException { + doAnswer(new Answer() { + + @Override + public Object answer(InvocationOnMock invocation) throws Throwable { + return null; + } + }).when(channel1).connect(); + } + public static class TestSftpSessionFactory extends DefaultSftpSessionFactory { @Override @@ -214,6 +341,7 @@ public class SftpOutboundTests { ChannelSftp channel = mock(ChannelSftp.class); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { File file = new File((String)invocation.getArguments()[1]); @@ -225,6 +353,7 @@ public class SftpOutboundTests { }).when(channel).put(Mockito.any(InputStream.class), Mockito.anyString()); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { File file = new File((String) invocation.getArguments()[0]); diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests-context.xml b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests-context.xml index da0b9d7d55..cf054cdccd 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests-context.xml +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests-context.xml @@ -77,4 +77,19 @@ + + + + + + + + + + + + + diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java index 3e5d978499..613b191496 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java @@ -18,6 +18,7 @@ package org.springframework.integration.sftp.outbound; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -27,7 +28,12 @@ import static org.mockito.Mockito.when; import java.io.ByteArrayOutputStream; import java.io.File; import java.io.IOException; +import java.io.PipedInputStream; +import java.io.PipedOutputStream; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import org.hamcrest.Matchers; import org.junit.After; @@ -44,6 +50,7 @@ import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.sftp.session.SftpFileInfo; +import org.springframework.integration.test.util.TestUtils; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.util.FileCopyUtils; @@ -160,6 +167,11 @@ public class SftpServerOutboundTests { @Test public void testInt2866LocalDirectoryExpressionGET() { + Session session = null; + boolean sharedSession = "realSSHSharedSession".equals(System.getProperty("spring.profiles.active")); + if (sharedSession) { + session = this.sessionFactory.getSession(); + } String dir = "sftpSource/"; this.inboundGet.send(new GenericMessage(dir + "sftpSource1.txt")); Message result = this.output.receive(1000); @@ -175,6 +187,11 @@ public class SftpServerOutboundTests { localFile = (File) result.getPayload(); assertThat(localFile.getPath().replaceAll(java.util.regex.Matcher.quoteReplacement(File.separator), "/"), Matchers.containsString(dir.toUpperCase())); + if (sharedSession) { + Session session2 = this.sessionFactory.getSession(); + assertSame(TestUtils.getPropertyValue(session, "targetSession.jschSession"), + TestUtils.getPropertyValue(session2, "targetSession.jschSession")); + } } @Test @@ -276,5 +293,65 @@ public class SftpServerOutboundTests { } } + @Test + public void testInt3047ConcurrentSharedSession() throws Exception { + if ("realSSHSharedSession".equals(System.getProperty("spring.profiles.active"))) { + final Session session1 = this.sessionFactory.getSession(); + final Session session2 = this.sessionFactory.getSession(); + final PipedInputStream pipe1 = new PipedInputStream(); + PipedOutputStream out1 = new PipedOutputStream(pipe1); + final PipedInputStream pipe2 = new PipedInputStream(); + PipedOutputStream out2 = new PipedOutputStream(pipe2); + final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(1); + Executors.newSingleThreadExecutor().execute(new Runnable() { + + @Override + public void run() { + try { + session1.write(pipe1, "foo.txt"); + } + catch (IOException e) { + e.printStackTrace(); + } + latch1.countDown(); + } + }); + Executors.newSingleThreadExecutor().execute(new Runnable() { + + @Override + public void run() { + try { + session2.write(pipe2, "bar.txt"); + } + catch (IOException e) { + e.printStackTrace(); + } + latch2.countDown(); + } + }); + + out1.write('a'); + out2.write('b'); + out1.write('c'); + out2.write('d'); + out1.write('e'); + out2.write('f'); + out1.close(); + out2.close(); + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + assertTrue(latch2.await(10, TimeUnit.SECONDS)); + ByteArrayOutputStream bos1 = new ByteArrayOutputStream(); + ByteArrayOutputStream bos2 = new ByteArrayOutputStream(); + session1.read("foo.txt", bos1); + session2.read("bar.txt", bos2); + assertEquals("ace", new String(bos1.toByteArray())); + assertEquals("bdf", new String(bos2.toByteArray())); + session1.remove("foo.txt"); + session2.remove("bar.txt"); + session1.close(); + session2.close(); + } + } } diff --git a/src/reference/docbook/ftp.xml b/src/reference/docbook/ftp.xml index 1e04a9960f..42747dd15d 100644 --- a/src/reference/docbook/ftp.xml +++ b/src/reference/docbook/ftp.xml @@ -154,6 +154,7 @@ protected void postProcessClientBeforeConnect(T client) throws IOException { filename-pattern="*.txt" remote-directory="some/remote/path" remote-file-separator="/" + preserve-timestamp="true" local-filename-generator-expression="#this.toUpperCase() + '.a'" local-filter="myFilter" local-directory="."> @@ -172,7 +173,11 @@ protected void postProcessClientBeforeConnect(T client) throws IOException { that's what it ultimately generates with the transferred file as its payload. So, the root object of the SpEL Evaluation Context is the original name of the remote file (String). - + + Starting with Spring Integration 3.0, you can specify the preserve-timestamp + attribute (default false); when true, the local file's modified timestamp will be set to the value + retrieved from the server; otherwise it will be set to the current time. + Sometimes file filtering based on the simple pattern specified via filename-pattern attribute might not be sufficient. If this is the case, you can use the filename-regex attribute to specify a Regular Expression @@ -500,8 +505,8 @@ protected void postProcessClientBeforeConnect(T client) throws IOException {
FTP Session Caching - Starting with version 3.0, sessions are no longer cached by default; the cache-sessions attribute - is no longer supported on endpoints. You must now use a CachingSessionFactory (see below) if you + Starting with Spring Integration version 3.0, sessions are no longer cached by default; the cache-sessions attribute + is no longer supported on endpoints. You must use a CachingSessionFactory (see below) if you wish to cache sessions. @@ -532,5 +537,10 @@ protected void postProcessClientBeforeConnect(T client) throws IOException { sessionCacheSize set to 10 and the sessionWaitTimeout set to 1 second (its value is in millliseconds). + + 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. New requests for sessions will establish new sessions as necessary. +
diff --git a/src/reference/docbook/sftp.xml b/src/reference/docbook/sftp.xml index a73cf9c6d6..5111c75d5f 100644 --- a/src/reference/docbook/sftp.xml +++ b/src/reference/docbook/sftp.xml @@ -59,6 +59,27 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp However, Spring Integration also supports the caching of SFTP sessions, please see 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 false) to use a single connection + to the server and create multiple JSch channels on that single connection. + + + When using this feature, you must wrap the session factory in a caching session + factory, as described below, 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 will be 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 as well as see which Sessions are polled you may enable it by @@ -79,6 +100,11 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp Below you will find all properties that are exposed by the DefaultSftpSessionFactory. + isSharedSession (constructor argument) + + When true, a single connection will be used and JSch Channels will be multiplexed. + Defaults to false. + clientVersion Allows you to set the client version property. It's default @@ -183,8 +209,8 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp
SFTP Session Caching - Starting with version 3.0, sessions are no longer cached by default; the cache-sessions attribute - is no longer supported on endpoints. You must now use a CachingSessionFactory (see below) if you + Starting with Spring Integration version 3.0, sessions are no longer cached by default; the cache-sessions attribute + is no longer supported on endpoints. You must use a CachingSessionFactory (see below) if you wish to cache sessions. @@ -217,6 +243,13 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp sessionCacheSize set to 10 and the sessionWaitTimeout set to 1 second (its value is in millliseconds). + + 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 will establish new sessions as necessary. +
@@ -230,6 +263,7 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp 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'" @@ -252,6 +286,11 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp that's what it ultimately generates with the transferred file as its payload. So, the root object of the SpEL Evaluation Context is the original name of the remote file (String). + + Starting with Spring Integration 3.0, you can specify the preserve-timestamp + attribute (default false); when true, the local file's modified timestamp will be set to the value + retrieved from the server; otherwise it will be set to the current time. + Sometimes file filtering based on the simple pattern specified via filename-pattern attribute might not be sufficient. If this is the case, you can use the filename-regex attribute to specify a Regular Expression diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 95186b1deb..c1271ca497 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -243,6 +243,16 @@ For more information, see and . + + The CachingConnectionFactory now provides a new method + resetCache(). This immediately closes idle sessions and causes in-use + sessions to be closed as and when they are returned to the cache. + + + The DefaultSftpSessionFactory (in conjunction with a + CachingSessionFactory) now supports multiplexing channels over + a single SSH connection (SFTP Only). +
FTP, SFTP and FTPS Inbound Adapters