Revert "refactor(core): Remove RxJava from ExecutionRepository interf…

…ace" (spinnaker#2228)

orca-core-tck/src/main/groovy/com/netflix/spinnaker/orca/pipeline/persistence/ExecutionRepositoryTck.groovy

@@ -22,10 +22,10 @@ import com.netflix.spinnaker.orca.pipeline.model.Execution
 import com.netflix.spinnaker.orca.pipeline.model.JenkinsTrigger
 import com.netflix.spinnaker.orca.pipeline.model.PipelineTrigger
 import com.netflix.spinnaker.orca.pipeline.persistence.ExecutionRepository.ExecutionCriteria
+import rx.schedulers.Schedulers
 import spock.lang.Specification
 import spock.lang.Subject
 import spock.lang.Unroll
-
 import static com.netflix.spinnaker.orca.ExecutionStatus.*
 import static com.netflix.spinnaker.orca.pipeline.model.Execution.ExecutionType
 import static com.netflix.spinnaker.orca.pipeline.model.Execution.ExecutionType.PIPELINE
@@ -66,23 +66,23 @@ abstract class ExecutionRepositoryTck<T extends ExecutionRepository> extends Spe
     repository.store(succeededExecution)
     def pipelines = repository.retrievePipelinesForPipelineConfigId(
       "pipeline-1", new ExecutionCriteria(limit: 5, statuses: ["RUNNING", "SUCCEEDED", "TERMINAL"])
-    ).toList()
+    ).subscribeOn(Schedulers.io()).toList().toBlocking().single()
 
     then:
     pipelines*.id.sort() == [runningExecution.id, succeededExecution.id].sort()
 
     when:
     pipelines = repository.retrievePipelinesForPipelineConfigId(
       "pipeline-1", new ExecutionCriteria(limit: 5, statuses: ["RUNNING"])
-    ).toList()
+    ).subscribeOn(Schedulers.io()).toList().toBlocking().single()
 
     then:
     pipelines*.id.sort() == [runningExecution.id].sort()
 
     when:
     pipelines = repository.retrievePipelinesForPipelineConfigId(
       "pipeline-1", new ExecutionCriteria(limit: 5, statuses: ["TERMINAL"])
-    ).toList()
+    ).subscribeOn(Schedulers.io()).toList().toBlocking().single()
 
     then:
     pipelines.isEmpty()
@@ -106,23 +106,23 @@ abstract class ExecutionRepositoryTck<T extends ExecutionRepository> extends Spe
     repository.store(succeededExecution)
     def orchestrations = repository.retrieveOrchestrationsForApplication(
       runningExecution.application, new ExecutionCriteria(limit: 5, statuses: ["RUNNING", "SUCCEEDED", "TERMINAL"])
-    ).toList()
+    ).subscribeOn(Schedulers.io()).toList().toBlocking().single()
 
     then:
     orchestrations*.id.sort() == [runningExecution.id, succeededExecution.id].sort()
 
     when:
     orchestrations = repository.retrieveOrchestrationsForApplication(
       runningExecution.application, new ExecutionCriteria(limit: 5, statuses: ["RUNNING"])
-    ).toList()
+    ).subscribeOn(Schedulers.io()).toList().toBlocking().single()
 
     then:
     orchestrations*.id.sort() == [runningExecution.id].sort()
 
     when:
     orchestrations = repository.retrieveOrchestrationsForApplication(
       runningExecution.application, new ExecutionCriteria(limit: 5, statuses: ["TERMINAL"])
-    ).toList()
+    ).subscribeOn(Schedulers.io()).toList().toBlocking().single()
 
     then:
     orchestrations.isEmpty()
@@ -154,7 +154,7 @@ abstract class ExecutionRepositoryTck<T extends ExecutionRepository> extends Spe
     repository.store(pipeline)
 
     expect:
-    repository.retrieve(PIPELINE).first().id == pipeline.id
+    repository.retrieve(PIPELINE).toBlocking().first().id == pipeline.id
 
     with(repository.retrieve(pipeline.type, pipeline.id)) {
       id == pipeline.id
@@ -223,7 +223,7 @@ abstract class ExecutionRepositoryTck<T extends ExecutionRepository> extends Spe
     thrown ExecutionNotFoundException
 
     and:
-    repository.retrieve(PIPELINE).toList() == []
+    repository.retrieve(PIPELINE).toList().toBlocking().first() == []
   }
 
   def "updateStatus sets startTime to current time if new status is RUNNING"() {

orca-core/src/main/java/com/netflix/spinnaker/orca/pipeline/persistence/ExecutionRepository.java

@@ -20,11 +20,11 @@
 import com.netflix.spinnaker.orca.pipeline.model.Execution;
 import com.netflix.spinnaker.orca.pipeline.model.Execution.ExecutionType;
 import com.netflix.spinnaker.orca.pipeline.model.Stage;
+import rx.Observable;
 
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 import java.util.*;
-import java.util.stream.Stream;
 
 import static java.util.stream.Collectors.toList;
 
@@ -57,14 +57,14 @@ public interface ExecutionRepository {
 
   void delete(@Nonnull ExecutionType type, @Nonnull String id);
 
-  @Nonnull Iterable<Execution> retrieve(ExecutionType type);
+  @Nonnull Observable<Execution> retrieve(ExecutionType type);
 
-  @Nonnull Iterable<Execution> retrievePipelinesForApplication(@Nonnull String application);
+  @Nonnull Observable<Execution> retrievePipelinesForApplication(@Nonnull String application);
 
-  @Nonnull Iterable<Execution> retrievePipelinesForPipelineConfigId(
+  @Nonnull Observable<Execution> retrievePipelinesForPipelineConfigId(
     @Nonnull String pipelineConfigId, @Nonnull ExecutionCriteria criteria);
 
-  @Nonnull Iterable<Execution> retrieveOrchestrationsForApplication(
+  @Nonnull Observable<Execution> retrieveOrchestrationsForApplication(
     @Nonnull String application, @Nonnull ExecutionCriteria criteria);
 
   @Nonnull Execution retrieveOrchestrationForCorrelationId(
@@ -117,11 +117,4 @@ public ExecutionCriteria setStatuses(ExecutionStatus... statuses) {
       return Objects.hash(limit, statuses);
     }
   }
-
-  final class IterableUtil {
-
-    public static <T> Stream<T> toStream(Iterable<T> iterable) {
-      return Lists.newArrayList(iterable).stream();
-    }
-  }
 }

orca-core/src/main/java/com/netflix/spinnaker/orca/pipeline/util/ArtifactResolver.java

@@ -30,15 +30,23 @@
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
+import rx.functions.Func2;
+import rx.schedulers.Schedulers;
 
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 import java.io.IOException;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
-import static com.netflix.spinnaker.orca.pipeline.persistence.ExecutionRepository.IterableUtil.toStream;
 import static java.lang.String.format;
 import static java.util.Collections.emptyList;
 import static java.util.stream.Collectors.toList;
@@ -131,10 +139,15 @@ List<Artifact> getArtifactsForPipelineId(
     @Nonnull String pipelineId,
     @Nonnull ExecutionCriteria criteria
   ) {
-    Execution execution = toStream(executionRepository.retrievePipelinesForPipelineConfigId(pipelineId, criteria))
-      .sorted(startTimeOrId)
-      .findFirst()
-      .orElse(null);
+    Execution execution = executionRepository
+        .retrievePipelinesForPipelineConfigId(pipelineId, criteria)
+        .subscribeOn(Schedulers.io())
+        .toSortedList(startTimeOrId)
+        .toBlocking()
+        .single()
+        .stream()
+        .findFirst()
+        .orElse(null);
 
     return execution == null ? Collections.emptyList() : getAllArtifacts(execution);
   }
@@ -253,7 +266,7 @@ public static class ResolveResult {
     Set<ExpectedArtifact> unresolvedExpectedArtifacts = new HashSet<>();
   }
 
-  private static Comparator<Execution> startTimeOrId = (a, b) -> {
+  private static Func2<Execution, Execution, Integer> startTimeOrId = (a, b) -> {
     Long aStartTime = Optional.ofNullable(a.getStartTime()).orElse(0L);
     Long bStartTime = Optional.ofNullable(b.getStartTime()).orElse(0L);
 

orca-pipelinetemplate/src/main/java/com/netflix/spinnaker/orca/pipelinetemplate/PipelineTemplateService.java

@@ -15,6 +15,10 @@
  */
 package com.netflix.spinnaker.orca.pipelinetemplate;
 
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import javax.annotation.Nullable;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.netflix.spinnaker.orca.jackson.OrcaObjectMapper;
 import com.netflix.spinnaker.orca.pipeline.model.Execution;
@@ -29,13 +33,6 @@
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
-
-import javax.annotation.Nullable;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.NoSuchElementException;
-
 import static com.netflix.spinnaker.orca.pipeline.model.Execution.ExecutionType.PIPELINE;
 
 @Component
@@ -88,12 +85,15 @@ public Execution retrievePipelineOrNewestExecution(@Nullable String executionId,
     } else if (pipelineConfigId != null) {
       // No executionId set - use last execution
       ExecutionRepository.ExecutionCriteria criteria = new ExecutionRepository.ExecutionCriteria().setLimit(1);
-
-      Iterator<Execution> executions = executionRepository.retrievePipelinesForPipelineConfigId(pipelineConfigId, criteria).iterator();
-      if (!executions.hasNext()) {
-        throw new ExecutionNotFoundException("No pipeline execution could be found for config id " + pipelineConfigId);
+      try {
+        return executionRepository.retrievePipelinesForPipelineConfigId(pipelineConfigId, criteria)
+          .toSingle()
+          .toBlocking()
+          .value();
+      } catch (NoSuchElementException e) {
+        throw new ExecutionNotFoundException("No pipeline execution could be found for config id " +
+          pipelineConfigId + ": " + e.getMessage());
       }
-      return executions.next();
     } else {
       throw new IllegalArgumentException("Either executionId or pipelineConfigId have to be set.");
     }

orca-queue/src/main/kotlin/com/netflix/spinnaker/orca/q/handler/OrcaMessageHandler.kt

@@ -94,10 +94,11 @@ internal interface OrcaMessageHandler<M : Message> : MessageHandler<M> {
     }
     return pipelineConfigId?.let { configId ->
       val criteria = ExecutionRepository.ExecutionCriteria().setLimit(1).setStatuses(ExecutionStatus.RUNNING)
-      repository
+      !repository
         .retrievePipelinesForPipelineConfigId(configId, criteria)
-        .iterator()
-        .hasNext()
+        .isEmpty
+        .toBlocking()
+        .first()
     } == true
   }
 }

orca-queue/src/test/kotlin/com/netflix/spinnaker/orca/q/handler/StartExecutionHandlerTest.kt

@@ -37,6 +37,7 @@ import org.jetbrains.spek.api.dsl.*
 import org.jetbrains.spek.api.lifecycle.CachingMode.GROUP
 import org.jetbrains.spek.subject.SubjectSpek
 import org.springframework.context.ApplicationEventPublisher
+import rx.Observable.just
 import java.util.*
 
 object StartExecutionHandlerTest : SubjectSpek<StartExecutionHandler>({
@@ -274,7 +275,10 @@ object StartExecutionHandlerTest : SubjectSpek<StartExecutionHandler>({
           pipeline.isLimitConcurrent = true
           runningPipeline.isLimitConcurrent = true
 
-          whenever(repository.retrievePipelinesForPipelineConfigId(eq(configId), any())) doReturn listOf(runningPipeline)
+          whenever(
+            repository
+              .retrievePipelinesForPipelineConfigId(configId, ExecutionRepository.ExecutionCriteria().setLimit(1).setStatuses(RUNNING))
+          ) doReturn just(runningPipeline)
           whenever(
             repository.retrieve(message.executionType, message.executionId)
           ) doReturn pipeline
@@ -305,7 +309,10 @@ object StartExecutionHandlerTest : SubjectSpek<StartExecutionHandler>({
           pipeline.isLimitConcurrent = false
           runningPipeline.isLimitConcurrent = false
 
-          whenever(repository.retrievePipelinesForPipelineConfigId(eq(configId), any())) doReturn listOf(runningPipeline)
+          whenever(
+            repository
+              .retrievePipelinesForPipelineConfigId(configId, ExecutionRepository.ExecutionCriteria().setLimit(1).setStatuses(RUNNING))
+          ) doReturn just(runningPipeline)
           whenever(
             repository.retrieve(message.executionType, message.executionId)
           ) doReturn pipeline

orca-redis/src/main/java/com/netflix/spinnaker/orca/migrations/MultiRedisOrchestrationMigrationNotificationAgent.groovy

@@ -122,8 +122,10 @@ class MultiRedisOrchestrationMigrationNotificationAgent extends AbstractPollingN
       def applicationName = application.name.toLowerCase()
       def unmigratedOrchestrations = executionRepositoryPrevious
         .retrieveOrchestrationsForApplication(applicationName, executionCriteria)
+        .filter({ orchestration -> !previouslyMigratedOrchestrationIds.contains(orchestration.id) })
         .toList()
-        .findAll { orchestration -> !previouslyMigratedOrchestrationIds.contains(orchestration.id) }
+        .toBlocking()
+        .single()
 
       def migratableOrchestrations = unmigratedOrchestrations.findAll { it.status.isComplete() }
       def pendingOrchestrations = unmigratedOrchestrations.findAll { !it.status.isComplete() }

orca-redis/src/main/java/com/netflix/spinnaker/orca/migrations/MultiRedisPipelineMigrationNotificationAgent.groovy

@@ -120,8 +120,10 @@ class MultiRedisPipelineMigrationNotificationAgent extends AbstractPollingNotifi
     allPipelineConfigIds.eachWithIndex { String pipelineConfigId, int index ->
       def unmigratedPipelines = executionRepositoryPrevious
         .retrievePipelinesForPipelineConfigId(pipelineConfigId, executionCriteria)
+        .filter({ pipeline -> !previouslyMigratedPipelineIds.contains(pipeline.id) })
         .toList()
-        .findAll { pipeline -> !previouslyMigratedPipelineIds.contains(pipeline.id) }
+        .toBlocking()
+        .single()
 
       def migratablePipelines = unmigratedPipelines.findAll { it.status.isComplete() }
       def pendingPipelines = unmigratedPipelines.findAll { !it.status.isComplete() }

orca-redis/src/main/java/com/netflix/spinnaker/orca/notifications/scheduling/OldPipelineCleanupPollingNotificationAgent.java

@@ -31,6 +31,7 @@
 import rx.Observable;
 import rx.Scheduler;
 import rx.Subscription;
+import rx.functions.Func1;
 import rx.schedulers.Schedulers;
 
 import javax.annotation.PreDestroy;
@@ -39,13 +40,9 @@
 import java.time.temporal.ChronoUnit;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
-import java.util.function.Function;
-import java.util.function.Predicate;
 import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 import static com.netflix.spinnaker.orca.pipeline.model.Execution.ExecutionType.PIPELINE;
-import static com.netflix.spinnaker.orca.pipeline.persistence.ExecutionRepository.IterableUtil.toStream;
 
 // TODO rz - Remove redis know-how and move back into orca-core
 @Component
@@ -59,9 +56,9 @@ public class OldPipelineCleanupPollingNotificationAgent implements ApplicationLi
   private Scheduler scheduler = Schedulers.io();
   private Subscription subscription;
 
-  private Predicate<Execution> filter = new Predicate<Execution>() {
+  private Func1<Execution, Boolean> filter = new Func1<Execution, Boolean>() {
     @Override
-    public boolean test(Execution execution) {
+    public Boolean call(Execution execution) {
       if (!COMPLETED_STATUSES.contains(execution.getStatus().toString())) {
         return false;
       }
@@ -70,7 +67,7 @@ public boolean test(Execution execution) {
     }
   };
 
-  private Function<Execution, PipelineExecutionDetails> mapper = execution -> new PipelineExecutionDetails(
+  private Func1<Execution, PipelineExecutionDetails> mapper = execution -> new PipelineExecutionDetails(
     execution.getId(),
     execution.getApplication(),
     execution.getPipelineConfigId() == null ? "ungrouped" : execution.getPipelineConfigId(),
@@ -141,16 +138,16 @@ private void tick() {
 
       applications.forEach(app -> {
         log.debug("Cleaning up " + app);
-        cleanupApp(toStream(executionRepository.retrievePipelinesForApplication(app)));
+        cleanupApp(executionRepository.retrievePipelinesForApplication(app));
       });
 
     } catch (Exception e) {
       log.error("Cleanup failed", e);
     }
   }
 
-  private void cleanupApp(Stream<Execution> executions) {
-    List<PipelineExecutionDetails> allPipelines = executions.filter(filter).map(mapper).collect(Collectors.toList());
+  private void cleanupApp(Observable<Execution> observable) {
+    List<PipelineExecutionDetails> allPipelines = observable.filter(filter).map(mapper).toList().toBlocking().single();
 
     Map<String, List<PipelineExecutionDetails>> groupedPipelines = new HashMap<>();
     allPipelines.forEach(p -> {