Add an interceptor for listExecutions

Author: Quinn-With-Two-Ns

temporal-sdk/src/main/java/io/temporal/client/ListWorkflowExecutionIterator.java

@@ -11,14 +11,14 @@
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 
-class ListWorkflowExecutionIterator
+public class ListWorkflowExecutionIterator
     extends EagerPaginator<ListWorkflowExecutionsResponse, WorkflowExecutionInfo> {
   private final @Nullable String query;
   private final @Nonnull String namespace;
   private final @Nullable Integer pageSize;
   private final @Nonnull GenericWorkflowClient genericClient;
 
-  ListWorkflowExecutionIterator(
+  public ListWorkflowExecutionIterator(
       @Nullable String query,
       @Nonnull String namespace,
       @Nullable Integer pageSize,

temporal-sdk/src/main/java/io/temporal/client/WorkflowClientInternalImpl.java

@@ -4,7 +4,6 @@
 
 import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
-import com.google.common.collect.Iterators;
 import com.google.common.reflect.TypeToken;
 import com.uber.m3.tally.Scope;
 import io.temporal.api.common.v1.WorkflowExecution;
@@ -243,22 +242,10 @@ public Stream<WorkflowExecutionMetadata> listExecutions(@Nullable String query)
 
   Stream<WorkflowExecutionMetadata> listExecutions(
       @Nullable String query, @Nullable Integer pageSize) {
-    ListWorkflowExecutionIterator iterator =
-        new ListWorkflowExecutionIterator(query, options.getNamespace(), pageSize, genericClient);
-    iterator.init();
-    Iterator<WorkflowExecutionMetadata> wrappedIterator =
-        Iterators.transform(
-            iterator, info -> new WorkflowExecutionMetadata(info, options.getDataConverter()));
-
-    // IMMUTABLE here means that "interference" (in Java Streams terms) to this spliterator is
-    // impossible
-    //  TODO We don't add DISTINCT to be safe. It's not explicitly stated if Temporal Server list
-    // API
-    // guarantees absence of duplicates
-    final int CHARACTERISTICS = Spliterator.ORDERED | Spliterator.NONNULL | Spliterator.IMMUTABLE;
-
-    return StreamSupport.stream(
-        Spliterators.spliteratorUnknownSize(wrappedIterator, CHARACTERISTICS), false);
+    return workflowClientCallsInvoker
+        .listWorkflowExecutions(
+            new WorkflowClientCallsInterceptor.ListWorkflowExecutionsInput(query, pageSize))
+        .getStream();
   }
 
   @Override

temporal-sdk/src/main/java/io/temporal/client/WorkflowExecutionMetadata.java

@@ -20,11 +20,12 @@
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 
+/** WorkflowExecutionMetadata contains information about a workflow execution. */
 public class WorkflowExecutionMetadata {
   private final @Nonnull WorkflowExecutionInfo info;
   private final @Nonnull DataConverter dataConverter;
 
-  WorkflowExecutionMetadata(
+  public WorkflowExecutionMetadata(
       @Nonnull WorkflowExecutionInfo info, @Nonnull DataConverter dataConverter) {
     this.info = Preconditions.checkNotNull(info, "info");
     this.dataConverter = Preconditions.checkNotNull(dataConverter, "dataConverter");

temporal-sdk/src/main/java/io/temporal/common/interceptors/WorkflowClientCallsInterceptor.java

@@ -10,6 +10,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.stream.Stream;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 
@@ -75,6 +76,40 @@ public interface WorkflowClientCallsInterceptor {
 
   DescribeWorkflowOutput describe(DescribeWorkflowInput input);
 
+  ListWorkflowExecutionsOutput listWorkflowExecutions(ListWorkflowExecutionsInput input);
+
+  final class ListWorkflowExecutionsInput {
+    private final String query;
+    private final Integer pageSize;
+
+    public ListWorkflowExecutionsInput(@Nullable String query, @Nullable Integer pageSize) {
+      this.query = query;
+      this.pageSize = pageSize;
+    }
+
+    @Nullable
+    public String getQuery() {
+      return query;
+    }
+
+    @Nullable
+    public Integer getPageSize() {
+      return pageSize;
+    }
+  }
+
+  final class ListWorkflowExecutionsOutput {
+    private final Stream<WorkflowExecutionMetadata> stream;
+
+    public ListWorkflowExecutionsOutput(Stream<WorkflowExecutionMetadata> stream) {
+      this.stream = stream;
+    }
+
+    public Stream<WorkflowExecutionMetadata> getStream() {
+      return stream;
+    }
+  }
+
   final class WorkflowStartInput {
     private final String workflowId;
     private final String workflowType;

temporal-sdk/src/main/java/io/temporal/common/interceptors/WorkflowClientCallsInterceptorBase.java

@@ -72,4 +72,9 @@ public TerminateOutput terminate(TerminateInput input) {
   public DescribeWorkflowOutput describe(DescribeWorkflowInput input) {
     return next.describe(input);
   }
+
+  @Override
+  public ListWorkflowExecutionsOutput listWorkflowExecutions(ListWorkflowExecutionsInput input) {
+    return next.listWorkflowExecutions(input);
+  }
 }

temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java

@@ -30,10 +30,13 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.stream.StreamSupport;
 import javax.annotation.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.google.common.collect.Iterators;
+
 public class RootWorkflowClientInvoker implements WorkflowClientCallsInterceptor {
   private static final Logger log = LoggerFactory.getLogger(RootWorkflowClientInvoker.class);
   private static final long POLL_UPDATE_TIMEOUT_S = 60L;
@@ -676,6 +679,29 @@ public DescribeWorkflowOutput describe(DescribeWorkflowInput input) {
         new WorkflowExecutionDescription(response, dataConverterWithWorkflowContext));
   }
 
+  @Override
+  public ListWorkflowExecutionsOutput listWorkflowExecutions(ListWorkflowExecutionsInput input) {
+    ListWorkflowExecutionIterator iterator =
+        new ListWorkflowExecutionIterator(
+            input.getQuery(), clientOptions.getNamespace(), input.getPageSize(), genericClient);
+    iterator.init();
+    Iterator<WorkflowExecutionMetadata> wrappedIterator =
+        Iterators.transform(
+            iterator,
+            info -> new WorkflowExecutionMetadata(info, clientOptions.getDataConverter()));
+
+    // IMMUTABLE here means that "interference" (in Java Streams terms) to this spliterator is
+    // impossible
+    //  TODO We don't add DISTINCT to be safe. It's not explicitly stated if Temporal Server list
+    // API
+    // guarantees absence of duplicates
+    final int CHARACTERISTICS = Spliterator.ORDERED | Spliterator.NONNULL | Spliterator.IMMUTABLE;
+
+    return new ListWorkflowExecutionsOutput(
+        StreamSupport.stream(
+            Spliterators.spliteratorUnknownSize(wrappedIterator, CHARACTERISTICS), false));
+  }
+
   private static <R> R convertResultPayloads(
       Optional<Payloads> resultValue,
       Class<R> resultClass,

temporal-sdk/src/test/java/io/temporal/client/ListWorkflowExecutionsInterceptorTest.java

@@ -0,0 +1,63 @@
+package io.temporal.client;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assume.assumeTrue;
+
+import io.temporal.common.interceptors.WorkflowClientCallsInterceptor;
+import io.temporal.common.interceptors.WorkflowClientCallsInterceptorBase;
+import io.temporal.common.interceptors.WorkflowClientInterceptorBase;
+import io.temporal.testing.internal.SDKTestWorkflowRule;
+import io.temporal.workflow.shared.TestWorkflows;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.junit.Rule;
+import org.junit.Test;
+
+public class ListWorkflowExecutionsInterceptorTest {
+  @Rule
+  public SDKTestWorkflowRule testWorkflowRule =
+      SDKTestWorkflowRule.newBuilder()
+          .setWorkflowTypes(TestWorkflows.DoNothingNoArgsWorkflow.class)
+          .build();
+
+  @Test
+  public void listExecutions_isIntercepted() throws InterruptedException {
+    assumeTrue(
+        "Test Server doesn't support listWorkflowExecutions endpoint yet",
+        SDKTestWorkflowRule.useExternalService);
+
+    AtomicInteger intercepted = new AtomicInteger();
+    WorkflowClient workflowClient =
+        WorkflowClient.newInstance(
+            testWorkflowRule.getWorkflowServiceStubs(),
+            WorkflowClientOptions.newBuilder(testWorkflowRule.getWorkflowClient().getOptions())
+                .setInterceptors(
+                    new WorkflowClientInterceptorBase() {
+                      @Override
+                      public WorkflowClientCallsInterceptor workflowClientCallsInterceptor(
+                          WorkflowClientCallsInterceptor next) {
+                        return new WorkflowClientCallsInterceptorBase(next) {
+                          @Override
+                          public ListWorkflowExecutionsOutput listWorkflowExecutions(
+                              ListWorkflowExecutionsInput input) {
+                            intercepted.incrementAndGet();
+                            return super.listWorkflowExecutions(input);
+                          }
+                        };
+                      }
+                    })
+                .validateAndBuildWithDefaults());
+
+    WorkflowStub.fromTyped(testWorkflowRule.newWorkflowStub(TestWorkflows.NoArgsWorkflow.class))
+        .start();
+
+    // Visibility API is eventually consistent
+    Thread.sleep(2_000);
+    java.util.List<WorkflowExecutionMetadata> result =
+        workflowClient
+            .listExecutions("TaskQueue='" + testWorkflowRule.getTaskQueue() + "'")
+            .collect(java.util.stream.Collectors.toList());
+    assertFalse(result.isEmpty());
+    assertEquals(1, intercepted.get());
+  }
+}