#153

pending completion

Build # #153

Build Type

push

github-actions

Committed by

web-flow

Commit Message

Eager Workflow Task Dispatch (#1674)

Issue #1646

Signed-off-by: Dmitry Spikhalskiy <dmitry@spikhalskiy.com>

Run Details

213 of 213 new or added lines in 22 files covered. (100.0%)

16682 of 20566 relevant lines covered (81.11%)

0.81 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

86.39

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

/*
 * Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
 *
 * Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
 *
 * Modifications copyright (C) 2017 Uber Technologies, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this material except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.temporal.client;

import static io.temporal.internal.WorkflowThreadMarker.enforceNonWorkflowThread;

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;
import io.temporal.api.history.v1.History;
import io.temporal.api.history.v1.HistoryEvent;
import io.temporal.client.WorkflowInvocationHandler.InvocationType;
import io.temporal.common.WorkflowExecutionHistory;
import io.temporal.common.interceptors.WorkflowClientCallsInterceptor;
import io.temporal.common.interceptors.WorkflowClientInterceptor;
import io.temporal.internal.WorkflowThreadMarker;
import io.temporal.internal.client.RootWorkflowClientInvoker;
import io.temporal.internal.client.WorkerFactoryRegistry;
import io.temporal.internal.client.WorkflowClientInternal;
import io.temporal.internal.client.external.GenericWorkflowClient;
import io.temporal.internal.client.external.GenericWorkflowClientImpl;
import io.temporal.internal.client.external.ManualActivityCompletionClientFactory;
import io.temporal.internal.sync.StubMarker;
import io.temporal.serviceclient.MetricsTag;
import io.temporal.serviceclient.WorkflowServiceStubs;
import io.temporal.worker.WorkerFactory;
import io.temporal.workflow.Functions;
import io.temporal.workflow.QueryMethod;
import io.temporal.workflow.SignalMethod;
import io.temporal.workflow.WorkflowMethod;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;

final class WorkflowClientInternalImpl implements WorkflowClient, WorkflowClientInternal {

  private final GenericWorkflowClient genericClient;
  private final WorkflowClientOptions options;
  private final ManualActivityCompletionClientFactory manualActivityCompletionClientFactory;
  private final WorkflowClientCallsInterceptor workflowClientCallsInvoker;
  private final WorkflowServiceStubs workflowServiceStubs;
  private final Scope metricsScope;
  private final WorkflowClientInterceptor[] interceptors;
  private final WorkerFactoryRegistry workerFactoryRegistry = new WorkerFactoryRegistry();

  /**
   * Creates client that connects to an instance of the Temporal Service. Cannot be used from within
   * workflow code.
   *
   * @param service client to the Temporal Service endpoint.
   * @param options Options (like {@link io.temporal.common.converter.DataConverter} override) for
   *     configuring client.
   */
  public static WorkflowClient newInstance(
      WorkflowServiceStubs service, WorkflowClientOptions options) {
    enforceNonWorkflowThread();
    return WorkflowThreadMarker.protectFromWorkflowThread(
        new WorkflowClientInternalImpl(service, options), WorkflowClient.class);
  }

  WorkflowClientInternalImpl(
      WorkflowServiceStubs workflowServiceStubs, WorkflowClientOptions options) {
    options = WorkflowClientOptions.newBuilder(options).validateAndBuildWithDefaults();
    this.options = options;
    this.workflowServiceStubs = workflowServiceStubs;
    this.metricsScope =
        workflowServiceStubs
            .getOptions()
            .getMetricsScope()
            .tagged(MetricsTag.defaultTags(options.getNamespace()));
    this.genericClient = new GenericWorkflowClientImpl(workflowServiceStubs, metricsScope);
    this.interceptors = options.getInterceptors();
    this.workflowClientCallsInvoker = initializeClientInvoker();
    this.manualActivityCompletionClientFactory =
        ManualActivityCompletionClientFactory.newFactory(
            workflowServiceStubs,
            options.getNamespace(),
            options.getIdentity(),
            options.getDataConverter());
  }

  private WorkflowClientCallsInterceptor initializeClientInvoker() {
    WorkflowClientCallsInterceptor workflowClientInvoker =
        new RootWorkflowClientInvoker(genericClient, options, workerFactoryRegistry);
    for (WorkflowClientInterceptor clientInterceptor : interceptors) {
      workflowClientInvoker =
          clientInterceptor.workflowClientCallsInterceptor(workflowClientInvoker);
    }
    return workflowClientInvoker;
  }

  @Override
  public WorkflowServiceStubs getWorkflowServiceStubs() {
    return workflowServiceStubs;
  }

  @Override
  public WorkflowClientOptions getOptions() {
    return options;
  }

  @Override
  @SuppressWarnings("unchecked")
  public <T> T newWorkflowStub(Class<T> workflowInterface, WorkflowOptions options) {
    checkAnnotation(workflowInterface, WorkflowMethod.class);
    WorkflowInvocationHandler invocationHandler =
        new WorkflowInvocationHandler(
            workflowInterface, this.getOptions(), workflowClientCallsInvoker, options);
    return (T)
        Proxy.newProxyInstance(
            workflowInterface.getClassLoader(),
            new Class<?>[] {workflowInterface, StubMarker.class},
            invocationHandler);
  }

  @SafeVarargs
  private static <T> void checkAnnotation(
      Class<T> workflowInterface, Class<? extends Annotation>... annotationClasses) {
    TypeToken<?>.TypeSet interfaces = TypeToken.of(workflowInterface).getTypes().interfaces();
    if (interfaces.isEmpty()) {
      throw new IllegalArgumentException("Workflow must implement at least one interface");
    }
    for (TypeToken<?> i : interfaces) {
      for (Method method : i.getRawType().getMethods()) {
        for (Class<? extends Annotation> annotationClass : annotationClasses) {
          Object workflowMethod = method.getAnnotation(annotationClass);
          if (workflowMethod != null) {
            return;
          }
        }
      }
    }
    throw new IllegalArgumentException(
        "Workflow interface "
            + workflowInterface.getName()
            + " doesn't have method annotated with any of "
            + Arrays.toString(annotationClasses));
  }

  @Override
  public <T> T newWorkflowStub(Class<T> workflowInterface, String workflowId) {
    return newWorkflowStub(workflowInterface, workflowId, Optional.empty());
  }

  @Override
  public <T> T newWorkflowStub(
      Class<T> workflowInterface, String workflowId, Optional<String> runId) {
    checkAnnotation(workflowInterface, WorkflowMethod.class, QueryMethod.class, SignalMethod.class);
    if (Strings.isNullOrEmpty(workflowId)) {
      throw new IllegalArgumentException("workflowId is null or empty");
    }
    WorkflowExecution execution =
        WorkflowExecution.newBuilder().setWorkflowId(workflowId).setRunId(runId.orElse("")).build();

    WorkflowInvocationHandler invocationHandler =
        new WorkflowInvocationHandler(
            workflowInterface, this.getOptions(), workflowClientCallsInvoker, execution);
    @SuppressWarnings("unchecked")
    T result =
        (T)
            Proxy.newProxyInstance(
                workflowInterface.getClassLoader(),
                new Class<?>[] {workflowInterface, StubMarker.class},
                invocationHandler);
    return result;
  }

  @Override
  public WorkflowStub newUntypedWorkflowStub(String workflowId) {
    return newUntypedWorkflowStub(workflowId, Optional.empty(), Optional.empty());
  }

  @Override
  @SuppressWarnings("deprecation")
  public WorkflowStub newUntypedWorkflowStub(String workflowType, WorkflowOptions workflowOptions) {
    WorkflowStub result =
        new WorkflowStubImpl(options, workflowClientCallsInvoker, workflowType, workflowOptions);
    for (WorkflowClientInterceptor i : interceptors) {
      result = i.newUntypedWorkflowStub(workflowType, workflowOptions, result);
    }
    return result;
  }

  @Override
  public WorkflowStub newUntypedWorkflowStub(
      String workflowId, Optional<String> runId, Optional<String> workflowType) {
    WorkflowExecution execution =
        WorkflowExecution.newBuilder().setWorkflowId(workflowId).setRunId(runId.orElse("")).build();
    return newUntypedWorkflowStub(execution, workflowType);
  }

  @Override
  @SuppressWarnings("deprecation")
  public WorkflowStub newUntypedWorkflowStub(
      WorkflowExecution execution, Optional<String> workflowType) {
    WorkflowStub result =
        new WorkflowStubImpl(options, workflowClientCallsInvoker, workflowType, execution);
    for (WorkflowClientInterceptor i : interceptors) {
      result = i.newUntypedWorkflowStub(execution, workflowType, result);
    }
    return result;
  }

  @Override
  public ActivityCompletionClient newActivityCompletionClient() {
    ActivityCompletionClient result =
        WorkflowThreadMarker.protectFromWorkflowThread(
            new ActivityCompletionClientImpl(
                manualActivityCompletionClientFactory, () -> {}, metricsScope),
            ActivityCompletionClient.class);
    for (WorkflowClientInterceptor i : interceptors) {
      result = i.newActivityCompletionClient(result);
    }
    return result;
  }

  @Override
  public BatchRequest newSignalWithStartRequest() {
    return new SignalWithStartBatchRequest();
  }

  @Override
  public WorkflowExecution signalWithStart(BatchRequest signalWithStartBatch) {
    return ((SignalWithStartBatchRequest) signalWithStartBatch).invoke();
  }

  @Override
  public Stream<WorkflowExecutionMetadata> listExecutions(@Nullable String query) {
    return listExecutions(query, null);
  }

  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);
  }

  @Override
  public Stream<HistoryEvent> streamHistory(@Nonnull String workflowId) {
    return streamHistory(workflowId, null);
  }

  @Override
  public Stream<HistoryEvent> streamHistory(@Nonnull String workflowId, @Nullable String runId) {
    Preconditions.checkNotNull(workflowId, "workflowId is required");

    WorkflowExecution.Builder executionBuilder =
        WorkflowExecution.newBuilder().setWorkflowId(workflowId);
    if (runId != null) {
      executionBuilder.setRunId(runId);
    }
    WorkflowExecution execution = executionBuilder.build();

    return streamHistory(execution);
  }

  @Override
  public WorkflowExecutionHistory fetchHistory(@Nonnull String workflowId) {
    return fetchHistory(workflowId, null);
  }

  @Override
  public WorkflowExecutionHistory fetchHistory(@Nonnull String workflowId, @Nullable String runId) {
    Preconditions.checkNotNull(workflowId, "execution is required");

    return new WorkflowExecutionHistory(
        History.newBuilder()
            .addAllEvents(streamHistory(workflowId, runId).collect(Collectors.toList()))
            .build(),
        workflowId);
  }

  public static WorkflowExecution start(Functions.Proc workflow) {
    enforceNonWorkflowThread();
    WorkflowInvocationHandler.initAsyncInvocation(InvocationType.START);
    try {
      workflow.apply();
      return WorkflowInvocationHandler.getAsyncInvocationResult(WorkflowExecution.class);
    } finally {
      WorkflowInvocationHandler.closeAsyncInvocation();
    }
  }

  public static <A1> WorkflowExecution start(Functions.Proc1<A1> workflow, A1 arg1) {
    return start(() -> workflow.apply(arg1));
  }

  public static <A1, A2> WorkflowExecution start(
      Functions.Proc2<A1, A2> workflow, A1 arg1, A2 arg2) {
    return start(() -> workflow.apply(arg1, arg2));
  }

  public static <A1, A2, A3> WorkflowExecution start(
      Functions.Proc3<A1, A2, A3> workflow, A1 arg1, A2 arg2, A3 arg3) {
    return start(() -> workflow.apply(arg1, arg2, arg3));
  }

  public static <A1, A2, A3, A4> WorkflowExecution start(
      Functions.Proc4<A1, A2, A3, A4> workflow, A1 arg1, A2 arg2, A3 arg3, A4 arg4) {
    return start(() -> workflow.apply(arg1, arg2, arg3, arg4));
  }

  public static <A1, A2, A3, A4, A5> WorkflowExecution start(
      Functions.Proc5<A1, A2, A3, A4, A5> workflow, A1 arg1, A2 arg2, A3 arg3, A4 arg4, A5 arg5) {
    return start(() -> workflow.apply(arg1, arg2, arg3, arg4, arg5));
  }

  public static <A1, A2, A3, A4, A5, A6> WorkflowExecution start(
      Functions.Proc6<A1, A2, A3, A4, A5, A6> workflow,
      A1 arg1,
      A2 arg2,
      A3 arg3,
      A4 arg4,
      A5 arg5,
      A6 arg6) {
    return start(() -> workflow.apply(arg1, arg2, arg3, arg4, arg5, arg6));
  }

  public static <R> WorkflowExecution start(Functions.Func<R> workflow) {
    return start((Functions.Proc) workflow::apply);
  }

  public static <A1, R> WorkflowExecution start(Functions.Func1<A1, R> workflow, A1 arg1) {
    return start(() -> workflow.apply(arg1));
  }

  public static <A1, A2, R> WorkflowExecution start(
      Functions.Func2<A1, A2, R> workflow, A1 arg1, A2 arg2) {
    return start(() -> workflow.apply(arg1, arg2));
  }

  public static <A1, A2, A3, R> WorkflowExecution start(
      Functions.Func3<A1, A2, A3, R> workflow, A1 arg1, A2 arg2, A3 arg3) {
    return start(() -> workflow.apply(arg1, arg2, arg3));
  }

  public static <A1, A2, A3, A4, R> WorkflowExecution start(
      Functions.Func4<A1, A2, A3, A4, R> workflow, A1 arg1, A2 arg2, A3 arg3, A4 arg4) {
    return start(() -> workflow.apply(arg1, arg2, arg3, arg4));
  }

  public static <A1, A2, A3, A4, A5, R> WorkflowExecution start(
      Functions.Func5<A1, A2, A3, A4, A5, R> workflow,
      A1 arg1,
      A2 arg2,
      A3 arg3,
      A4 arg4,
      A5 arg5) {
    return start(() -> workflow.apply(arg1, arg2, arg3, arg4, arg5));
  }

  public static <A1, A2, A3, A4, A5, A6, R> WorkflowExecution start(
      Functions.Func6<A1, A2, A3, A4, A5, A6, R> workflow,
      A1 arg1,
      A2 arg2,
      A3 arg3,
      A4 arg4,
      A5 arg5,
      A6 arg6) {
    return start(() -> workflow.apply(arg1, arg2, arg3, arg4, arg5, arg6));
  }

  @SuppressWarnings("unchecked")
  public static CompletableFuture<Void> execute(Functions.Proc workflow) {
    enforceNonWorkflowThread();
    WorkflowInvocationHandler.initAsyncInvocation(InvocationType.EXECUTE);
    try {
      workflow.apply();
      return WorkflowInvocationHandler.getAsyncInvocationResult(CompletableFuture.class);
    } finally {
      WorkflowInvocationHandler.closeAsyncInvocation();
    }
  }

  public static <A1> CompletableFuture<Void> execute(Functions.Proc1<A1> workflow, A1 arg1) {
    return execute(() -> workflow.apply(arg1));
  }

  public static <A1, A2> CompletableFuture<Void> execute(
      Functions.Proc2<A1, A2> workflow, A1 arg1, A2 arg2) {
    return execute(() -> workflow.apply(arg1, arg2));
  }

  public static <A1, A2, A3> CompletableFuture<Void> execute(
      Functions.Proc3<A1, A2, A3> workflow, A1 arg1, A2 arg2, A3 arg3) {
    return execute(() -> workflow.apply(arg1, arg2, arg3));
  }

  public static <A1, A2, A3, A4> CompletableFuture<Void> execute(
      Functions.Proc4<A1, A2, A3, A4> workflow, A1 arg1, A2 arg2, A3 arg3, A4 arg4) {
    return execute(() -> workflow.apply(arg1, arg2, arg3, arg4));
  }

  public static <A1, A2, A3, A4, A5> CompletableFuture<Void> execute(
      Functions.Proc5<A1, A2, A3, A4, A5> workflow, A1 arg1, A2 arg2, A3 arg3, A4 arg4, A5 arg5) {
    return execute(() -> workflow.apply(arg1, arg2, arg3, arg4, arg5));
  }

  public static <A1, A2, A3, A4, A5, A6> CompletableFuture<Void> execute(
      Functions.Proc6<A1, A2, A3, A4, A5, A6> workflow,
      A1 arg1,
      A2 arg2,
      A3 arg3,
      A4 arg4,
      A5 arg5,
      A6 arg6) {
    return execute(() -> workflow.apply(arg1, arg2, arg3, arg4, arg5, arg6));
  }

  @SuppressWarnings("unchecked")
  public static <R> CompletableFuture<R> execute(Functions.Func<R> workflow) {
    return (CompletableFuture<R>) execute((Functions.Proc) workflow::apply);
  }

  public static <A1, R> CompletableFuture<R> execute(Functions.Func1<A1, R> workflow, A1 arg1) {
    return execute(() -> workflow.apply(arg1));
  }

  public static <A1, A2, R> CompletableFuture<R> execute(
      Functions.Func2<A1, A2, R> workflow, A1 arg1, A2 arg2) {
    return execute(() -> workflow.apply(arg1, arg2));
  }

  public static <A1, A2, A3, R> CompletableFuture<R> execute(
      Functions.Func3<A1, A2, A3, R> workflow, A1 arg1, A2 arg2, A3 arg3) {
    return execute(() -> workflow.apply(arg1, arg2, arg3));
  }

  public static <A1, A2, A3, A4, R> CompletableFuture<R> execute(
      Functions.Func4<A1, A2, A3, A4, R> workflow, A1 arg1, A2 arg2, A3 arg3, A4 arg4) {
    return execute(() -> workflow.apply(arg1, arg2, arg3, arg4));
  }

  public static <A1, A2, A3, A4, A5, R> CompletableFuture<R> execute(
      Functions.Func5<A1, A2, A3, A4, A5, R> workflow,
      A1 arg1,
      A2 arg2,
      A3 arg3,
      A4 arg4,
      A5 arg5) {
    return execute(() -> workflow.apply(arg1, arg2, arg3, arg4, arg5));
  }

  public static <A1, A2, A3, A4, A5, A6, R> CompletableFuture<R> execute(
      Functions.Func6<A1, A2, A3, A4, A5, A6, R> workflow,
      A1 arg1,
      A2 arg2,
      A3 arg3,
      A4 arg4,
      A5 arg5,
      A6 arg6) {
    return execute(() -> workflow.apply(arg1, arg2, arg3, arg4, arg5, arg6));
  }

  Stream<HistoryEvent> streamHistory(WorkflowExecution execution) {
    Preconditions.checkNotNull(execution, "execution is required");

    GetWorkflowExecutionHistoryIterator iterator =
        new GetWorkflowExecutionHistoryIterator(
            options.getNamespace(), execution, null, genericClient);
    iterator.init();

    // IMMUTABLE here means that "interference" (in Java Streams terms) to this spliterator is
    // impossible
    final int CHARACTERISTICS =
        Spliterator.ORDERED | Spliterator.NONNULL | Spliterator.DISTINCT | Spliterator.IMMUTABLE;

    return StreamSupport.stream(
        Spliterators.spliteratorUnknownSize(iterator, CHARACTERISTICS), false);
  }

  @Override
  public Object getInternal() {
    return this;
  }

  @Override
  public void registerWorkerFactory(WorkerFactory workerFactory) {
    workerFactoryRegistry.register(workerFactory);
  }

  @Override
  public void deregisterWorkerFactory(WorkerFactory workerFactory) {
    workerFactoryRegistry.deregister(workerFactory);
  }
}

