Merge pull request bcgov#150 from peggy-ntt/sftpFilter

add filter for sftp only process file once.

Author: alexjoybc

README.md

@@ -187,7 +187,9 @@ Sets the maximum attempt to reprocess a message in the queue.
 
 #### Description
 
-Using this plugin you can receive messages from a specified Sftp server.
+Using this plugin you can receive messages from a specified Sftp server when there is a new file.
+This Sftp plugin will only receive a file once, it uses the server's file timestamp to detect if we've already 'processed' this file.
+It needs redis data structure store as Metadata Store.
 
 #### Setup
 
@@ -489,6 +491,10 @@ To view the message in a queue, login to [rabbitmq management console](http://lo
 
 #### If you want to run the sample app using sftp do the following:
 
+step 0. Create a redis container 
+```bash
+docker run --name some-redis -p 6379:6379 -d redis
+```
 step 1. Create a sftp server container (from WindowsPowerShell or GitBash)
 ```bash
 docker run -p 22:22 -d atmoz/sftp myname:pass:::upload

jrcc-access-spring-boot-autoconfigure/pom.xml

@@ -42,6 +42,10 @@
 			<groupId>org.springframework.integration</groupId>
 			<artifactId>spring-integration-sftp</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.integration</groupId>
+			<artifactId>spring-integration-redis</artifactId>
+		</dependency>
 		<dependency>
 			<groupId>org.hibernate.validator</groupId>
 			<artifactId>hibernate-validator</artifactId>

jrcc-access-spring-boot-autoconfigure/src/main/java/ca/bc/gov/open/jrccaccess/autoconfigure/plugins/rabbitmq/RabbitMqDocumentInput.java

@@ -1,6 +1,7 @@
 package ca.bc.gov.open.jrccaccess.autoconfigure.plugins.rabbitmq;
 
 import ca.bc.gov.open.jrccaccess.autoconfigure.common.Constants;
+import ca.bc.gov.open.jrccaccess.autoconfigure.redis.RedisStorageService;
 import ca.bc.gov.open.jrccaccess.autoconfigure.services.DocumentReadyHandler;
 import ca.bc.gov.open.jrccaccess.libs.DocumentReadyMessage;
 import ca.bc.gov.open.jrccaccess.libs.DocumentStorageProperties;

jrcc-access-spring-boot-autoconfigure/src/main/java/ca/bc/gov/open/jrccaccess/autoconfigure/plugins/rabbitmq/RabbitMqDocumentOutput.java

@@ -1,6 +1,7 @@
 package ca.bc.gov.open.jrccaccess.autoconfigure.plugins.rabbitmq;
 
 import ca.bc.gov.open.jrccaccess.autoconfigure.AccessProperties;
+import ca.bc.gov.open.jrccaccess.autoconfigure.redis.RedisStorageService;
 import ca.bc.gov.open.jrccaccess.libs.*;
 import ca.bc.gov.open.jrccaccess.libs.services.exceptions.DocumentMessageException;
 import org.slf4j.Logger;

jrcc-access-spring-boot-autoconfigure/src/main/java/ca/bc/gov/open/jrccaccess/autoconfigure/plugins/sftp/AutoConfiguration.java

@@ -4,6 +4,7 @@
 import ca.bc.gov.open.jrccaccess.autoconfigure.config.exceptions.KnownHostFileNotDefinedException;
 import ca.bc.gov.open.jrccaccess.autoconfigure.config.exceptions.KnownHostFileNotFoundException;
 import com.jcraft.jsch.ChannelSftp;
+import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
@@ -15,8 +16,11 @@
 import org.springframework.integration.annotation.Poller;
 import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.integration.core.MessageSource;
+import org.springframework.integration.file.filters.ChainFileListFilter;
 import org.springframework.integration.file.remote.session.CachingSessionFactory;
 import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.integration.metadata.ConcurrentMetadataStore;
+import org.springframework.integration.sftp.filters.SftpPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter;
 import org.springframework.integration.sftp.inbound.SftpStreamingMessageSource;
 import org.springframework.integration.sftp.session.DefaultSftpSessionFactory;
@@ -30,7 +34,7 @@
 @ComponentScan
 @EnableConfigurationProperties(SftpInputProperties.class)
 @ConditionalOnProperty(
-        value="bcgov.access.input.plugin",
+        value = "bcgov.access.input.plugin",
         havingValue = "sftp"
 )
 public class AutoConfiguration {
@@ -41,7 +45,6 @@ public class AutoConfiguration {
 
     public AutoConfiguration(SftpInputProperties sftpInputProperties) {
         this.properties = sftpInputProperties;
-
         logger.debug("SFTP Configuration: Host => [{}]", this.properties.getHost());
         logger.debug("SFTP Configuration: Port => [{}]", this.properties.getPort());
         logger.debug("SFTP Configuration: Username => [{}]", this.properties.getUsername());
@@ -50,10 +53,8 @@ public AutoConfiguration(SftpInputProperties sftpInputProperties) {
         logger.debug("SFTP Configuration: Cron => [{}]", this.properties.getCron());
         logger.debug("SFTP Configuration: Max Message Per Poll => [{}]", this.properties.getMaxMessagePerPoll());
         logger.debug("SFTP Configuration: Known Host File => [{}]", this.properties.getKnownHostFile());
-
     }
 
-
     @Bean
     public SessionFactory<ChannelSftp.LsEntry> sftpSessionFactory() throws InvalidConfigException {
         DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
@@ -68,13 +69,13 @@ public SessionFactory<ChannelSftp.LsEntry> sftpSessionFactory() throws InvalidCo
         }
         boolean isAllowUnknownKeys = properties.isAllowUnknownKeys();
         factory.setAllowUnknownKeys(isAllowUnknownKeys);
-        if(!isAllowUnknownKeys){
+        if (!isAllowUnknownKeys) {
             String knownHostFileStr = properties.getKnownHostFile();
-            if(knownHostFileStr == null || knownHostFileStr.equals("") )
+            if (StringUtils.isBlank(knownHostFileStr))
                 throw new KnownHostFileNotDefinedException("Must define known_hosts file when allow-unknown-keys is false. ");
 
             File knownHostFile = new File(knownHostFileStr);
-            if( ! knownHostFile.exists() )
+            if (!knownHostFile.exists())
                 throw new KnownHostFileNotFoundException("Cannot find known_hosts file when allow-unknown-keys is false.");
 
             factory.setKnownHosts(properties.getKnownHostFile());
@@ -87,20 +88,23 @@ public SessionFactory<ChannelSftp.LsEntry> sftpSessionFactory() throws InvalidCo
     public SftpRemoteFileTemplate template() {
         try {
             return new SftpRemoteFileTemplate(sftpSessionFactory());
-        }catch(InvalidConfigException ex)
-        {
+        } catch (InvalidConfigException ex) {
             logger.error(ex.getMessage());
         }
         return null;
     }
 
     @Bean
     @InboundChannelAdapter(channel = "sftpChannel", poller = @Poller(cron = "${bcgov.access.input.sftp.cron}", maxMessagesPerPoll = "${bcgov.access.input.sftp.max-message-per-poll}"))
-    public MessageSource<InputStream> sftpMessageSource() {
+    public MessageSource<InputStream> sftpMessageSource(ConcurrentMetadataStore concurrentMetadataStore) {
+        ChainFileListFilter<ChannelSftp.LsEntry> filterChain = new ChainFileListFilter<>();
+        if (properties.getFilterPattern() != null && !"".equals(properties.getFilterPattern()))
+            filterChain.addFilter(new SftpRegexPatternFileListFilter(properties.getFilterPattern()));
+        filterChain.addFilter(new SftpPersistentAcceptOnceFileListFilter(concurrentMetadataStore, "sftpSource"));
         SftpStreamingMessageSource messageSource = new SftpStreamingMessageSource(template());
         messageSource.setRemoteDirectory(properties.getRemoteDirectory());
-        if(properties.getFilterPattern() != null && !"".equals(properties.getFilterPattern()))
-            messageSource.setFilter(new SftpRegexPatternFileListFilter(properties.getFilterPattern()));
+        messageSource.setFilter(filterChain);
+
         return messageSource;
     }
 

jrcc-access-spring-boot-autoconfigure/src/main/java/ca/bc/gov/open/jrccaccess/autoconfigure/plugins/rabbitmq/RedisConfiguration.java renamed to jrcc-access-spring-boot-autoconfigure/src/main/java/ca/bc/gov/open/jrccaccess/autoconfigure/redis/AutoConfiguration.java

@@ -1,11 +1,13 @@
-package ca.bc.gov.open.jrccaccess.autoconfigure.plugins.rabbitmq;
+package ca.bc.gov.open.jrccaccess.autoconfigure.redis;
 
+import ca.bc.gov.open.jrccaccess.autoconfigure.plugins.rabbitmq.RabbitMqOutputProperties;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
 import org.springframework.boot.autoconfigure.data.redis.RedisProperties.Sentinel;
 import org.springframework.cache.CacheManager;
 import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.ComponentScan;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.data.redis.cache.RedisCacheConfiguration;
 import org.springframework.data.redis.cache.RedisCacheManager;
@@ -14,21 +16,22 @@
 import org.springframework.data.redis.connection.RedisSentinelConfiguration;
 import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
 import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
+import org.springframework.integration.metadata.ConcurrentMetadataStore;
+import org.springframework.integration.redis.metadata.RedisMetadataStore;
 
 import javax.naming.OperationNotSupportedException;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 
-
 /**
  * Redis configuration properties
  * @author alexjoybc
  * @since 0.4.0
  */
 @Configuration
-@ConditionalOnExpression("'${bcgov.access.input.plugin}' == 'rabbitmq' || '${bcgov.access.output.plugin}' == 'rabbitmq'")
-public class RedisConfiguration {
+@ComponentScan
+public class AutoConfiguration {
 
 	/**
 	 * Configure the JedisConnectionFactory
@@ -38,9 +41,9 @@ public class RedisConfiguration {
 	 */
 	@Bean
 	@ConditionalOnMissingBean(JedisConnectionFactory.class)
+	@ConditionalOnExpression("'${bcgov.access.input.plugin}' == 'rabbitmq' || '${bcgov.access.output.plugin}' == 'rabbitmq' || '${bcgov.access.input.plugin}' == 'sftp'")
 	public JedisConnectionFactory jedisConnectionFactory(RedisProperties properties) {
-		
-		
+
 		if(properties.getCluster() != null) {
 			RedisClusterConfiguration redisClusterConfiguration = new RedisClusterConfiguration(properties.getCluster().getNodes());
 			redisClusterConfiguration.setPassword(properties.getPassword());
@@ -65,8 +68,7 @@ public JedisConnectionFactory jedisConnectionFactory(RedisProperties properties)
 		redisStandaloneConfiguration.setPassword(properties.getPassword());
 		return new JedisConnectionFactory(redisStandaloneConfiguration);
 	}
-	
-	
+
 	private List<RedisNode> createSentinels(Sentinel sentinel) {
         List<RedisNode> nodes = new ArrayList<>();
         for (String node : sentinel.getNodes()) {
@@ -90,6 +92,7 @@ private List<RedisNode> createSentinels(Sentinel sentinel) {
 	 */
 	@Bean(name = "Document")
 	@ConditionalOnMissingBean(CacheManager.class)
+	@ConditionalOnExpression("'${bcgov.access.input.plugin}' == 'rabbitmq' || '${bcgov.access.output.plugin}' == 'rabbitmq'")
     public CacheManager cacheManager(JedisConnectionFactory jedisConnectionFactory, RabbitMqOutputProperties rabbitMqOutputProperties) {
 
 		RedisCacheConfiguration redisCacheConfiguration = RedisCacheConfiguration.defaultCacheConfig()
@@ -100,5 +103,11 @@ public CacheManager cacheManager(JedisConnectionFactory jedisConnectionFactory,
 	   return RedisCacheManager.RedisCacheManagerBuilder.fromConnectionFactory(jedisConnectionFactory)
 	                    .cacheDefaults(redisCacheConfiguration).build();
     }
-	
+
+    @Bean
+	@ConditionalOnMissingBean(ConcurrentMetadataStore.class)
+	@ConditionalOnExpression("'${bcgov.access.input.plugin}' == 'sftp'")
+	public ConcurrentMetadataStore redisMetadataStore(JedisConnectionFactory jedisConnectionFactory){
+		return new RedisMetadataStore(jedisConnectionFactory);
+	}
 }

jrcc-access-spring-boot-autoconfigure/src/main/java/ca/bc/gov/open/jrccaccess/autoconfigure/plugins/rabbitmq/RedisStorageService.java renamed to jrcc-access-spring-boot-autoconfigure/src/main/java/ca/bc/gov/open/jrccaccess/autoconfigure/redis/RedisStorageService.java

@@ -1,4 +1,4 @@
-package ca.bc.gov.open.jrccaccess.autoconfigure.plugins.rabbitmq;
+package ca.bc.gov.open.jrccaccess.autoconfigure.redis;
 
 import ca.bc.gov.open.jrccaccess.autoconfigure.AccessProperties;
 import ca.bc.gov.open.jrccaccess.libs.DocumentStorageProperties;
@@ -101,7 +101,7 @@ public String getString(String key, String digest) throws DocumentMessageExcepti
 	public Boolean deleteString(String key) throws DocumentMessageException {
 
 		try {
-                        this.cacheManager.getCache(accessProperties.getInput().getDocumentType()).evict(key);
+			this.cacheManager.getCache(accessProperties.getInput().getDocumentType()).evict(key);
 			return true;
 		} catch (RedisConnectionFailureException e) {
 			throw new DocumentMessageException(serviceUnavailableMessage, e.getCause());

jrcc-access-spring-boot-autoconfigure/src/main/resources/META-INF/spring.factories

@@ -1,5 +1,6 @@
 org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
 ca.bc.gov.open.jrccaccess.autoconfigure.AccessAutoConfiguration,\
+ca.bc.gov.open.jrccaccess.autoconfigure.redis.AutoConfiguration,\
 ca.bc.gov.open.jrccaccess.autoconfigure.plugins.rabbitmq.AutoConfiguration,\
 ca.bc.gov.open.jrccaccess.autoconfigure.plugins.console.AutoConfiguration,\
 ca.bc.gov.open.jrccaccess.autoconfigure.plugins.http.AutoConfiguration, \

jrcc-access-spring-boot-autoconfigure/src/test/java/ca/bc/gov/open/jrccaccess/autoconfigure/plugins/rabbitmq/RabbitMqDocumentInputTests.java

@@ -2,6 +2,7 @@
 
 import ca.bc.gov.open.jrccaccess.autoconfigure.AccessProperties;
 import ca.bc.gov.open.jrccaccess.autoconfigure.helpers.RandomHelper;
+import ca.bc.gov.open.jrccaccess.autoconfigure.redis.RedisStorageService;
 import ca.bc.gov.open.jrccaccess.autoconfigure.services.DocumentReadyHandler;
 import ca.bc.gov.open.jrccaccess.libs.DocumentReadyMessage;
 import ca.bc.gov.open.jrccaccess.libs.DocumentStorageProperties;

jrcc-access-spring-boot-autoconfigure/src/test/java/ca/bc/gov/open/jrccaccess/autoconfigure/plugins/rabbitmq/RabbitMqDocumentOutputTests.java

@@ -2,6 +2,7 @@
 
 import ca.bc.gov.open.jrccaccess.autoconfigure.AccessProperties;
 import ca.bc.gov.open.jrccaccess.autoconfigure.AccessProperties.PluginConfig;
+import ca.bc.gov.open.jrccaccess.autoconfigure.redis.RedisStorageService;
 import ca.bc.gov.open.jrccaccess.libs.DocumentStorageProperties;
 import ca.bc.gov.open.jrccaccess.libs.TransactionInfo;
 import org.junit.Before;

jrcc-access-spring-boot-autoconfigure/src/test/java/ca/bc/gov/open/jrccaccess/autoconfigure/plugins/rabbitmq/RedisStorageServiceTests.java renamed to jrcc-access-spring-boot-autoconfigure/src/test/java/ca/bc/gov/open/jrccaccess/autoconfigure/redis/RedisStorageServiceTests.java

@@ -1,4 +1,4 @@
-package ca.bc.gov.open.jrccaccess.autoconfigure.plugins.rabbitmq;
+package ca.bc.gov.open.jrccaccess.autoconfigure.redis;
 
 import ca.bc.gov.open.jrccaccess.autoconfigure.AccessProperties;
 import ca.bc.gov.open.jrccaccess.autoconfigure.AccessProperties.PluginConfig;

jrcc-access-spring-boot-sample-app/src/main/resources/application.yml

@@ -7,7 +7,7 @@ logging:
         bc: DEBUG
   pattern:
     console: "%d{yyyy-MM-dd HH:mm:ss} [%thread] %-5level %logger{36} %X{transaction.filename} %X{transaction.id} - %msg%n"
-    file: "%d{yyyy-MM-dd HH:mm:ss} [%thread] %-5level %logger{36} %X{transaction.filename} %X{transaction.id} - %msg%n"
+    file: "%d{yyyy-MM-dd HH:mm:ss} [%thread] %-5level %logger{36} %X{transaction.filename} - %msg%n"
 bcgov:
   access:
     input:
@@ -16,4 +16,4 @@ bcgov:
       plugin: console
     output:
       document-type: test-doc
-      plugin: console
+      plugin: console