@Test public void multiple() throws Exception { List<File> sources = new ArrayList<>(); List<Path> paths = new ArrayList<>(); for (int i = 0; i < 100; i++) { File source = put(newFile(), String.format("hello%02d", i)); sources.add(source); paths.add(path(source)); } Map<File, File> results; ExecutorService executor = Executors.newFixedThreadPool( Math.min(4, Runtime.getRuntime().availableProcessors())); try { BatchFileCacheRepository batch = new ConcurrentBatchFileCacheRepository( new MockFileCacheRepository(folder.newFolder()), executor); results = files(batch.resolve(paths)); } finally { executor.shutdown(); } assertThat(results.size(), is(sources.size())); for (File source : sources) { File target = results.get(source); assertThat(target, is(notNullValue())); assertThat(get(target), is(get(source))); assertThat(target, is(not(source))); } }
@Override public Map<Path, Path> resolve(List<? extends Path> files) throws IOException, InterruptedException { Map<Path, Future<Path>> futures = new LinkedHashMap<>(); Map<Path, Path> results = new HashMap<>(); for (Path file : files) { if (futures.containsKey(file)) { continue; } Future<Path> future = executor.submit(new Task(repository, file)); futures.put(file, future); results.put(file, null); } try { while (futures.isEmpty() == false) { for (Path file : files) { Future<Path> future = futures.get(file); if (future == null) { continue; } try { Path result = future.get(CHECK_TIMEOUT, TimeUnit.MILLISECONDS); futures.remove(file); if (result != null) { results.put(file, result); } } catch (CancellationException e) { futures.remove(file); if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Processing cache is cancelled: {0}", file), e); } } catch (ExecutionException e) { futures.remove(file); Path result = handleException(file, e); if (result != null) { results.put(file, result); } } catch (TimeoutException e) { if (LOG.isTraceEnabled()) { LOG.trace(MessageFormat.format( "Trying to wait for next task complete: {0}", file), e); } } } } } finally { for (Map.Entry<Path, Future<Path>> entry : futures.entrySet()) { if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Request cancel for processing cache: {0}", entry.getKey())); } entry.getValue().cancel(true); } } return results; }
ConcurrentBatchFileCacheRepository implements BatchFileCacheRepository { @Override public Map<Path, Path> resolve(List<? extends Path> files) throws IOException, InterruptedException { Map<Path, Future<Path>> futures = new LinkedHashMap<>(); Map<Path, Path> results = new HashMap<>(); for (Path file : files) { if (futures.containsKey(file)) { continue; } Future<Path> future = executor.submit(new Task(repository, file)); futures.put(file, future); results.put(file, null); } try { while (futures.isEmpty() == false) { for (Path file : files) { Future<Path> future = futures.get(file); if (future == null) { continue; } try { Path result = future.get(CHECK_TIMEOUT, TimeUnit.MILLISECONDS); futures.remove(file); if (result != null) { results.put(file, result); } } catch (CancellationException e) { futures.remove(file); if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Processing cache is cancelled: {0}", file), e); } } catch (ExecutionException e) { futures.remove(file); Path result = handleException(file, e); if (result != null) { results.put(file, result); } } catch (TimeoutException e) { if (LOG.isTraceEnabled()) { LOG.trace(MessageFormat.format( "Trying to wait for next task complete: {0}", file), e); } } } } } finally { for (Map.Entry<Path, Future<Path>> entry : futures.entrySet()) { if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Request cancel for processing cache: {0}", entry.getKey())); } entry.getValue().cancel(true); } } return results; } }
ConcurrentBatchFileCacheRepository implements BatchFileCacheRepository { @Override public Map<Path, Path> resolve(List<? extends Path> files) throws IOException, InterruptedException { Map<Path, Future<Path>> futures = new LinkedHashMap<>(); Map<Path, Path> results = new HashMap<>(); for (Path file : files) { if (futures.containsKey(file)) { continue; } Future<Path> future = executor.submit(new Task(repository, file)); futures.put(file, future); results.put(file, null); } try { while (futures.isEmpty() == false) { for (Path file : files) { Future<Path> future = futures.get(file); if (future == null) { continue; } try { Path result = future.get(CHECK_TIMEOUT, TimeUnit.MILLISECONDS); futures.remove(file); if (result != null) { results.put(file, result); } } catch (CancellationException e) { futures.remove(file); if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Processing cache is cancelled: {0}", file), e); } } catch (ExecutionException e) { futures.remove(file); Path result = handleException(file, e); if (result != null) { results.put(file, result); } } catch (TimeoutException e) { if (LOG.isTraceEnabled()) { LOG.trace(MessageFormat.format( "Trying to wait for next task complete: {0}", file), e); } } } } } finally { for (Map.Entry<Path, Future<Path>> entry : futures.entrySet()) { if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Request cancel for processing cache: {0}", entry.getKey())); } entry.getValue().cancel(true); } } return results; } ConcurrentBatchFileCacheRepository(FileCacheRepository repository, ExecutorService executor);  ConcurrentBatchFileCacheRepository(
            FileCacheRepository repository,
            ExecutorService executor,
            ExceptionHandler exceptionHandler); }
