Google Git
Sign in
gerrit/plugins/replication/9f6c8c838c0386ac407cf016ab3f24e4cf3fa3d6^!/.
commit
9f6c8c838c0386ac407cf016ab3f24e4cf3fa3d6
[log]
author
Martin Fick <mfick@codeaurora.org>
Fri Oct 09 11:50:31 2020 -0600
committer
Martin Fick <mfick@codeaurora.org>
Fri Oct 09 12:41:38 2020 -0600
tree
89fe9d41f38ed3f2434e22032f54f5ecd5daa202
parent
63fb2b4ba85380d798acbdc076e8673353507569 [diff]
ReplicationTaskStorage: Use Stream API instead of List

Change-Id: I76c23bf472ebc0cad804207699fdca0e9cf4de14

diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java
index cd4a3fb..90d159d 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java

@@ -202,13 +202,17 @@
     replaying = true;
     try {
       replaying = true;
-      for (ReplicationTasksStorage.ReplicateRefUpdate t : replicationTasksStorage.listWaiting()) {
-        try {
-          fire(new URIish(t.uri()), Project.nameKey(t.project()), t.ref());
-        } catch (URISyntaxException e) {
-          repLog.atSevere().withCause(e).log("Encountered malformed URI for persisted event %s", t);
-        }
-      }
+      replicationTasksStorage
+          .streamWaiting()
+          .forEach(
+              t -> {
+                try {
+                  fire(new URIish(t.uri()), Project.nameKey(t.project()), t.ref());
+                } catch (URISyntaxException e) {
+                  repLog.atSevere().withCause(e).log(
+                      "Encountered malformed URI for persisted event %s", t);
+                }
+              });
     } catch (Throwable e) {
       repLog.atSevere().withCause(e).log("Unexpected error while firing pending events");
     } finally {

diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorage.java b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorage.java
index 68661cf..c67bb49 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorage.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorage.java

@@ -32,9 +32,7 @@
 import java.nio.file.NotDirectoryException;
 import java.nio.file.Path;
 import java.nio.file.StandardCopyOption;
-import java.util.List;
 import java.util.Optional;
-import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.eclipse.jgit.lib.ObjectId;
 import org.eclipse.jgit.transport.URIish;
@@ -147,9 +145,7 @@
   }
 
   public synchronized void resetAll() {
-    for (ReplicateRefUpdate r : list(createDir(runningUpdates))) {
-      new Task(r).reset();
-    }
+    streamRunning().forEach(r -> new Task(r).reset());
   }
 
   public boolean isWaiting(UriUpdates uriUpdates) {
@@ -164,16 +160,12 @@
     }
   }
 
-  public List<ReplicateRefUpdate> listWaiting() {
-    return list(createDir(waitingUpdates));
+  public Stream<ReplicateRefUpdate> streamWaiting() {
+    return streamRecursive(createDir(waitingUpdates));
   }
 
-  public List<ReplicateRefUpdate> listRunning() {
-    return list(createDir(runningUpdates));
-  }
-
-  private List<ReplicateRefUpdate> list(Path taskDir) {
-    return streamRecursive(taskDir).collect(Collectors.toList());
+  public Stream<ReplicateRefUpdate> streamRunning() {
+    return streamRecursive(createDir(runningUpdates));
   }
 
   private Stream<ReplicateRefUpdate> streamRecursive(Path dir) {

diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationFanoutIT.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationFanoutIT.java
index 7e336d1..7f8282a 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationFanoutIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationFanoutIT.java

@@ -247,7 +247,7 @@
   private List<ReplicateRefUpdate> listIncompleteTasks(String refRegex) {
     Pattern refmaskPattern = Pattern.compile(refRegex);
     synchronized (tasksStorage) {
-      return Stream.concat(tasksStorage.listWaiting().stream(), tasksStorage.listRunning().stream())
+      return Stream.concat(tasksStorage.streamWaiting(), tasksStorage.streamRunning())
           .filter(task -> refmaskPattern.matcher(task.ref()).matches())
           .collect(toList());
     }

diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationIT.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationIT.java
index f0d2347..e0e3380 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationIT.java

@@ -49,6 +49,7 @@
 import java.util.Optional;
 import java.util.function.Supplier;
 import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.eclipse.jgit.lib.Constants;
 import org.eclipse.jgit.lib.Ref;
@@ -453,11 +454,11 @@
     config.setInt("remote", "task_cleanup_project", "replicationRetry", 0);
     config.save();
     reloadConfig();
-    assertThat(tasksStorage.listRunning()).hasSize(0);
+    assertThat(listRunning()).hasSize(0);
     Project.NameKey sourceProject = createTestProject("task_cleanup_project");
 
     waitUntil(() -> nonEmptyProjectExists(Project.nameKey(sourceProject + "replica.git")));
-    waitUntil(() -> tasksStorage.listRunning().size() == 0);
+    waitUntil(() -> listRunning().size() == 0);
   }
 
   @Test
@@ -466,7 +467,7 @@
     config.setInt("remote", "task_cleanup_locks_project", "replicationRetry", 0);
     config.save();
     reloadConfig();
-    assertThat(tasksStorage.listRunning()).hasSize(0);
+    assertThat(listRunning()).hasSize(0);
     Project.NameKey sourceProject = createTestProject("task_cleanup_locks_project");
 
     waitUntil(() -> nonEmptyProjectExists(Project.nameKey(sourceProject + "replica.git")));
@@ -492,7 +493,7 @@
             .findFirst()
             .get();
 
-    waitUntil(() -> tasksStorage.listRunning().size() == 0);
+    waitUntil(() -> listRunning().size() == 0);
 
     createTestProject(projectName);
 
@@ -525,7 +526,7 @@
 
     String changeRef = createChange().getPatchSet().refName();
 
-    changeReplicationTasksForRemote(tasksStorage.listWaiting().stream(), changeRef, remote1)
+    changeReplicationTasksForRemote(tasksStorage.streamWaiting(), changeRef, remote1)
         .forEach(
             (update) -> {
               try {
@@ -675,26 +676,33 @@
     return projectOperations.newProject().name(name).create();
   }
 
-  private List<ReplicateRefUpdate> listIncompleteTasks(String refRegex) {
-    Pattern refmaskPattern = Pattern.compile(refRegex);
-    return streamIncompleteTasks()
-        .filter(task -> refmaskPattern.matcher(task.ref()).matches())
-        .collect(toList());
-  }
-
-  private List<ReplicateRefUpdate> listIncompleteTasks() {
-    return streamIncompleteTasks().collect(toList());
+  public List<ReplicateRefUpdate> listRunning() {
+    return tasksStorage.streamRunning().collect(Collectors.toList());
   }
 
   @SuppressWarnings(
       "SynchronizeOnNonFinalField") // tasksStorage is non-final but only set in setUpTestPlugin()
-  private Stream<ReplicateRefUpdate> streamIncompleteTasks() {
+  private List<ReplicateRefUpdate> listIncompleteTasks(String refRegex) {
+    Pattern refmaskPattern = Pattern.compile(refRegex);
     synchronized (tasksStorage) {
-      return Stream.concat(
-          tasksStorage.listWaiting().stream(), tasksStorage.listRunning().stream());
+      return streamIncompleteTasks()
+          .filter(task -> refmaskPattern.matcher(task.ref()).matches())
+          .collect(toList());
     }
   }
 
+  @SuppressWarnings(
+      "SynchronizeOnNonFinalField") // tasksStorage is non-final but only set in setUpTestPlugin()
+  private List<ReplicateRefUpdate> listIncompleteTasks() {
+    synchronized (tasksStorage) {
+      return streamIncompleteTasks().collect(toList());
+    }
+  }
+
+  private Stream<ReplicateRefUpdate> streamIncompleteTasks() {
+    return Stream.concat(tasksStorage.streamWaiting(), tasksStorage.streamRunning());
+  }
+
   public void cleanupReplicationTasks() throws IOException {
     cleanupReplicationTasks(storagePath);
   }

diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorageTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorageTest.java
index 948044b..4b09d52 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorageTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorageTest.java

@@ -20,10 +20,13 @@
 
 import com.google.common.jimfs.Configuration;
 import com.google.common.jimfs.Jimfs;
+import com.google.common.truth.IterableSubject;
 import com.googlesource.gerrit.plugins.replication.ReplicationTasksStorage.ReplicateRefUpdate;
 import java.net.URISyntaxException;
 import java.nio.file.FileSystem;
 import java.nio.file.Path;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 import org.eclipse.jgit.transport.URIish;
 import org.junit.After;
 import org.junit.Before;
@@ -57,22 +60,22 @@
 
   @Test
   public void canListEmptyStorage() throws Exception {
-    assertThat(storage.listWaiting()).isEmpty();
-    assertThat(storage.listRunning()).isEmpty();
+    assertThatStream(storage.streamWaiting()).isEmpty();
+    assertThatStream(storage.streamRunning()).isEmpty();
   }
 
   @Test
   public void canListWaitingUpdate() throws Exception {
     storage.create(REF_UPDATE);
-    assertThat(storage.listWaiting()).containsExactly(REF_UPDATE);
+    assertThatStream(storage.streamWaiting()).containsExactly(REF_UPDATE);
   }
 
   @Test
   public void canStartWaitingUpdate() throws Exception {
     storage.create(REF_UPDATE);
     storage.start(uriUpdates);
-    assertThat(storage.listWaiting()).isEmpty();
-    assertThat(storage.listRunning()).containsExactly(REF_UPDATE);
+    assertThatStream(storage.streamWaiting()).isEmpty();
+    assertThatStream(storage.streamRunning()).containsExactly(REF_UPDATE);
   }
 
   @Test
@@ -87,22 +90,22 @@
   public void instancesOfTheSameStorageHaveTheSameElements() throws Exception {
     ReplicationTasksStorage persistedView = new ReplicationTasksStorage(storageSite);
 
-    assertThat(storage.listWaiting()).isEmpty();
-    assertThat(persistedView.listWaiting()).isEmpty();
+    assertThatStream(storage.streamWaiting()).isEmpty();
+    assertThatStream(persistedView.streamWaiting()).isEmpty();
 
     storage.create(REF_UPDATE);
-    assertThat(storage.listWaiting()).containsExactly(REF_UPDATE);
-    assertThat(persistedView.listWaiting()).containsExactly(REF_UPDATE);
+    assertThatStream(storage.streamWaiting()).containsExactly(REF_UPDATE);
+    assertThatStream(persistedView.streamWaiting()).containsExactly(REF_UPDATE);
 
     storage.start(uriUpdates);
-    assertThat(storage.listWaiting()).isEmpty();
-    assertThat(persistedView.listWaiting()).isEmpty();
-    assertThat(storage.listRunning()).containsExactly(REF_UPDATE);
-    assertThat(persistedView.listRunning()).containsExactly(REF_UPDATE);
+    assertThatStream(storage.streamWaiting()).isEmpty();
+    assertThatStream(persistedView.streamWaiting()).isEmpty();
+    assertThatStream(storage.streamRunning()).containsExactly(REF_UPDATE);
+    assertThatStream(persistedView.streamRunning()).containsExactly(REF_UPDATE);
 
     storage.finish(uriUpdates);
-    assertThat(storage.listRunning()).isEmpty();
-    assertThat(persistedView.listRunning()).isEmpty();
+    assertThatStream(storage.streamRunning()).isEmpty();
+    assertThatStream(persistedView.streamRunning()).isEmpty();
   }
 
   @Test
@@ -110,7 +113,7 @@
     String key = storage.create(REF_UPDATE);
     String secondKey = storage.create(REF_UPDATE);
     assertEquals(key, secondKey);
-    assertThat(storage.listWaiting()).containsExactly(REF_UPDATE);
+    assertThatStream(storage.streamWaiting()).containsExactly(REF_UPDATE);
   }
 
   @Test
@@ -124,7 +127,7 @@
 
     String keyA = storage.create(REF_UPDATE);
     String keyB = storage.create(updateB);
-    assertThat(storage.listWaiting()).hasSize(2);
+    assertThatStream(storage.streamWaiting()).hasSize(2);
     assertNotEquals(keyA, keyB);
   }
 
@@ -141,12 +144,12 @@
     storage.create(updateB);
 
     storage.start(uriUpdates);
-    assertThat(storage.listWaiting()).containsExactly(updateB);
-    assertThat(storage.listRunning()).containsExactly(REF_UPDATE);
+    assertThatStream(storage.streamWaiting()).containsExactly(updateB);
+    assertThatStream(storage.streamRunning()).containsExactly(REF_UPDATE);
 
     storage.start(uriUpdatesB);
-    assertThat(storage.listWaiting()).isEmpty();
-    assertThat(storage.listRunning()).containsExactly(REF_UPDATE, updateB);
+    assertThatStream(storage.streamWaiting()).isEmpty();
+    assertThatStream(storage.streamRunning()).containsExactly(REF_UPDATE, updateB);
   }
 
   @Test
@@ -164,10 +167,10 @@
     storage.start(uriUpdatesB);
 
     storage.finish(uriUpdates);
-    assertThat(storage.listRunning()).containsExactly(updateB);
+    assertThatStream(storage.streamRunning()).containsExactly(updateB);
 
     storage.finish(uriUpdatesB);
-    assertThat(storage.listRunning()).isEmpty();
+    assertThatStream(storage.streamRunning()).isEmpty();
   }
 
   @Test
@@ -183,7 +186,7 @@
     storage.create(updateB);
     storage.create(REF_UPDATE);
     storage.create(updateB);
-    assertThat(storage.listWaiting()).hasSize(2);
+    assertThatStream(storage.streamWaiting()).hasSize(2);
   }
 
   @Test
@@ -193,7 +196,7 @@
 
     String keyA = storage.create(refA);
     String keyB = storage.create(refB);
-    assertThat(storage.listWaiting()).hasSize(2);
+    assertThatStream(storage.streamWaiting()).hasSize(2);
     assertNotEquals(keyA, keyB);
   }
 
@@ -209,10 +212,10 @@
     storage.start(uriUpdatesB);
 
     storage.finish(uriUpdatesA);
-    assertThat(storage.listRunning()).containsExactly(refUpdateB);
+    assertThatStream(storage.streamRunning()).containsExactly(refUpdateB);
 
     storage.finish(uriUpdatesB);
-    assertThat(storage.listRunning()).isEmpty();
+    assertThatStream(storage.streamRunning()).isEmpty();
   }
 
   @Test
