FtpPersistentAcceptOnceFileListFilterTests.java

@Test
	@RedisAvailable
	public void testFtpWithRedisMetadataStore() throws Exception {
		RedisTemplate<String, ?> template = new RedisTemplate<String, Object>();
		template.setConnectionFactory(this.getConnectionFactoryForTest());
		template.setKeySerializer(new StringRedisSerializer());
		template.afterPropertiesSet();
		template.delete("persistentAcceptOnceFileListFilterRedisTests");

		final FTPFile ftpFile = new FTPFile();
		ftpFile.setName("foo");
		ftpFile.setTimestamp(Calendar.getInstance());

		ExecutorService executorService = Executors.newCachedThreadPool();

		List<Future<Integer>> filters = new ArrayList<Future<Integer>>();

		for (int i = 0; i < 10; i++) {
			filters.add(executorService.submit(new Callable<Integer>() {

				@Override
				public Integer call() throws Exception {
					JedisConnectionFactory connectionFactory = new JedisConnectionFactory();
					connectionFactory.setPort(RedisAvailableRule.REDIS_PORT);
					connectionFactory.afterPropertiesSet();

					RedisMetadataStore metadataStore = new RedisMetadataStore(connectionFactory,
							"persistentAcceptOnceFileListFilterRedisTests");

					FileListFilter<FTPFile> filter1 =
							new FtpPersistentAcceptOnceFileListFilter(metadataStore, "foo:");

					int size = filter1.filterFiles(new FTPFile[]{ftpFile}).size();
					System.out.println(Thread.currentThread().getId() + ": " + size);
					return size;
				}

			}));
		}

		boolean accepted = false;

		for (Future<Integer> filter : filters) {
			Integer theResult = filter.get(10, TimeUnit.SECONDS);
			if (accepted) {
				if (theResult == 1) {
					fail("Only one filter should accept the file");
				}
			}
			else {
				accepted = theResult == 1;
			}

		}

		template.delete("persistentAcceptOnceFileListFilterRedisTests");
	}