#351

Committed 04 Aug 2026 11:58PM UTC coverage: 67.781% (-0.5%) from 68.245%

Build # #351

Build Type

push

github

Committed by

web-flow

Commit Message

Add dev server downloader & runner to testing package (#2982)

Coverage Stats

7311 of 12876 branches covered (56.78%)

Branch coverage included in aggregate %.

331 of 703 new or added lines in 12 files covered. (47.08%)

28 existing lines in 13 files now uncovered.

30207 of 42476 relevant lines covered (71.12%)

0.71 hits per line

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

79.65

/temporal-testing/src/main/java/io/temporal/testing/TestWorkflowEnvironmentInternal.java

package io.temporal.testing;

import com.google.common.base.Preconditions;
import com.google.common.collect.ObjectArrays;
import com.google.protobuf.ByteString;
import com.google.protobuf.Empty;
import com.uber.m3.tally.NoopScope;
import com.uber.m3.tally.Scope;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.temporal.api.common.v1.Payload;
import io.temporal.api.common.v1.WorkflowExecution;
import io.temporal.api.enums.v1.IndexedValueType;
import io.temporal.api.nexus.v1.Endpoint;
import io.temporal.api.nexus.v1.EndpointSpec;
import io.temporal.api.nexus.v1.EndpointTarget;
import io.temporal.api.operatorservice.v1.AddSearchAttributesRequest;
import io.temporal.api.operatorservice.v1.CreateNexusEndpointRequest;
import io.temporal.api.testservice.v1.SleepRequest;
import io.temporal.client.ActivityClient;
import io.temporal.client.ActivityClientOptions;
import io.temporal.client.WorkflowClient;
import io.temporal.client.WorkflowClientOptions;
import io.temporal.common.WorkflowExecutionHistory;
import io.temporal.common.interceptors.ActivityClientInterceptor;
import io.temporal.internal.common.ProtobufTimeUtils;
import io.temporal.internal.testservice.TestWorkflowService;
import io.temporal.serviceclient.*;
import io.temporal.testserver.TestServer;
import io.temporal.worker.Worker;
import io.temporal.worker.WorkerFactory;
import io.temporal.worker.WorkerOptions;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;

public final class TestWorkflowEnvironmentInternal implements TestWorkflowEnvironment {

  private final WorkflowClientOptions workflowClientOptions;
  private final ActivityClientOptions activityClientOptions;
  private final WorkflowServiceStubs workflowServiceStubs;
  private final OperatorServiceStubs operatorServiceStubs;
  private final @Nullable TestServiceStubs testServiceStubs;
  private final @Nullable TestServer.InProcessTestServer inProcessServer;
  private final @Nullable TestWorkflowService service;
  private final WorkerFactory workerFactory;
  private final @Nullable TimeLockingInterceptor timeLockingInterceptor;
  private final IdempotentTimeLocker constructorTimeLock;
  private final @Nullable TemporalDevServer ownedDevServer;

  public TestWorkflowEnvironmentInternal(@Nullable TestEnvironmentOptions testEnvironmentOptions) {
    this(testEnvironmentOptions, null);
  }

  TestWorkflowEnvironmentInternal(
      @Nullable TestEnvironmentOptions testEnvironmentOptions,
      @Nullable TemporalDevServer ownedDevServer) {
    this.ownedDevServer = ownedDevServer;
    if (testEnvironmentOptions == null) {
      testEnvironmentOptions = TestEnvironmentOptions.getDefaultInstance();
    }
    testEnvironmentOptions =
        TestEnvironmentOptions.newBuilder(testEnvironmentOptions).validateAndBuildWithDefaults();
    this.workflowClientOptions = testEnvironmentOptions.getWorkflowClientOptions();
    this.activityClientOptions = testEnvironmentOptions.getActivityClientOptions();

    WorkflowServiceStubsOptions.Builder stubsOptionsBuilder =
        testEnvironmentOptions.getWorkflowServiceStubsOptions() != null
            ? WorkflowServiceStubsOptions.newBuilder(
                testEnvironmentOptions.getWorkflowServiceStubsOptions())
            : WorkflowServiceStubsOptions.newBuilder();

    Scope metricsScope = testEnvironmentOptions.getMetricsScope();
    if (metricsScope != null && !(NoopScope.class.equals(metricsScope.getClass()))) {
      stubsOptionsBuilder = stubsOptionsBuilder.setMetricsScope(metricsScope);
    }

    if (testEnvironmentOptions.isUseExternalService()) {
      this.inProcessServer = null;
      this.service = null;
      this.workflowServiceStubs =
          WorkflowServiceStubs.newServiceStubs(
              stubsOptionsBuilder.setTarget(testEnvironmentOptions.getTarget()).build());
      this.testServiceStubs = null;
      this.timeLockingInterceptor = null;
      this.constructorTimeLock = null;
    } else {
      this.inProcessServer =
          TestServer.createServer(true, testEnvironmentOptions.getInitialTimeMillis());
      this.service = fetchWorkflowService();

      WorkflowServiceStubsOptions workflowServiceStubsOptions =
          stubsOptionsBuilder
              .setChannel(this.inProcessServer.getChannel())
              .setTarget(null)
              .validateAndBuildWithDefaults();
      this.workflowServiceStubs = WorkflowServiceStubs.newServiceStubs(workflowServiceStubsOptions);
      this.testServiceStubs =
          TestServiceStubs.newServiceStubs(
              TestServiceStubsOptions.newBuilder(workflowServiceStubsOptions)
                  // we don't want long calls to test service to throw with DEADLINE_EXCEEDED
                  .setRpcTimeout(Duration.ofMillis(Long.MAX_VALUE))
                  .validateAndBuildWithDefaults());
      this.timeLockingInterceptor = new TimeLockingInterceptor(this.testServiceStubs);

      if (!testEnvironmentOptions.isUseTimeskipping()) {
        // If the options ask for no timeskipping, lock one extra time. There will never be a
        // corresponding unlock, so timeskipping will always be off.
        this.constructorTimeLock = new IdempotentTimeLocker(this.testServiceStubs);
        this.constructorTimeLock.lockTimeSkipping();
      } else {
        this.constructorTimeLock = null;
      }
    }

    this.operatorServiceStubs =
        OperatorServiceStubs.newServiceStubs(
            OperatorServiceStubsOptions.newBuilder()
                .setChannel(workflowServiceStubs.getRawChannel())
                .validateAndBuildWithDefaults());

    WorkflowClient client =
        WorkflowClient.newInstance(this.workflowServiceStubs, this.workflowClientOptions);
    this.workerFactory =
        WorkerFactory.newInstance(client, testEnvironmentOptions.getWorkerFactoryOptions());

    testEnvironmentOptions.getSearchAttributes().forEach(this::registerSearchAttribute);
  }

  @SuppressWarnings("deprecation")
  private TestWorkflowService fetchWorkflowService() {
    return this.inProcessServer.getWorkflowService();
  }

  @Override
  public Worker newWorker(String taskQueue) {
    return workerFactory.newWorker(taskQueue, WorkerOptions.getDefaultInstance());
  }

  @Override
  public Worker newWorker(String taskQueue, WorkerOptions options) {
    return workerFactory.newWorker(taskQueue, options);
  }

  @Override
  public WorkflowClient getWorkflowClient() {
    WorkflowClientOptions options;
    if (timeLockingInterceptor != null) {
      options =
          WorkflowClientOptions.newBuilder(workflowClientOptions)
              .setInterceptors(
                  ObjectArrays.concat(
                      workflowClientOptions.getInterceptors(), timeLockingInterceptor))
              .build();
    } else {
      options = workflowClientOptions;
    }
    return WorkflowClient.newInstance(workflowServiceStubs, options);
  }

  @Override
  public ActivityClient getActivityClient() {
    ActivityClientOptions options;
    if (testServiceStubs != null) {
      List<ActivityClientInterceptor> interceptors =
          new ArrayList<>(activityClientOptions.getInterceptors());
      options =
          ActivityClientOptions.newBuilder(activityClientOptions)
              .setInterceptors(interceptors)
              .build();
    } else {
      options = activityClientOptions;
    }
    return ActivityClient.newInstance(workflowServiceStubs, options);
  }

  @Override
  public long currentTimeMillis() {
    if (testServiceStubs != null) {
      return ProtobufTimeUtils.toJavaInstant(
              testServiceStubs.blockingStub().getCurrentTime(Empty.newBuilder().build()).getTime())
          .toEpochMilli();
    } else {
      return System.currentTimeMillis();
    }
  }

  @Override
  public void sleep(Duration duration) {
    if (testServiceStubs != null) {
      testServiceStubs
          .blockingStub()
          .unlockTimeSkippingWithSleep(
              SleepRequest.newBuilder()
                  .setDuration(ProtobufTimeUtils.toProtoDuration(duration))
                  .build());
    } else {
      try {
        Thread.sleep(duration.toMillis());
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new RuntimeException(e);
      }
    }
  }

  @Override
  public void registerDelayedCallback(Duration delay, Runnable r) {
    Preconditions.checkState(
        service != null, "registerDelayedCallback is not supported with the external service");
    service.registerDelayedCallback(delay, r);
  }

  @Override
  public boolean registerSearchAttribute(String name, IndexedValueType type) {
    if (IndexedValueType.INDEXED_VALUE_TYPE_UNSPECIFIED.equals(type)) {
      throw new IllegalArgumentException(
          "Class " + type + " can't be used as a search attribute type");
    }
    AddSearchAttributesRequest request =
        AddSearchAttributesRequest.newBuilder()
            .setNamespace(getNamespace())
            .putSearchAttributes(name, type)
            .build();
    try {
      operatorServiceStubs.blockingStub().addSearchAttributes(request);
      return true;
    } catch (StatusRuntimeException e) {
      if (Status.Code.ALREADY_EXISTS.equals(e.getStatus().getCode())) {
        return false;
      }
      throw e;
    }
  }

  @Override
  public Endpoint createNexusEndpoint(String name, String taskQueue) {
    EndpointSpec spec =
        EndpointSpec.newBuilder()
            .setName(name)
            .setDescription(
                Payload.newBuilder()
                    .setData(
                        ByteString.copyFromUtf8(
                            "Test Nexus endpoint created by the Java SDK WorkflowTestEnvironment")))
            .setTarget(
                EndpointTarget.newBuilder()
                    .setWorker(
                        EndpointTarget.Worker.newBuilder()
                            .setNamespace(getNamespace())
                            .setTaskQueue(taskQueue)))
            .build();
    CreateNexusEndpointRequest request =
        CreateNexusEndpointRequest.newBuilder().setSpec(spec).build();
    return operatorServiceStubs.blockingStub().createNexusEndpoint(request).getEndpoint();
  }

  public void deleteNexusEndpoint(Endpoint endpoint) {
    operatorServiceStubs
        .blockingStub()
        .deleteNexusEndpoint(
            io.temporal.api.operatorservice.v1.DeleteNexusEndpointRequest.newBuilder()
                .setId(endpoint.getId())
                .setVersion(endpoint.getVersion())
                .build());
  }

  @Deprecated
  public WorkflowServiceStubs getWorkflowService() {
    return getWorkflowServiceStubs();
  }

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

  @Override
  public OperatorServiceStubs getOperatorServiceStubs() {
    return operatorServiceStubs;
  }

  @Override
  public String getNamespace() {
    return workflowClientOptions.getNamespace();
  }

  @Override
  public String getDiagnostics() {
    Preconditions.checkState(
        service != null, "getDiagnostics is not supported with the external service");
    StringBuilder result = new StringBuilder();
    service.getDiagnostics(result);
    return result.toString();
  }

  @Override
  @Deprecated
  public WorkflowExecutionHistory getWorkflowExecutionHistory(
      @Nonnull WorkflowExecution execution) {
    Preconditions.checkNotNull(execution, "execution is required");
    return getWorkflowClient().fetchHistory(execution.getWorkflowId(), execution.getRunId());
  }

  @Override
  public void close() {
    RuntimeException failure = null;
    try {
      if (testServiceStubs != null) {
        failure = runCleanup(failure, testServiceStubs::shutdownNow);
      }
      failure = runCleanup(failure, operatorServiceStubs::shutdownNow);
      failure = runCleanup(failure, workerFactory::shutdownNow);
      failure = runCleanup(failure, () -> workerFactory.awaitTermination(10, TimeUnit.SECONDS));
      if (constructorTimeLock != null) {
        failure = runCleanup(failure, constructorTimeLock::unlockTimeSkipping);
      }
      failure = runCleanup(failure, workflowServiceStubs::shutdownNow);
      if (testServiceStubs != null) {
        failure = runCleanup(failure, () -> testServiceStubs.awaitTermination(1, TimeUnit.SECONDS));
      }
      failure =
          runCleanup(failure, () -> operatorServiceStubs.awaitTermination(1, TimeUnit.SECONDS));
      failure =
          runCleanup(failure, () -> workflowServiceStubs.awaitTermination(1, TimeUnit.SECONDS));
      if (inProcessServer != null) {
        failure = runCleanup(failure, inProcessServer::close);
      }
    } finally {
      if (ownedDevServer != null) {
        failure = runCleanup(failure, ownedDevServer::close);
      }
    }
    if (failure != null) {
      throw failure;
    }
  }

  private static RuntimeException runCleanup(
      @Nullable RuntimeException previousFailure, @Nonnull Runnable cleanup) {
    try {
      cleanup.run();
    } catch (RuntimeException failure) {
      if (previousFailure == null) {
        return failure;
      }
      previousFailure.addSuppressed(failure);
    }
    return previousFailure;
  }

  @Override
  public void start() {
    workerFactory.start();
  }

  @Override
  public boolean isStarted() {
    return workerFactory.isStarted();
  }

  @Override
  public boolean isShutdown() {
    return workerFactory.isShutdown();
  }

  @Override
  public boolean isTerminated() {
    return workerFactory.isTerminated();
  }

  @Override
  @Deprecated
  public void shutdownTestService() {
    if (service != null) {
      service.close();
    }
  }

  @Override
  public void shutdown() {
    workerFactory.shutdown();
  }

  @Override
  public void shutdownNow() {
    workerFactory.shutdownNow();
  }

  @Override
  public void awaitTermination(long timeout, TimeUnit unit) {
    workerFactory.awaitTermination(timeout, unit);
  }

  @Override
  public WorkerFactory getWorkerFactory() {
    return workerFactory;
  }
}

1	package io.temporal.testing;
2
3	import com.google.common.base.Preconditions;
4	import com.google.common.collect.ObjectArrays;
5	import com.google.protobuf.ByteString;
6	import com.google.protobuf.Empty;
7	import com.uber.m3.tally.NoopScope;
8	import com.uber.m3.tally.Scope;
9	import io.grpc.Status;
10	import io.grpc.StatusRuntimeException;
11	import io.temporal.api.common.v1.Payload;
12	import io.temporal.api.common.v1.WorkflowExecution;
13	import io.temporal.api.enums.v1.IndexedValueType;
14	import io.temporal.api.nexus.v1.Endpoint;
15	import io.temporal.api.nexus.v1.EndpointSpec;
16	import io.temporal.api.nexus.v1.EndpointTarget;
17	import io.temporal.api.operatorservice.v1.AddSearchAttributesRequest;
18	import io.temporal.api.operatorservice.v1.CreateNexusEndpointRequest;
19	import io.temporal.api.testservice.v1.SleepRequest;
20	import io.temporal.client.ActivityClient;
21	import io.temporal.client.ActivityClientOptions;
22	import io.temporal.client.WorkflowClient;
23	import io.temporal.client.WorkflowClientOptions;
24	import io.temporal.common.WorkflowExecutionHistory;
25	import io.temporal.common.interceptors.ActivityClientInterceptor;
26	import io.temporal.internal.common.ProtobufTimeUtils;
27	import io.temporal.internal.testservice.TestWorkflowService;
28	import io.temporal.serviceclient.*;
29	import io.temporal.testserver.TestServer;
30	import io.temporal.worker.Worker;
31	import io.temporal.worker.WorkerFactory;
32	import io.temporal.worker.WorkerOptions;
33	import java.time.Duration;
34	import java.util.ArrayList;
35	import java.util.List;
36	import java.util.concurrent.TimeUnit;
37	import javax.annotation.Nonnull;
38	import javax.annotation.Nullable;
39
40	public final class TestWorkflowEnvironmentInternal implements TestWorkflowEnvironment {
41
42	private final WorkflowClientOptions workflowClientOptions;
43	private final ActivityClientOptions activityClientOptions;
44	private final WorkflowServiceStubs workflowServiceStubs;
45	private final OperatorServiceStubs operatorServiceStubs;
46	private final @Nullable TestServiceStubs testServiceStubs;
47	private final @Nullable TestServer.InProcessTestServer inProcessServer;
48	private final @Nullable TestWorkflowService service;
49	private final WorkerFactory workerFactory;
50	private final @Nullable TimeLockingInterceptor timeLockingInterceptor;
51	private final IdempotentTimeLocker constructorTimeLock;
52	private final @Nullable TemporalDevServer ownedDevServer;
53
54	public TestWorkflowEnvironmentInternal(@Nullable TestEnvironmentOptions testEnvironmentOptions) {
55	this(testEnvironmentOptions, null);	1✔
56	}	1✔
57
58	TestWorkflowEnvironmentInternal(
59	@Nullable TestEnvironmentOptions testEnvironmentOptions,
60	@Nullable TemporalDevServer ownedDevServer) {	1✔
61	this.ownedDevServer = ownedDevServer;	1✔
62	if (testEnvironmentOptions == null) {	1!
63	testEnvironmentOptions = TestEnvironmentOptions.getDefaultInstance();	×
64	}
65	testEnvironmentOptions =	1✔
66	TestEnvironmentOptions.newBuilder(testEnvironmentOptions).validateAndBuildWithDefaults();	1✔
67	this.workflowClientOptions = testEnvironmentOptions.getWorkflowClientOptions();	1✔
68	this.activityClientOptions = testEnvironmentOptions.getActivityClientOptions();	1✔
69
70	WorkflowServiceStubsOptions.Builder stubsOptionsBuilder =
71	testEnvironmentOptions.getWorkflowServiceStubsOptions() != null	1!
72	? WorkflowServiceStubsOptions.newBuilder(	1✔
73	testEnvironmentOptions.getWorkflowServiceStubsOptions())	1✔
74	: WorkflowServiceStubsOptions.newBuilder();	1✔
75
76	Scope metricsScope = testEnvironmentOptions.getMetricsScope();	1✔
77	if (metricsScope != null && !(NoopScope.class.equals(metricsScope.getClass()))) {	1!
78	stubsOptionsBuilder = stubsOptionsBuilder.setMetricsScope(metricsScope);	1✔
79	}
80
81	if (testEnvironmentOptions.isUseExternalService()) {	1✔
82	this.inProcessServer = null;	1✔
83	this.service = null;	1✔
84	this.workflowServiceStubs =	1✔
85	WorkflowServiceStubs.newServiceStubs(	1✔
86	stubsOptionsBuilder.setTarget(testEnvironmentOptions.getTarget()).build());	1✔
87	this.testServiceStubs = null;	1✔
88	this.timeLockingInterceptor = null;	1✔
89	this.constructorTimeLock = null;	1✔
90	} else {
91	this.inProcessServer =	1✔
92	TestServer.createServer(true, testEnvironmentOptions.getInitialTimeMillis());	1✔
93	this.service = fetchWorkflowService();	1✔
94
95	WorkflowServiceStubsOptions workflowServiceStubsOptions =	1✔
96	stubsOptionsBuilder
97	.setChannel(this.inProcessServer.getChannel())	1✔
98	.setTarget(null)	1✔
99	.validateAndBuildWithDefaults();	1✔
100	this.workflowServiceStubs = WorkflowServiceStubs.newServiceStubs(workflowServiceStubsOptions);	1✔
101	this.testServiceStubs =	1✔
102	TestServiceStubs.newServiceStubs(	1✔
103	TestServiceStubsOptions.newBuilder(workflowServiceStubsOptions)	1✔
104	// we don't want long calls to test service to throw with DEADLINE_EXCEEDED
105	.setRpcTimeout(Duration.ofMillis(Long.MAX_VALUE))	1✔
106	.validateAndBuildWithDefaults());	1✔
107	this.timeLockingInterceptor = new TimeLockingInterceptor(this.testServiceStubs);	1✔
108
109	if (!testEnvironmentOptions.isUseTimeskipping()) {	1✔
110	// If the options ask for no timeskipping, lock one extra time. There will never be a
111	// corresponding unlock, so timeskipping will always be off.
112	this.constructorTimeLock = new IdempotentTimeLocker(this.testServiceStubs);	1✔
113	this.constructorTimeLock.lockTimeSkipping();	1✔
114	} else {
115	this.constructorTimeLock = null;	1✔
116	}
117	}
118
119	this.operatorServiceStubs =	1✔
120	OperatorServiceStubs.newServiceStubs(	1✔
121	OperatorServiceStubsOptions.newBuilder()	1✔
122	.setChannel(workflowServiceStubs.getRawChannel())	1✔
123	.validateAndBuildWithDefaults());	1✔
124
125	WorkflowClient client =	1✔
126	WorkflowClient.newInstance(this.workflowServiceStubs, this.workflowClientOptions);	1✔
127	this.workerFactory =	1✔
128	WorkerFactory.newInstance(client, testEnvironmentOptions.getWorkerFactoryOptions());	1✔
129
130	testEnvironmentOptions.getSearchAttributes().forEach(this::registerSearchAttribute);	1✔
131	}	1✔
132
133	@SuppressWarnings("deprecation")
134	private TestWorkflowService fetchWorkflowService() {
135	return this.inProcessServer.getWorkflowService();	1✔
136	}
137
138	@Override
139	public Worker newWorker(String taskQueue) {
140	return workerFactory.newWorker(taskQueue, WorkerOptions.getDefaultInstance());	1✔
141	}
142
143	@Override
144	public Worker newWorker(String taskQueue, WorkerOptions options) {
145	return workerFactory.newWorker(taskQueue, options);	1✔
146	}
147
148	@Override
149	public WorkflowClient getWorkflowClient() {
150	WorkflowClientOptions options;
151	if (timeLockingInterceptor != null) {	1!
152	options =	1✔
153	WorkflowClientOptions.newBuilder(workflowClientOptions)	1✔
154	.setInterceptors(	1✔
155	ObjectArrays.concat(	1✔
156	workflowClientOptions.getInterceptors(), timeLockingInterceptor))	1✔
157	.build();	1✔
158	} else {
159	options = workflowClientOptions;	×
160	}
161	return WorkflowClient.newInstance(workflowServiceStubs, options);	1✔
162	}
163
164	@Override
165	public ActivityClient getActivityClient() {
166	ActivityClientOptions options;
167	if (testServiceStubs != null) {	1!
168	List<ActivityClientInterceptor> interceptors =	1✔
169	new ArrayList<>(activityClientOptions.getInterceptors());	1✔
170	options =	1✔
171	ActivityClientOptions.newBuilder(activityClientOptions)	1✔
172	.setInterceptors(interceptors)	1✔
173	.build();	1✔
174	} else {	1✔
175	options = activityClientOptions;	×
176	}
177	return ActivityClient.newInstance(workflowServiceStubs, options);	1✔
178	}
179
180	@Override
181	public long currentTimeMillis() {
182	if (testServiceStubs != null) {	1!
183	return ProtobufTimeUtils.toJavaInstant(	1✔
184	testServiceStubs.blockingStub().getCurrentTime(Empty.newBuilder().build()).getTime())	1✔
185	.toEpochMilli();	1✔
186	} else {
187	return System.currentTimeMillis();	×
188	}
189	}
190
191	@Override
192	public void sleep(Duration duration) {
193	if (testServiceStubs != null) {	1!
194	testServiceStubs	1✔
195	.blockingStub()	1✔
196	.unlockTimeSkippingWithSleep(	1✔
197	SleepRequest.newBuilder()	1✔
198	.setDuration(ProtobufTimeUtils.toProtoDuration(duration))	1✔
199	.build());	1✔
200	} else {
201	try {
202	Thread.sleep(duration.toMillis());	×
203	} catch (InterruptedException e) {	×
204	Thread.currentThread().interrupt();	×
205	throw new RuntimeException(e);	×
206	}	×
207	}
208	}	1✔
209
210	@Override
211	public void registerDelayedCallback(Duration delay, Runnable r) {
212	Preconditions.checkState(	1!
213	service != null, "registerDelayedCallback is not supported with the external service");
214	service.registerDelayedCallback(delay, r);	1✔
215	}	1✔
216
217	@Override
218	public boolean registerSearchAttribute(String name, IndexedValueType type) {
219	if (IndexedValueType.INDEXED_VALUE_TYPE_UNSPECIFIED.equals(type)) {	1!
220	throw new IllegalArgumentException(	×
221	"Class " + type + " can't be used as a search attribute type");
222	}
223	AddSearchAttributesRequest request =
224	AddSearchAttributesRequest.newBuilder()	1✔
225	.setNamespace(getNamespace())	1✔
226	.putSearchAttributes(name, type)	1✔
227	.build();	1✔
228	try {
229	operatorServiceStubs.blockingStub().addSearchAttributes(request);	1✔
230	return true;	1✔
231	} catch (StatusRuntimeException e) {	×
232	if (Status.Code.ALREADY_EXISTS.equals(e.getStatus().getCode())) {	×
233	return false;	×
234	}
235	throw e;	×
236	}
237	}
238
239	@Override
240	public Endpoint createNexusEndpoint(String name, String taskQueue) {
241	EndpointSpec spec =
242	EndpointSpec.newBuilder()	1✔
243	.setName(name)	1✔
244	.setDescription(	1✔
245	Payload.newBuilder()	1✔
246	.setData(	1✔
247	ByteString.copyFromUtf8(	1✔
248	"Test Nexus endpoint created by the Java SDK WorkflowTestEnvironment")))
249	.setTarget(	1✔
250	EndpointTarget.newBuilder()	1✔
251	.setWorker(	1✔
252	EndpointTarget.Worker.newBuilder()	1✔
253	.setNamespace(getNamespace())	1✔
254	.setTaskQueue(taskQueue)))	1✔
255	.build();	1✔
256	CreateNexusEndpointRequest request =
257	CreateNexusEndpointRequest.newBuilder().setSpec(spec).build();	1✔
258	return operatorServiceStubs.blockingStub().createNexusEndpoint(request).getEndpoint();	1✔
259	}
260
261	public void deleteNexusEndpoint(Endpoint endpoint) {
262	operatorServiceStubs	1✔
263	.blockingStub()	1✔
264	.deleteNexusEndpoint(	1✔
265	io.temporal.api.operatorservice.v1.DeleteNexusEndpointRequest.newBuilder()	1✔
266	.setId(endpoint.getId())	1✔
267	.setVersion(endpoint.getVersion())	1✔
268	.build());	1✔
269	}	1✔
270
271	@Deprecated
272	public WorkflowServiceStubs getWorkflowService() {
273	return getWorkflowServiceStubs();	×
274	}
275
276	@Override
277	public WorkflowServiceStubs getWorkflowServiceStubs() {
278	return workflowServiceStubs;	1✔
279	}
280
281	@Override
282	public OperatorServiceStubs getOperatorServiceStubs() {
283	return operatorServiceStubs;	1✔
284	}
285
286	@Override
287	public String getNamespace() {
288	return workflowClientOptions.getNamespace();	1✔
289	}
290
291	@Override
292	public String getDiagnostics() {
293	Preconditions.checkState(	×
294	service != null, "getDiagnostics is not supported with the external service");
295	StringBuilder result = new StringBuilder();	×
296	service.getDiagnostics(result);	×
297	return result.toString();	×
298	}
299
300	@Override
301	@Deprecated
302	public WorkflowExecutionHistory getWorkflowExecutionHistory(
303	@Nonnull WorkflowExecution execution) {
304	Preconditions.checkNotNull(execution, "execution is required");	×
305	return getWorkflowClient().fetchHistory(execution.getWorkflowId(), execution.getRunId());	×
306	}
307
308	@Override
309	public void close() {
310	RuntimeException failure = null;	1✔
311	try {
312	if (testServiceStubs != null) {	1✔
313	failure = runCleanup(failure, testServiceStubs::shutdownNow);	1✔
314	}
315	failure = runCleanup(failure, operatorServiceStubs::shutdownNow);	1✔
316	failure = runCleanup(failure, workerFactory::shutdownNow);	1✔
317	failure = runCleanup(failure, () -> workerFactory.awaitTermination(10, TimeUnit.SECONDS));	1✔
318	if (constructorTimeLock != null) {	1✔
319	failure = runCleanup(failure, constructorTimeLock::unlockTimeSkipping);	1✔
320	}
321	failure = runCleanup(failure, workflowServiceStubs::shutdownNow);	1✔
322	if (testServiceStubs != null) {	1✔
323	failure = runCleanup(failure, () -> testServiceStubs.awaitTermination(1, TimeUnit.SECONDS));	1✔
324	}
325	failure =	1✔
326	runCleanup(failure, () -> operatorServiceStubs.awaitTermination(1, TimeUnit.SECONDS));	1✔
327	failure =	1✔
328	runCleanup(failure, () -> workflowServiceStubs.awaitTermination(1, TimeUnit.SECONDS));	1✔
329	if (inProcessServer != null) {	1✔
330	failure = runCleanup(failure, inProcessServer::close);	1✔
331	}
332	} finally {
333	if (ownedDevServer != null) {	1!
NEW 334	failure = runCleanup(failure, ownedDevServer::close);	×
335	}
336	}
337	if (failure != null) {	1!
NEW 338	throw failure;	×
339	}
340	}	1✔
341
342	private static RuntimeException runCleanup(
343	@Nullable RuntimeException previousFailure, @Nonnull Runnable cleanup) {
344	try {
345	cleanup.run();	1✔
NEW 346	} catch (RuntimeException failure) {	×
NEW 347	if (previousFailure == null) {	×
NEW 348	return failure;	×
349	}
NEW 350	previousFailure.addSuppressed(failure);	×
351	}	1✔
352	return previousFailure;	1✔
353	}
354
355	@Override
356	public void start() {
357	workerFactory.start();	1✔
358	}	1✔
359
360	@Override
361	public boolean isStarted() {
362	return workerFactory.isStarted();	1✔
363	}
364
365	@Override
366	public boolean isShutdown() {
367	return workerFactory.isShutdown();	×
368	}
369
370	@Override
371	public boolean isTerminated() {
372	return workerFactory.isTerminated();	×
373	}
374
375	@Override
376	@Deprecated
377	public void shutdownTestService() {
378	if (service != null) {	1!
379	service.close();	1✔
380	}
381	}	1✔
382
383	@Override
384	public void shutdown() {
385	workerFactory.shutdown();	1✔
386	}	1✔
387
388	@Override
389	public void shutdownNow() {
390	workerFactory.shutdownNow();	1✔
391	}	1✔
392
393	@Override
394	public void awaitTermination(long timeout, TimeUnit unit) {
395	workerFactory.awaitTermination(timeout, unit);	1✔
396	}	1✔
397
398	@Override
399	public WorkerFactory getWorkerFactory() {
400	return workerFactory;	1✔
401	}
402	}

temporalio / sdk-java / #351

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