@@ -221,8 +224,8 @@
     storage.start(uriUpdates);
 
     storage.reset(uriUpdates);
-    assertThat(storage.listWaiting()).containsExactly(REF_UPDATE);
-    assertThat(storage.listRunning()).isEmpty();
+    assertThatStream(storage.streamWaiting()).containsExactly(REF_UPDATE);
+    assertThatStream(storage.streamRunning()).isEmpty();
   }
 
   @Test
@@ -232,8 +235,8 @@
     storage.reset(uriUpdates);
 
     storage.start(uriUpdates);
-    assertThat(storage.listRunning()).containsExactly(REF_UPDATE);
-    assertThat(storage.listWaiting()).isEmpty();
+    assertThatStream(storage.streamRunning()).containsExactly(REF_UPDATE);
+    assertThatStream(storage.streamWaiting()).isEmpty();
 
     storage.finish(uriUpdates);
     assertNoIncompleteTasks(storage);
@@ -251,8 +254,8 @@
     storage.start(uriUpdates);
 
     storage.resetAll();
-    assertThat(storage.listWaiting()).containsExactly(REF_UPDATE);
-    assertThat(storage.listRunning()).isEmpty();
+    assertThatStream(storage.streamWaiting()).containsExactly(REF_UPDATE);
+    assertThatStream(storage.streamRunning()).isEmpty();
   }
 
   @Test
@@ -262,8 +265,8 @@
     storage.resetAll();
 
     storage.start(uriUpdates);
-    assertThat(storage.listRunning()).containsExactly(REF_UPDATE);
-    assertThat(storage.listWaiting()).isEmpty();
+    assertThatStream(storage.streamRunning()).containsExactly(REF_UPDATE);
+    assertThatStream(storage.streamWaiting()).isEmpty();
 
     storage.finish(uriUpdates);
     assertNoIncompleteTasks(storage);
@@ -284,7 +287,7 @@
     storage.start(uriUpdatesB);
 
     storage.resetAll();
-    assertThat(storage.listWaiting()).containsExactly(REF_UPDATE, updateB);
+    assertThatStream(storage.streamWaiting()).containsExactly(REF_UPDATE, updateB);
   }
 
   @Test
@@ -303,12 +306,12 @@
     storage.resetAll();
 
     storage.start(uriUpdates);
-    assertThat(storage.listRunning()).containsExactly(REF_UPDATE);
-    assertThat(storage.listWaiting()).containsExactly(updateB);
+    assertThatStream(storage.streamRunning()).containsExactly(REF_UPDATE);
+    assertThatStream(storage.streamWaiting()).containsExactly(updateB);
 
     storage.start(uriUpdatesB);
-    assertThat(storage.listRunning()).containsExactly(REF_UPDATE, updateB);
-    assertThat(storage.listWaiting()).isEmpty();
+    assertThatStream(storage.streamRunning()).containsExactly(REF_UPDATE, updateB);
+    assertThatStream(storage.streamWaiting()).isEmpty();
 
     storage.finish(uriUpdates);
     storage.finish(uriUpdatesB);
@@ -347,12 +350,16 @@
 
     storage.finish(uriUpdates);
     storage.finish(uriUpdatesB);
-    assertThat(storage.listRunning()).isEmpty();
+    assertThatStream(storage.streamRunning()).isEmpty();
   }
 
   private void assertNoIncompleteTasks(ReplicationTasksStorage storage) {
-    assertThat(storage.listWaiting()).isEmpty();
-    assertThat(storage.listRunning()).isEmpty();
+    assertThatStream(storage.streamWaiting()).isEmpty();
+    assertThatStream(storage.streamRunning()).isEmpty();
+  }
+
+  private IterableSubject assertThatStream(Stream<?> stream) {
+    return assertThat(stream.collect(Collectors.toList()));
   }
 
   public static URIish getUrish(String uri) {