#253

Committed 21 May 2024 07:13PM UTC coverage: 77.429% (-0.09%) from 77.517%

Build # #253

Build Type

push

github

Committed by

web-flow

Commit Message

Don't return update handles until desired stage reached (#2066)

Run Details

45 of 54 new or added lines in 7 files covered. (83.33%)

17 existing lines in 5 files now uncovered.

19169 of 24757 relevant lines covered (77.43%)

0.77 hits per line

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

78.05

/temporal-sdk/src/main/java/io/temporal/client/LazyUpdateHandleImpl.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 io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.temporal.api.common.v1.WorkflowExecution;
import io.temporal.common.Experimental;
import io.temporal.common.interceptors.WorkflowClientCallsInterceptor;
import io.temporal.serviceclient.CheckedExceptionWrapper;
import java.lang.reflect.Type;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

@Experimental
final class LazyUpdateHandleImpl<T> implements UpdateHandle<T> {

  private final WorkflowClientCallsInterceptor workflowClientInvoker;
  private final String workflowType;
  private final String updateName;
  private final String id;
  private final WorkflowExecution execution;
  private final Class<T> resultClass;
  private final Type resultType;
  private WorkflowClientCallsInterceptor.PollWorkflowUpdateOutput<T> waitCompletedPollCall;

  LazyUpdateHandleImpl(
      WorkflowClientCallsInterceptor workflowClientInvoker,
      String workflowType,
      String updateName,
      String id,
      WorkflowExecution execution,
      Class<T> resultClass,
      Type resultType) {
    this.workflowClientInvoker = workflowClientInvoker;
    this.workflowType = workflowType;
    this.updateName = updateName;
    this.id = id;
    this.execution = execution;
    this.resultClass = resultClass;
    this.resultType = resultType;
  }

  @Override
  public WorkflowExecution getExecution() {
    return execution;
  }

  @Override
  public String getId() {
    return id;
  }

  @Override
  public CompletableFuture<T> getResultAsync(long timeout, TimeUnit unit) {

    WorkflowClientCallsInterceptor.PollWorkflowUpdateOutput<T> pollCall = null;
    boolean setFromWaitCompleted = false;

    // If waitCompleted was called, use the result from that call.
    synchronized (this) {
      if (waitCompletedPollCall != null) {
        pollCall = waitCompletedPollCall;
        waitCompletedPollCall = null;
      }
    }

    if (!setFromWaitCompleted) {
      pollCall = pollUntilComplete(timeout, unit);
    }

    return pollCall
        .getResult()
        .exceptionally(
            failure -> {
              if (failure instanceof CompletionException) {
                // unwrap the CompletionException
                failure = ((Throwable) failure).getCause();
              }
              failure = CheckedExceptionWrapper.unwrap((Throwable) failure);
              if (failure instanceof Error) {
                throw (Error) failure;
              }
              if (failure instanceof StatusRuntimeException) {
                StatusRuntimeException sre = (StatusRuntimeException) failure;
                if (Status.Code.NOT_FOUND.equals(sre.getStatus().getCode())) {
                  // Currently no way to tell if the NOT_FOUND was because the workflow ID
                  // does not exist or because the update ID does not exist.
                  throw sre;
                }
              } else if (failure instanceof WorkflowException) {
                throw (WorkflowException) failure;
              } else if (failure instanceof TimeoutException) {
                throw new CompletionException((TimeoutException) failure);
              }
              throw new WorkflowServiceException(execution, workflowType, (Throwable) failure);
            });
  }

  @Override
  public CompletableFuture<T> getResultAsync() {
    return this.getResultAsync(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
  }

  // Can be called immediately after initialization to wait for the update to be completed, but
  // still have the result be returned by getResultAsync.
  void waitCompleted() {
    waitCompletedPollCall = pollUntilComplete(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
  }

  private WorkflowClientCallsInterceptor.PollWorkflowUpdateOutput<T> pollUntilComplete(
      long timeout, TimeUnit unit) {
    return workflowClientInvoker.pollWorkflowUpdate(
        new WorkflowClientCallsInterceptor.PollWorkflowUpdateInput<>(
            execution, updateName, id, resultClass, resultType, timeout, unit));
  }
}

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 io.grpc.Status;
24	import io.grpc.StatusRuntimeException;
25	import io.temporal.api.common.v1.WorkflowExecution;
26	import io.temporal.common.Experimental;
27	import io.temporal.common.interceptors.WorkflowClientCallsInterceptor;
28	import io.temporal.serviceclient.CheckedExceptionWrapper;
29	import java.lang.reflect.Type;
30	import java.util.concurrent.CompletableFuture;
31	import java.util.concurrent.CompletionException;
32	import java.util.concurrent.TimeUnit;
33	import java.util.concurrent.TimeoutException;
34
35	@Experimental
36	final class LazyUpdateHandleImpl<T> implements UpdateHandle<T> {
37
38	private final WorkflowClientCallsInterceptor workflowClientInvoker;
39	private final String workflowType;
40	private final String updateName;
41	private final String id;
42	private final WorkflowExecution execution;
43	private final Class<T> resultClass;
44	private final Type resultType;
45	private WorkflowClientCallsInterceptor.PollWorkflowUpdateOutput<T> waitCompletedPollCall;
46
47	LazyUpdateHandleImpl(
48	WorkflowClientCallsInterceptor workflowClientInvoker,
49	String workflowType,
50	String updateName,
51	String id,
52	WorkflowExecution execution,
53	Class<T> resultClass,
54	Type resultType) {	1✔
55	this.workflowClientInvoker = workflowClientInvoker;	1✔
56	this.workflowType = workflowType;	1✔
57	this.updateName = updateName;	1✔
58	this.id = id;	1✔
59	this.execution = execution;	1✔
60	this.resultClass = resultClass;	1✔
61	this.resultType = resultType;	1✔
62	}	1✔
63
64	@Override
65	public WorkflowExecution getExecution() {
66	return execution;	×
67	}
68
69	@Override
70	public String getId() {
71	return id;	×
72	}
73
74	@Override
75	public CompletableFuture<T> getResultAsync(long timeout, TimeUnit unit) {
76
77	WorkflowClientCallsInterceptor.PollWorkflowUpdateOutput<T> pollCall = null;	1✔
78	boolean setFromWaitCompleted = false;	1✔
79
80	// If waitCompleted was called, use the result from that call.
81	synchronized (this) {	1✔
82	if (waitCompletedPollCall != null) {	1✔
NEW 83	pollCall = waitCompletedPollCall;	×
NEW 84	waitCompletedPollCall = null;	×
85	}
86	}	1✔
87
88	if (!setFromWaitCompleted) {	1✔
89	pollCall = pollUntilComplete(timeout, unit);	1✔
90	}
91
92	return pollCall	1✔
93	.getResult()	1✔
94	.exceptionally(	1✔
95	failure -> {
96	if (failure instanceof CompletionException) {	1✔
97	// unwrap the CompletionException
98	failure = ((Throwable) failure).getCause();	1✔
99	}
100	failure = CheckedExceptionWrapper.unwrap((Throwable) failure);	1✔
101	if (failure instanceof Error) {	1✔
102	throw (Error) failure;	×
103	}
104	if (failure instanceof StatusRuntimeException) {	1✔
105	StatusRuntimeException sre = (StatusRuntimeException) failure;	1✔
106	if (Status.Code.NOT_FOUND.equals(sre.getStatus().getCode())) {	1✔
107	// Currently no way to tell if the NOT_FOUND was because the workflow ID
108	// does not exist or because the update ID does not exist.
109	throw sre;	1✔
110	}
111	} else if (failure instanceof WorkflowException) {	1✔
112	throw (WorkflowException) failure;	×
113	} else if (failure instanceof TimeoutException) {	1✔
114	throw new CompletionException((TimeoutException) failure);	1✔
115	}
116	throw new WorkflowServiceException(execution, workflowType, (Throwable) failure);	×
117	});
118	}
119
120	@Override
121	public CompletableFuture<T> getResultAsync() {
122	return this.getResultAsync(Long.MAX_VALUE, TimeUnit.MILLISECONDS);	1✔
123	}
124
125	// Can be called immediately after initialization to wait for the update to be completed, but
126	// still have the result be returned by getResultAsync.
127	void waitCompleted() {
NEW 128	waitCompletedPollCall = pollUntilComplete(Long.MAX_VALUE, TimeUnit.MILLISECONDS);	×
NEW 129	}	×
130
131	private WorkflowClientCallsInterceptor.PollWorkflowUpdateOutput<T> pollUntilComplete(
132	long timeout, TimeUnit unit) {
133	return workflowClientInvoker.pollWorkflowUpdate(	1✔
134	new WorkflowClientCallsInterceptor.PollWorkflowUpdateInput<>(
135	execution, updateName, id, resultClass, resultType, timeout, unit));
136	}
137	}

temporalio / sdk-java / #253

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