1	/*
2	* Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
3	*
4	* Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
5	*
6	* Modifications copyright (C) 2017 Uber Technologies, Inc.
7	*
8	* Licensed under the Apache License, Version 2.0 (the "License");
9	* you may not use this material except in compliance with the License.
10	* You may obtain a copy of the License at
11	*
12	* http://www.apache.org/licenses/LICENSE-2.0
13	*
14	* Unless required by applicable law or agreed to in writing, software
15	* distributed under the License is distributed on an "AS IS" BASIS,
16	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17	* See the License for the specific language governing permissions and
18	* limitations under the License.
19	*/
20
21	package io.temporal.client;
22
23	import static io.temporal.internal.WorkflowThreadMarker.enforceNonWorkflowThread;
24
25	import com.google.common.base.Preconditions;
26	import com.google.common.base.Strings;
27	import com.google.common.collect.Iterators;
28	import com.google.common.reflect.TypeToken;
29	import com.uber.m3.tally.Scope;
30	import io.temporal.api.common.v1.WorkflowExecution;
31	import io.temporal.api.history.v1.History;
32	import io.temporal.api.history.v1.HistoryEvent;
33	import io.temporal.client.WorkflowInvocationHandler.InvocationType;
34	import io.temporal.common.WorkflowExecutionHistory;
35	import io.temporal.common.interceptors.WorkflowClientCallsInterceptor;
36	import io.temporal.common.interceptors.WorkflowClientInterceptor;
37	import io.temporal.internal.WorkflowThreadMarker;
38	import io.temporal.internal.client.RootWorkflowClientInvoker;
39	import io.temporal.internal.client.WorkerFactoryRegistry;
40	import io.temporal.internal.client.WorkflowClientInternal;
41	import io.temporal.internal.client.external.GenericWorkflowClient;
42	import io.temporal.internal.client.external.GenericWorkflowClientImpl;
43	import io.temporal.internal.client.external.ManualActivityCompletionClientFactory;
44	import io.temporal.internal.sync.StubMarker;
45	import io.temporal.serviceclient.MetricsTag;
46	import io.temporal.serviceclient.WorkflowServiceStubs;
47	import io.temporal.worker.WorkerFactory;
48	import io.temporal.workflow.Functions;
49	import io.temporal.workflow.QueryMethod;
50	import io.temporal.workflow.SignalMethod;
51	import io.temporal.workflow.WorkflowMethod;
52	import java.lang.annotation.Annotation;
53	import java.lang.reflect.Method;
54	import java.lang.reflect.Proxy;
55	import java.util.*;
56	import java.util.concurrent.CompletableFuture;
57	import java.util.stream.Collectors;
58	import java.util.stream.Stream;
59	import java.util.stream.StreamSupport;
60	import javax.annotation.Nonnull;
61	import javax.annotation.Nullable;
62
63	final class WorkflowClientInternalImpl implements WorkflowClient, WorkflowClientInternal {
64
65	private final GenericWorkflowClient genericClient;
66	private final WorkflowClientOptions options;
67	private final ManualActivityCompletionClientFactory manualActivityCompletionClientFactory;
68	private final WorkflowClientCallsInterceptor workflowClientCallsInvoker;
69	private final WorkflowServiceStubs workflowServiceStubs;
70	private final Scope metricsScope;
71	private final WorkflowClientInterceptor[] interceptors;
72	private final WorkerFactoryRegistry workerFactoryRegistry = new WorkerFactoryRegistry();	1✔
73
74	/**
75	* Creates client that connects to an instance of the Temporal Service. Cannot be used from within
76	* workflow code.
77	*
78	* @param service client to the Temporal Service endpoint.
79	* @param options Options (like {@link io.temporal.common.converter.DataConverter} override) for
80	* configuring client.
81	*/
82	public static WorkflowClient newInstance(
83	WorkflowServiceStubs service, WorkflowClientOptions options) {
84	enforceNonWorkflowThread();	1✔
85	return WorkflowThreadMarker.protectFromWorkflowThread(	1✔
86	new WorkflowClientInternalImpl(service, options), WorkflowClient.class);
87	}
88
89	WorkflowClientInternalImpl(
90	WorkflowServiceStubs workflowServiceStubs, WorkflowClientOptions options) {	1✔
91	options = WorkflowClientOptions.newBuilder(options).validateAndBuildWithDefaults();	1✔
92	this.options = options;	1✔
93	this.workflowServiceStubs = workflowServiceStubs;	1✔
94	this.metricsScope =	1✔
95	workflowServiceStubs
96	.getOptions()	1✔
97	.getMetricsScope()	1✔
98	.tagged(MetricsTag.defaultTags(options.getNamespace()));	1✔
99	this.genericClient = new GenericWorkflowClientImpl(workflowServiceStubs, metricsScope);	1✔
100	this.interceptors = options.getInterceptors();	1✔
101	this.workflowClientCallsInvoker = initializeClientInvoker();	1✔
102	this.manualActivityCompletionClientFactory =	1✔
103	ManualActivityCompletionClientFactory.newFactory(	1✔
104	workflowServiceStubs,
105	options.getNamespace(),	1✔
106	options.getIdentity(),	1✔
107	options.getDataConverter());	1✔
108	}	1✔
109
110	private WorkflowClientCallsInterceptor initializeClientInvoker() {
111	WorkflowClientCallsInterceptor workflowClientInvoker =	1✔
112	new RootWorkflowClientInvoker(genericClient, options, workerFactoryRegistry);
113	for (WorkflowClientInterceptor clientInterceptor : interceptors) {	1✔
114	workflowClientInvoker =	1✔
115	clientInterceptor.workflowClientCallsInterceptor(workflowClientInvoker);	1✔
116	}
117	return workflowClientInvoker;	1✔
118	}
119
120	@Override
121	public WorkflowServiceStubs getWorkflowServiceStubs() {
122	return workflowServiceStubs;	1✔
123	}
124
125	@Override
126	public WorkflowClientOptions getOptions() {
127	return options;	1✔
128	}
129
130	@Override
131	@SuppressWarnings("unchecked")
132	public <T> T newWorkflowStub(Class<T> workflowInterface, WorkflowOptions options) {
133	checkAnnotation(workflowInterface, WorkflowMethod.class);	1✔
134	WorkflowInvocationHandler invocationHandler =	1✔
135	new WorkflowInvocationHandler(
136	workflowInterface, this.getOptions(), workflowClientCallsInvoker, options);	1✔
137	return (T)	1✔
138	Proxy.newProxyInstance(	1✔
139	workflowInterface.getClassLoader(),	1✔
140	new Class<?>[] {workflowInterface, StubMarker.class},
141	invocationHandler);
142	}
143
144	@SafeVarargs
145	private static <T> void checkAnnotation(
146	Class<T> workflowInterface, Class<? extends Annotation>... annotationClasses) {
147	TypeToken<?>.TypeSet interfaces = TypeToken.of(workflowInterface).getTypes().interfaces();	1✔
148	if (interfaces.isEmpty()) {	1✔
149	throw new IllegalArgumentException("Workflow must implement at least one interface");	×
150	}
151	for (TypeToken<?> i : interfaces) {	1✔
152	for (Method method : i.getRawType().getMethods()) {	1✔
153	for (Class<? extends Annotation> annotationClass : annotationClasses) {	1✔
154	Object workflowMethod = method.getAnnotation(annotationClass);	1✔
155	if (workflowMethod != null) {	1✔
156	return;	1✔
157	}
158	}
159	}
160	}	×
161	throw new IllegalArgumentException(	×
162	"Workflow interface "
163	+ workflowInterface.getName()	×
164	+ " doesn't have method annotated with any of "
165	+ Arrays.toString(annotationClasses));	×
166	}
167
168	@Override
169	public <T> T newWorkflowStub(Class<T> workflowInterface, String workflowId) {
170	return newWorkflowStub(workflowInterface, workflowId, Optional.empty());	1✔
171	}
172
173	@Override
174	public <T> T newWorkflowStub(
175	Class<T> workflowInterface, String workflowId, Optional<String> runId) {
176	checkAnnotation(workflowInterface, WorkflowMethod.class, QueryMethod.class, SignalMethod.class);	1✔
177	if (Strings.isNullOrEmpty(workflowId)) {	1✔
178	throw new IllegalArgumentException("workflowId is null or empty");	×
179	}
180	WorkflowExecution execution =
181	WorkflowExecution.newBuilder().setWorkflowId(workflowId).setRunId(runId.orElse("")).build();	1✔
182
183	WorkflowInvocationHandler invocationHandler =	1✔
184	new WorkflowInvocationHandler(
185	workflowInterface, this.getOptions(), workflowClientCallsInvoker, execution);	1✔
186	@SuppressWarnings("unchecked")
187	T result =	1✔
188	(T)
189	Proxy.newProxyInstance(	1✔
190	workflowInterface.getClassLoader(),	1✔
191	new Class<?>[] {workflowInterface, StubMarker.class},
192	invocationHandler);
193	return result;	1✔
194	}
195
196	@Override
197	public WorkflowStub newUntypedWorkflowStub(String workflowId) {
198	return newUntypedWorkflowStub(workflowId, Optional.empty(), Optional.empty());	×
199	}
200
201	@Override
202	@SuppressWarnings("deprecation")
203	public WorkflowStub newUntypedWorkflowStub(String workflowType, WorkflowOptions workflowOptions) {
204	WorkflowStub result =	1✔
205	new WorkflowStubImpl(options, workflowClientCallsInvoker, workflowType, workflowOptions);
206	for (WorkflowClientInterceptor i : interceptors) {	1✔
207	result = i.newUntypedWorkflowStub(workflowType, workflowOptions, result);	1✔
208	}
209	return result;	1✔
210	}
211
212	@Override
213	public WorkflowStub newUntypedWorkflowStub(
214	String workflowId, Optional<String> runId, Optional<String> workflowType) {
215	WorkflowExecution execution =
216	WorkflowExecution.newBuilder().setWorkflowId(workflowId).setRunId(runId.orElse("")).build();	×
217	return newUntypedWorkflowStub(execution, workflowType);	×
218	}
219
220	@Override
221	@SuppressWarnings("deprecation")
222	public WorkflowStub newUntypedWorkflowStub(
223	WorkflowExecution execution, Optional<String> workflowType) {
224	WorkflowStub result =	1✔
225	new WorkflowStubImpl(options, workflowClientCallsInvoker, workflowType, execution);
226	for (WorkflowClientInterceptor i : interceptors) {	1✔
227	result = i.newUntypedWorkflowStub(execution, workflowType, result);	1✔
228	}
229	return result;	1✔
230	}
231
232	@Override
233	public ActivityCompletionClient newActivityCompletionClient() {
234	ActivityCompletionClient result =	1✔
235	WorkflowThreadMarker.protectFromWorkflowThread(	1✔
236	new ActivityCompletionClientImpl(
237	manualActivityCompletionClientFactory, () -> {}, metricsScope),	1✔
238	ActivityCompletionClient.class);
239	for (WorkflowClientInterceptor i : interceptors) {	1✔
240	result = i.newActivityCompletionClient(result);	1✔
241	}
242	return result;	1✔
243	}
244
245	@Override
246	public BatchRequest newSignalWithStartRequest() {
247	return new SignalWithStartBatchRequest();	1✔
248	}
249
250	@Override
251	public WorkflowExecution signalWithStart(BatchRequest signalWithStartBatch) {
252	return ((SignalWithStartBatchRequest) signalWithStartBatch).invoke();	1✔
253	}
254
255	@Override
256	public Stream<WorkflowExecutionMetadata> listExecutions(@Nullable String query) {
257	return listExecutions(query, null);	×
258	}
259
260	Stream<WorkflowExecutionMetadata> listExecutions(
261	@Nullable String query, @Nullable Integer pageSize) {
262	ListWorkflowExecutionIterator iterator =	×
263	new ListWorkflowExecutionIterator(query, options.getNamespace(), pageSize, genericClient);	×
264	iterator.init();	×
265	Iterator<WorkflowExecutionMetadata> wrappedIterator =	×
266	Iterators.transform(	×
267	iterator, info -> new WorkflowExecutionMetadata(info, options.getDataConverter()));	×
268
269	// IMMUTABLE here means that "interference" (in Java Streams terms) to this spliterator is
270	// impossible
271	// TODO We don't add DISTINCT to be safe. It's not explicitly stated if Temporal Server list
272	// API
273	// guarantees absence of duplicates
274	final int CHARACTERISTICS = Spliterator.ORDERED \| Spliterator.NONNULL \| Spliterator.IMMUTABLE;	×
275
276	return StreamSupport.stream(	×
277	Spliterators.spliteratorUnknownSize(wrappedIterator, CHARACTERISTICS), false);	×
278	}
279
280	@Override
281	public Stream<HistoryEvent> streamHistory(@Nonnull String workflowId) {
282	return streamHistory(workflowId, null);	×
283	}
284
285	@Override
286	public Stream<HistoryEvent> streamHistory(@Nonnull String workflowId, @Nullable String runId) {
287	Preconditions.checkNotNull(workflowId, "workflowId is required");	1✔
288
289	WorkflowExecution.Builder executionBuilder =
290	WorkflowExecution.newBuilder().setWorkflowId(workflowId);	1✔
291	if (runId != null) {	1✔
292	executionBuilder.setRunId(runId);	1✔
293	}
294	WorkflowExecution execution = executionBuilder.build();	1✔
295
296	return streamHistory(execution);	1✔
297	}
298
299	@Override
300	public WorkflowExecutionHistory fetchHistory(@Nonnull String workflowId) {
301	return fetchHistory(workflowId, null);	1✔
302	}
303
304	@Override
305	public WorkflowExecutionHistory fetchHistory(@Nonnull String workflowId, @Nullable String runId) {
306	Preconditions.checkNotNull(workflowId, "execution is required");	1✔
307
308	return new WorkflowExecutionHistory(	1✔
309	History.newBuilder()	1✔
310	.addAllEvents(streamHistory(workflowId, runId).collect(Collectors.toList()))	1✔
311	.build(),	1✔
312	workflowId);
313	}
314
315	public static WorkflowExecution start(Functions.Proc workflow) {
316	enforceNonWorkflowThread();	1✔
317	WorkflowInvocationHandler.initAsyncInvocation(InvocationType.START);	1✔
318	try {
319	workflow.apply();	1✔
320	return WorkflowInvocationHandler.getAsyncInvocationResult(WorkflowExecution.class);	1✔
321	} finally {
322	WorkflowInvocationHandler.closeAsyncInvocation();	1✔
323	}
324	}
325
326	public static <A1> WorkflowExecution start(Functions.Proc1<A1> workflow, A1 arg1) {
327	return start(() -> workflow.apply(arg1));	1✔
328	}
329
330	public static <A1, A2> WorkflowExecution start(
331	Functions.Proc2<A1, A2> workflow, A1 arg1, A2 arg2) {
332	return start(() -> workflow.apply(arg1, arg2));	1✔
333	}
334
335	public static <A1, A2, A3> WorkflowExecution start(
336	Functions.Proc3<A1, A2, A3> workflow, A1 arg1, A2 arg2, A3 arg3) {
337	return start(() -> workflow.apply(arg1, arg2, arg3));	1✔
338	}
339
340	public static <A1, A2, A3, A4> WorkflowExecution start(
341	Functions.Proc4<A1, A2, A3, A4> workflow, A1 arg1, A2 arg2, A3 arg3, A4 arg4) {
342	return start(() -> workflow.apply(arg1, arg2, arg3, arg4));	1✔
343	}
344
345	public static <A1, A2, A3, A4, A5> WorkflowExecution start(
346	Functions.Proc5<A1, A2, A3, A4, A5> workflow, A1 arg1, A2 arg2, A3 arg3, A4 arg4, A5 arg5) {
347	return start(() -> workflow.apply(arg1, arg2, arg3, arg4, arg5));	1✔
348	}
349
350	public static <A1, A2, A3, A4, A5, A6> WorkflowExecution start(
351	Functions.Proc6<A1, A2, A3, A4, A5, A6> workflow,
352	A1 arg1,
353	A2 arg2,
354	A3 arg3,
355	A4 arg4,
356	A5 arg5,
357	A6 arg6) {
358	return start(() -> workflow.apply(arg1, arg2, arg3, arg4, arg5, arg6));	1✔
359	}
360
361	public static <R> WorkflowExecution start(Functions.Func<R> workflow) {
362	return start((Functions.Proc) workflow::apply);	1✔
363	}
364
365	public static <A1, R> WorkflowExecution start(Functions.Func1<A1, R> workflow, A1 arg1) {
366	return start(() -> workflow.apply(arg1));	1✔
367	}
368
369	public static <A1, A2, R> WorkflowExecution start(
370	Functions.Func2<A1, A2, R> workflow, A1 arg1, A2 arg2) {
371	return start(() -> workflow.apply(arg1, arg2));	1✔
372	}
373
374	public static <A1, A2, A3, R> WorkflowExecution start(
375	Functions.Func3<A1, A2, A3, R> workflow, A1 arg1, A2 arg2, A3 arg3) {
376	return start(() -> workflow.apply(arg1, arg2, arg3));	1✔
377	}
378
379	public static <A1, A2, A3, A4, R> WorkflowExecution start(
380	Functions.Func4<A1, A2, A3, A4, R> workflow, A1 arg1, A2 arg2, A3 arg3, A4 arg4) {
381	return start(() -> workflow.apply(arg1, arg2, arg3, arg4));	1✔
382	}
383
384	public static <A1, A2, A3, A4, A5, R> WorkflowExecution start(
385	Functions.Func5<A1, A2, A3, A4, A5, R> workflow,
386	A1 arg1,
387	A2 arg2,
388	A3 arg3,
389	A4 arg4,
390	A5 arg5) {
391	return start(() -> workflow.apply(arg1, arg2, arg3, arg4, arg5));	1✔
392	}
393
394	public static <A1, A2, A3, A4, A5, A6, R> WorkflowExecution start(
395	Functions.Func6<A1, A2, A3, A4, A5, A6, R> workflow,
396	A1 arg1,
397	A2 arg2,
398	A3 arg3,
399	A4 arg4,
400	A5 arg5,
401	A6 arg6) {
402	return start(() -> workflow.apply(arg1, arg2, arg3, arg4, arg5, arg6));	1✔
403	}
404
405	@SuppressWarnings("unchecked")
406	public static CompletableFuture<Void> execute(Functions.Proc workflow) {
407	enforceNonWorkflowThread();	1✔
408	WorkflowInvocationHandler.initAsyncInvocation(InvocationType.EXECUTE);	1✔
409	try {
410	workflow.apply();	1✔
411	return WorkflowInvocationHandler.getAsyncInvocationResult(CompletableFuture.class);	1✔
412	} finally {
413	WorkflowInvocationHandler.closeAsyncInvocation();	1✔
414	}
415	}
416
417	public static <A1> CompletableFuture<Void> execute(Functions.Proc1<A1> workflow, A1 arg1) {
418	return execute(() -> workflow.apply(arg1));	1✔
419	}
420
421	public static <A1, A2> CompletableFuture<Void> execute(
422	Functions.Proc2<A1, A2> workflow, A1 arg1, A2 arg2) {
423	return execute(() -> workflow.apply(arg1, arg2));	1✔
424	}
425
426	public static <A1, A2, A3> CompletableFuture<Void> execute(
427	Functions.Proc3<A1, A2, A3> workflow, A1 arg1, A2 arg2, A3 arg3) {
428	return execute(() -> workflow.apply(arg1, arg2, arg3));	1✔
429	}
430
431	public static <A1, A2, A3, A4> CompletableFuture<Void> execute(
432	Functions.Proc4<A1, A2, A3, A4> workflow, A1 arg1, A2 arg2, A3 arg3, A4 arg4) {
433	return execute(() -> workflow.apply(arg1, arg2, arg3, arg4));	1✔
434	}
435
436	public static <A1, A2, A3, A4, A5> CompletableFuture<Void> execute(
437	Functions.Proc5<A1, A2, A3, A4, A5> workflow, A1 arg1, A2 arg2, A3 arg3, A4 arg4, A5 arg5) {
438	return execute(() -> workflow.apply(arg1, arg2, arg3, arg4, arg5));	1✔
439	}
440
441	public static <A1, A2, A3, A4, A5, A6> CompletableFuture<Void> execute(
442	Functions.Proc6<A1, A2, A3, A4, A5, A6> workflow,
443	A1 arg1,
444	A2 arg2,
445	A3 arg3,
446	A4 arg4,
447	A5 arg5,
448	A6 arg6) {
449	return execute(() -> workflow.apply(arg1, arg2, arg3, arg4, arg5, arg6));	1✔
450	}
451
452	@SuppressWarnings("unchecked")
453	public static <R> CompletableFuture<R> execute(Functions.Func<R> workflow) {
454	return (CompletableFuture<R>) execute((Functions.Proc) workflow::apply);	1✔
455	}
456
457	public static <A1, R> CompletableFuture<R> execute(Functions.Func1<A1, R> workflow, A1 arg1) {
458	return execute(() -> workflow.apply(arg1));	1✔
459	}
460
461	public static <A1, A2, R> CompletableFuture<R> execute(
462	Functions.Func2<A1, A2, R> workflow, A1 arg1, A2 arg2) {
463	return execute(() -> workflow.apply(arg1, arg2));	1✔
464	}
465
466	public static <A1, A2, A3, R> CompletableFuture<R> execute(
467	Functions.Func3<A1, A2, A3, R> workflow, A1 arg1, A2 arg2, A3 arg3) {
468	return execute(() -> workflow.apply(arg1, arg2, arg3));	1✔
469	}
470
471	public static <A1, A2, A3, A4, R> CompletableFuture<R> execute(
472	Functions.Func4<A1, A2, A3, A4, R> workflow, A1 arg1, A2 arg2, A3 arg3, A4 arg4) {
473	return execute(() -> workflow.apply(arg1, arg2, arg3, arg4));	1✔
474	}
475
476	public static <A1, A2, A3, A4, A5, R> CompletableFuture<R> execute(
477	Functions.Func5<A1, A2, A3, A4, A5, R> workflow,
478	A1 arg1,
479	A2 arg2,
480	A3 arg3,
481	A4 arg4,
482	A5 arg5) {
483	return execute(() -> workflow.apply(arg1, arg2, arg3, arg4, arg5));	1✔
484	}
485
486	public static <A1, A2, A3, A4, A5, A6, R> CompletableFuture<R> execute(
487	Functions.Func6<A1, A2, A3, A4, A5, A6, R> workflow,
488	A1 arg1,
489	A2 arg2,
490	A3 arg3,
491	A4 arg4,
492	A5 arg5,
493	A6 arg6) {
494	return execute(() -> workflow.apply(arg1, arg2, arg3, arg4, arg5, arg6));	1✔
495	}
496
497	Stream<HistoryEvent> streamHistory(WorkflowExecution execution) {
498	Preconditions.checkNotNull(execution, "execution is required");	1✔
499
500	GetWorkflowExecutionHistoryIterator iterator =	1✔
501	new GetWorkflowExecutionHistoryIterator(
502	options.getNamespace(), execution, null, genericClient);	1✔
503	iterator.init();	1✔
504
505	// IMMUTABLE here means that "interference" (in Java Streams terms) to this spliterator is
506	// impossible
507	final int CHARACTERISTICS =	1✔
508	Spliterator.ORDERED \| Spliterator.NONNULL \| Spliterator.DISTINCT \| Spliterator.IMMUTABLE;
509
510	return StreamSupport.stream(	1✔
511	Spliterators.spliteratorUnknownSize(iterator, CHARACTERISTICS), false);	1✔
512	}
513
514	@Override
515	public Object getInternal() {
516	return this;	1✔
517	}
518
519	@Override
520	public void registerWorkerFactory(WorkerFactory workerFactory) {
521	workerFactoryRegistry.register(workerFactory);	1✔
522	}	1✔
523
524	@Override
525	public void deregisterWorkerFactory(WorkerFactory workerFactory) {
526	workerFactoryRegistry.deregister(workerFactory);	1✔
527	}	1✔
528	}

temporalio / sdk-java / #153

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous