@Test void testWatchWithQueryCancellation() throws Exception { final AtomicInteger numSubtasks = new AtomicInteger(); final CountDownLatch subtaskCancelled = new CountDownLatch(1); watchConsumer = f -> { numSubtasks.getAndIncrement(); f.exceptionally(cause -> { if (cause instanceof CancellationException) { subtaskCancelled.countDown(); } return null; }); }; final CompletableFuture<Entry<JsonNode>> f = repo.watch(HEAD, Query.ofJsonPath(jsonPaths[0], "$")); assertThatThrownBy(() -> f.get(500, TimeUnit.MILLISECONDS)) .isInstanceOf(TimeoutException.class); assertThat(numSubtasks.get()).isEqualTo(1); assertThat(subtaskCancelled.getCount()).isEqualTo(1L); assertThat(f.cancel(true)).isTrue(); assertThatThrownBy(() -> f.get(3, TimeUnit.SECONDS)) .isInstanceOf(CancellationException.class); assertThat(subtaskCancelled.await(3, TimeUnit.SECONDS)).isTrue(); assertThat(numSubtasks.get()).isEqualTo(1); ensureWatcherCleanUp(); }
@Override public CompletableFuture<Revision> watch(Revision lastKnownRevision, String pathPattern) { requireNonNull(lastKnownRevision, "lastKnownRevision"); requireNonNull(pathPattern, "pathPattern"); final ServiceRequestContext ctx = context(); final Revision normLastKnownRevision = normalizeNow(lastKnownRevision); final CompletableFuture<Revision> future = new CompletableFuture<>(); CompletableFuture.runAsync(() -> { failFastIfTimedOut(this, logger, ctx, "watch", lastKnownRevision, pathPattern); readLock(); try { final Revision latestRevision = blockingFindLatestRevision(normLastKnownRevision, pathPattern); if (latestRevision != null) { future.complete(latestRevision); } else { commitWatchers.add(normLastKnownRevision, pathPattern, future); } } finally { readUnlock(); } }, repositoryWorker).exceptionally(cause -> { future.completeExceptionally(cause); return null; }); return future; }
GitRepository implements Repository { @Override public CompletableFuture<Revision> watch(Revision lastKnownRevision, String pathPattern) { requireNonNull(lastKnownRevision, "lastKnownRevision"); requireNonNull(pathPattern, "pathPattern"); final ServiceRequestContext ctx = context(); final Revision normLastKnownRevision = normalizeNow(lastKnownRevision); final CompletableFuture<Revision> future = new CompletableFuture<>(); CompletableFuture.runAsync(() -> { failFastIfTimedOut(this, logger, ctx, "watch", lastKnownRevision, pathPattern); readLock(); try { final Revision latestRevision = blockingFindLatestRevision(normLastKnownRevision, pathPattern); if (latestRevision != null) { future.complete(latestRevision); } else { commitWatchers.add(normLastKnownRevision, pathPattern, future); } } finally { readUnlock(); } }, repositoryWorker).exceptionally(cause -> { future.completeExceptionally(cause); return null; }); return future; } }
GitRepository implements Repository { @Override public CompletableFuture<Revision> watch(Revision lastKnownRevision, String pathPattern) { requireNonNull(lastKnownRevision, "lastKnownRevision"); requireNonNull(pathPattern, "pathPattern"); final ServiceRequestContext ctx = context(); final Revision normLastKnownRevision = normalizeNow(lastKnownRevision); final CompletableFuture<Revision> future = new CompletableFuture<>(); CompletableFuture.runAsync(() -> { failFastIfTimedOut(this, logger, ctx, "watch", lastKnownRevision, pathPattern); readLock(); try { final Revision latestRevision = blockingFindLatestRevision(normLastKnownRevision, pathPattern); if (latestRevision != null) { future.complete(latestRevision); } else { commitWatchers.add(normLastKnownRevision, pathPattern, future); } } finally { readUnlock(); } }, repositoryWorker).exceptionally(cause -> { future.completeExceptionally(cause); return null; }); return future; } @VisibleForTesting  GitRepository(Project parent, File repoDir, Executor repositoryWorker,
                  long creationTimeMillis, Author author);   GitRepository(Project parent, File repoDir, GitRepositoryFormat format, Executor repositoryWorker,
                  long creationTimeMillis, Author author, @Nullable RepositoryCache cache);   GitRepository(Project parent, File repoDir, Executor repositoryWorker, @Nullable RepositoryCache cache); }
GitRepository implements Repository { @Override public CompletableFuture<Revision> watch(Revision lastKnownRevision, String pathPattern) { requireNonNull(lastKnownRevision, "lastKnownRevision"); requireNonNull(pathPattern, "pathPattern"); final ServiceRequestContext ctx = context(); final Revision normLastKnownRevision = normalizeNow(lastKnownRevision); final CompletableFuture<Revision> future = new CompletableFuture<>(); CompletableFuture.runAsync(() -> { failFastIfTimedOut(this, logger, ctx, "watch", lastKnownRevision, pathPattern); readLock(); try { final Revision latestRevision = blockingFindLatestRevision(normLastKnownRevision, pathPattern); if (latestRevision != null) { future.complete(latestRevision); } else { commitWatchers.add(normLastKnownRevision, pathPattern, future); } } finally { readUnlock(); } }, repositoryWorker).exceptionally(cause -> { future.completeExceptionally(cause); return null; }); return future; } @VisibleForTesting  GitRepository(Project parent, File repoDir, Executor repositoryWorker,
                  long creationTimeMillis, Author author);   GitRepository(Project parent, File repoDir, GitRepositoryFormat format, Executor repositoryWorker,
                  long creationTimeMillis, Author author, @Nullable RepositoryCache cache);   GitRepository(Project parent, File repoDir, Executor repositoryWorker, @Nullable RepositoryCache cache); @Override Project parent(); @Override String name(); GitRepositoryFormat format(); boolean needsMigration(GitRepositoryFormat preferredFormat); @Override Revision normalizeNow(Revision revision); @Override RevisionRange normalizeNow(Revision from, Revision to); @Override CompletableFuture<Map<String, Entry<?>>> find(
            Revision revision, String pathPattern, Map<FindOption<?>, ?> options); @Override CompletableFuture<List<Commit>> history(
            Revision from, Revision to, String pathPattern, int maxCommits); @Override CompletableFuture<Map<String, Change<?>>> diff(Revision from, Revision to, String pathPattern); @Override CompletableFuture<Map<String, Change<?>>> previewDiff(Revision baseRevision,
                                                                 Iterable<Change<?>> changes); @Override CompletableFuture<Revision> commit(
            Revision baseRevision, long commitTimeMillis, Author author, String summary,
            String detail, Markup markup, Iterable<Change<?>> changes); @Override CompletableFuture<Revision> findLatestRevision(Revision lastKnownRevision, String pathPattern); @Override CompletableFuture<Revision> watch(Revision lastKnownRevision, String pathPattern); void cloneTo(File newRepoDir); void cloneTo(File newRepoDir, BiConsumer<Integer, Integer> progressListener); void cloneTo(File newRepoDir, GitRepositoryFormat format); void cloneTo(File newRepoDir, GitRepositoryFormat format,
                        BiConsumer<Integer, Integer> progressListener); @Override String toString(); }
GitRepository implements Repository { @Override public CompletableFuture<Revision> watch(Revision lastKnownRevision, String pathPattern) { requireNonNull(lastKnownRevision, "lastKnownRevision"); requireNonNull(pathPattern, "pathPattern"); final ServiceRequestContext ctx = context(); final Revision normLastKnownRevision = normalizeNow(lastKnownRevision); final CompletableFuture<Revision> future = new CompletableFuture<>(); CompletableFuture.runAsync(() -> { failFastIfTimedOut(this, logger, ctx, "watch", lastKnownRevision, pathPattern); readLock(); try { final Revision latestRevision = blockingFindLatestRevision(normLastKnownRevision, pathPattern); if (latestRevision != null) { future.complete(latestRevision); } else { commitWatchers.add(normLastKnownRevision, pathPattern, future); } } finally { readUnlock(); } }, repositoryWorker).exceptionally(cause -> { future.completeExceptionally(cause); return null; }); return future; } @VisibleForTesting  GitRepository(Project parent, File repoDir, Executor repositoryWorker,
                  long creationTimeMillis, Author author);   GitRepository(Project parent, File repoDir, GitRepositoryFormat format, Executor repositoryWorker,
                  long creationTimeMillis, Author author, @Nullable RepositoryCache cache);   GitRepository(Project parent, File repoDir, Executor repositoryWorker, @Nullable RepositoryCache cache); @Override Project parent(); @Override String name(); GitRepositoryFormat format(); boolean needsMigration(GitRepositoryFormat preferredFormat); @Override Revision normalizeNow(Revision revision); @Override RevisionRange normalizeNow(Revision from, Revision to); @Override CompletableFuture<Map<String, Entry<?>>> find(
            Revision revision, String pathPattern, Map<FindOption<?>, ?> options); @Override CompletableFuture<List<Commit>> history(
            Revision from, Revision to, String pathPattern, int maxCommits); @Override CompletableFuture<Map<String, Change<?>>> diff(Revision from, Revision to, String pathPattern); @Override CompletableFuture<Map<String, Change<?>>> previewDiff(Revision baseRevision,
                                                                 Iterable<Change<?>> changes); @Override CompletableFuture<Revision> commit(
            Revision baseRevision, long commitTimeMillis, Author author, String summary,
            String detail, Markup markup, Iterable<Change<?>> changes); @Override CompletableFuture<Revision> findLatestRevision(Revision lastKnownRevision, String pathPattern); @Override CompletableFuture<Revision> watch(Revision lastKnownRevision, String pathPattern); void cloneTo(File newRepoDir); void cloneTo(File newRepoDir, BiConsumer<Integer, Integer> progressListener); void cloneTo(File newRepoDir, GitRepositoryFormat format); void cloneTo(File newRepoDir, GitRepositoryFormat format,
                        BiConsumer<Integer, Integer> progressListener); @Override String toString();  }