ConcurrentBatchFileCacheRepository implements BatchFileCacheRepository { @Override public Map<Path, Path> resolve(List<? extends Path> files) throws IOException, InterruptedException { Map<Path, Future<Path>> futures = new LinkedHashMap<>(); Map<Path, Path> results = new HashMap<>(); for (Path file : files) { if (futures.containsKey(file)) { continue; } Future<Path> future = executor.submit(new Task(repository, file)); futures.put(file, future); results.put(file, null); } try { while (futures.isEmpty() == false) { for (Path file : files) { Future<Path> future = futures.get(file); if (future == null) { continue; } try { Path result = future.get(CHECK_TIMEOUT, TimeUnit.MILLISECONDS); futures.remove(file); if (result != null) { results.put(file, result); } } catch (CancellationException e) { futures.remove(file); if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Processing cache is cancelled: {0}", file), e); } } catch (ExecutionException e) { futures.remove(file); Path result = handleException(file, e); if (result != null) { results.put(file, result); } } catch (TimeoutException e) { if (LOG.isTraceEnabled()) { LOG.trace(MessageFormat.format( "Trying to wait for next task complete: {0}", file), e); } } } } } finally { for (Map.Entry<Path, Future<Path>> entry : futures.entrySet()) { if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Request cancel for processing cache: {0}", entry.getKey())); } entry.getValue().cancel(true); } } return results; } ConcurrentBatchFileCacheRepository(FileCacheRepository repository, ExecutorService executor);  ConcurrentBatchFileCacheRepository(
            FileCacheRepository repository,
            ExecutorService executor,
            ExceptionHandler exceptionHandler); @Override Map<Path, Path> resolve(List<? extends Path> files); }
ConcurrentBatchFileCacheRepository implements BatchFileCacheRepository { @Override public Map<Path, Path> resolve(List<? extends Path> files) throws IOException, InterruptedException { Map<Path, Future<Path>> futures = new LinkedHashMap<>(); Map<Path, Path> results = new HashMap<>(); for (Path file : files) { if (futures.containsKey(file)) { continue; } Future<Path> future = executor.submit(new Task(repository, file)); futures.put(file, future); results.put(file, null); } try { while (futures.isEmpty() == false) { for (Path file : files) { Future<Path> future = futures.get(file); if (future == null) { continue; } try { Path result = future.get(CHECK_TIMEOUT, TimeUnit.MILLISECONDS); futures.remove(file); if (result != null) { results.put(file, result); } } catch (CancellationException e) { futures.remove(file); if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Processing cache is cancelled: {0}", file), e); } } catch (ExecutionException e) { futures.remove(file); Path result = handleException(file, e); if (result != null) { results.put(file, result); } } catch (TimeoutException e) { if (LOG.isTraceEnabled()) { LOG.trace(MessageFormat.format( "Trying to wait for next task complete: {0}", file), e); } } } } } finally { for (Map.Entry<Path, Future<Path>> entry : futures.entrySet()) { if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "Request cancel for processing cache: {0}", entry.getKey())); } entry.getValue().cancel(true); } } return results; } ConcurrentBatchFileCacheRepository(FileCacheRepository repository, ExecutorService executor);  ConcurrentBatchFileCacheRepository(
            FileCacheRepository repository,
            ExecutorService executor,
            ExceptionHandler exceptionHandler); @Override Map<Path, Path> resolve(List<? extends Path> files);  }