#343

Committed 31 Oct 2024 06:31PM UTC coverage: 75.148% (-3.6%) from 78.794%

Build # #343

Build Type

push

github

Committed by

web-flow

Commit Message

Fix jacoco coverage (#2304)

Run Details

5139 of 8240 branches covered (62.37%)

Branch coverage included in aggregate %.

22841 of 28993 relevant lines covered (78.78%)

0.79 hits per line

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

88.89

/temporal-test-server/src/main/java/io/temporal/internal/testservice/RequestContext.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.internal.testservice;

import com.google.common.base.MoreObjects;
import com.google.protobuf.Timestamp;
import com.google.protobuf.util.Timestamps;
import io.grpc.Status;
import io.temporal.api.common.v1.WorkflowExecution;
import io.temporal.api.history.v1.HistoryEvent;
import io.temporal.internal.common.WorkflowExecutionUtils;
import io.temporal.internal.testservice.TestWorkflowStore.ActivityTask;
import io.temporal.internal.testservice.TestWorkflowStore.WorkflowTask;
import io.temporal.workflow.Functions;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.function.LongSupplier;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;

final class RequestContext {

  @FunctionalInterface
  interface CommitCallback {

    void apply(int historySize);
  }

  static final class Timer {

    private final Duration delay;
    private final Runnable callback;
    private final String taskInfo;
    private Functions.Proc cancellationHandle;
    private final Functions.Proc cancellationHandleWrapper;

    Timer(Duration delay, Runnable callback, String taskInfo) {
      this.delay = delay;
      this.callback = callback;
      this.taskInfo = taskInfo;
      this.cancellationHandleWrapper =
          () -> {
            if (this.cancellationHandle != null) {
              this.cancellationHandle.apply();
            }
          };
    }

    Duration getDelay() {
      return delay;
    }

    Runnable getCallback() {
      return callback;
    }

    String getTaskInfo() {
      return taskInfo;
    }

    public void setCancellationHandle(Functions.Proc cancellationHandle) {
      this.cancellationHandle = cancellationHandle;
    }

    public Functions.Proc getCancellationHandle() {
      return cancellationHandleWrapper;
    }
  }

  static final class TimerLockChange {
    private final String caller;

    /** +1 or -1 */
    private final int change;

    TimerLockChange(String caller, int change) {
      this.caller = Objects.requireNonNull(caller);
      if (change != -1 && change != 1) {
        throw new IllegalArgumentException("Invalid change: " + change);
      }
      this.change = change;
    }

    public String getCaller() {
      return caller;
    }

    public int getChange() {
      return change;
    }
  }

  private final LongSupplier clock;

  private final ExecutionId executionId;

  private final TestWorkflowMutableState workflowMutableState;

  private final long initialEventId;

  private final List<HistoryEvent> events = new ArrayList<>();
  private final List<CommitCallback> commitCallbacks = new ArrayList<>();
  // Contains a workflow task created by the updater that needs to be persisted into a task queue on
  // a commit.
  // If an eager dispatch was performed, it should be reset to null
  private WorkflowTask workflowTaskForMatching;
  private final List<ActivityTask> activityTasks = new ArrayList<>();
  private final List<TestWorkflowStore.NexusTask> nexusTasks = new ArrayList<>();
  private final List<Timer> timers = new ArrayList<>();
  private long workflowCompletedAtEventId = -1;
  private boolean needWorkflowTask;
  // How many times call SelfAdvancedTimer#lockTimeSkipping.
  // Negative means how many times to call SelfAdvancedTimer#unlockTimeSkipping.
  private final List<TimerLockChange> timerLocks = new ArrayList<>();

  // Contains an exception that may be published by the updater in case if updater needs to perform
  // and commit the changes.
  // The updater can't just throw the exception because it will prevent the changes to be committed.
  // This exception should be thrown at the very end after performing all the commit actions
  private RuntimeException exception;

  /**
   * Creates an instance of the RequestContext
   *
   * @param clock clock used to timestamp events and schedule timers.
   * @param workflowMutableState state of the execution being updated
   * @param initialEventId expected id of the next event added to the history
   */
  RequestContext(
      LongSupplier clock, TestWorkflowMutableState workflowMutableState, long initialEventId) {
    this.clock = Objects.requireNonNull(clock);
    this.workflowMutableState = Objects.requireNonNull(workflowMutableState);
    this.executionId = Objects.requireNonNull(workflowMutableState.getExecutionId());
    if (initialEventId <= 0) {
      throw new IllegalArgumentException("Invalid initialEventId: " + initialEventId);
    }
    this.initialEventId = initialEventId;
  }

  void add(RequestContext ctx) {
    this.activityTasks.addAll(ctx.getActivityTasks());
    this.nexusTasks.addAll(ctx.getNexusTasks());
    this.timers.addAll(ctx.getTimers());
    this.events.addAll(ctx.getEvents());
  }

  void lockTimer(String caller) {
    timerLocks.add(new TimerLockChange(caller, +1));
  }

  void unlockTimer(String caller) {
    timerLocks.add(new TimerLockChange(caller, -1));
  }

  List<TimerLockChange> getTimerLocks() {
    return timerLocks;
  }

  void clearTimersAndLocks() {
    timerLocks.clear();
    timers.clear();
  }

  Timestamp currentTime() {
    return Timestamps.fromMillis(clock.getAsLong());
  }

  /** Returns eventId of the added event; */
  long addEvent(HistoryEvent event) {
    if (workflowMutableState.isTerminalState()) {
      throw Status.NOT_FOUND
          .withDescription("workflow execution already completed")
          .asRuntimeException();
    }
    long eventId = initialEventId + events.size();
    if (WorkflowExecutionUtils.isWorkflowExecutionClosedEvent(event)) {
      workflowCompletedAtEventId = eventId;
    } else {
      if (workflowCompletedAtEventId > 0 && workflowCompletedAtEventId < eventId) {
        throw new IllegalStateException("Event added after the workflow completion event");
      }
    }
    events.add(event);
    return eventId;
  }

  WorkflowExecution getExecution() {
    return executionId.getExecution();
  }

  public TestWorkflowMutableState getWorkflowMutableState() {
    return workflowMutableState;
  }

  String getNamespace() {
    return executionId.getNamespace();
  }

  public long getInitialEventId() {
    return initialEventId;
  }

  public long getNextEventId() {
    return initialEventId + events.size();
  }

  /**
   * Command needed, but there is one already running. So initiate another one as soon as it
   * completes.
   */
  void setNeedWorkflowTask(boolean needWorkflowTask) {
    this.needWorkflowTask = needWorkflowTask;
  }

  boolean isNeedWorkflowTask() {
    return needWorkflowTask;
  }

  void setWorkflowTaskForMatching(@Nonnull WorkflowTask workflowTaskForMatching) {
    this.workflowTaskForMatching = Objects.requireNonNull(workflowTaskForMatching);
  }

  @Nullable
  WorkflowTask resetWorkflowTaskForMatching() {
    WorkflowTask existingTask = this.workflowTaskForMatching;
    this.workflowTaskForMatching = null;
    return existingTask;
  }

  WorkflowTask getWorkflowTaskForMatching() {
    return workflowTaskForMatching;
  }

  void addActivityTask(ActivityTask activityTask) {
    this.activityTasks.add(activityTask);
  }

  void addNexusTask(TestWorkflowStore.NexusTask nexusTask) {
    this.nexusTasks.add(nexusTask);
  }

  /**
   * @return cancellation handle
   */
  Functions.Proc addTimer(Duration delay, Runnable callback, String name) {
    Timer timer = new Timer(delay, callback, name);
    this.timers.add(timer);
    return timer.getCancellationHandle();
  }

  public List<Timer> getTimers() {
    return timers;
  }

  List<ActivityTask> getActivityTasks() {
    return activityTasks;
  }

  List<TestWorkflowStore.NexusTask> getNexusTasks() {
    return nexusTasks;
  }

  List<HistoryEvent> getEvents() {
    return events;
  }

  void onCommit(CommitCallback callback) {
    commitCallbacks.add(callback);
  }

  /**
   * @return nextEventId
   */
  long commitChanges(TestWorkflowStore store) {
    return store.save(this);
  }

  /** Called by {@link TestWorkflowStore#save(RequestContext)} */
  void fireCallbacks(int historySize) {
    for (CommitCallback callback : commitCallbacks) {
      callback.apply(historySize);
    }
  }

  ExecutionId getExecutionId() {
    return executionId;
  }

  public RuntimeException getException() {
    return exception;
  }

  public void setExceptionIfEmpty(RuntimeException exception) {
    this.exception = MoreObjects.firstNonNull(this.exception, exception);
  }
}

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.internal.testservice;
22
23	import com.google.common.base.MoreObjects;
24	import com.google.protobuf.Timestamp;
25	import com.google.protobuf.util.Timestamps;
26	import io.grpc.Status;
27	import io.temporal.api.common.v1.WorkflowExecution;
28	import io.temporal.api.history.v1.HistoryEvent;
29	import io.temporal.internal.common.WorkflowExecutionUtils;
30	import io.temporal.internal.testservice.TestWorkflowStore.ActivityTask;
31	import io.temporal.internal.testservice.TestWorkflowStore.WorkflowTask;
32	import io.temporal.workflow.Functions;
33	import java.time.Duration;
34	import java.util.ArrayList;
35	import java.util.List;
36	import java.util.Objects;
37	import java.util.function.LongSupplier;
38	import javax.annotation.Nonnull;
39	import javax.annotation.Nullable;
40
41	final class RequestContext {
42
43	@FunctionalInterface
44	interface CommitCallback {
45
46	void apply(int historySize);
47	}
48
49	static final class Timer {
50
51	private final Duration delay;
52	private final Runnable callback;
53	private final String taskInfo;
54	private Functions.Proc cancellationHandle;
55	private final Functions.Proc cancellationHandleWrapper;
56
57	Timer(Duration delay, Runnable callback, String taskInfo) {	1✔
58	this.delay = delay;	1✔
59	this.callback = callback;	1✔
60	this.taskInfo = taskInfo;	1✔
61	this.cancellationHandleWrapper =	1✔
62	() -> {
63	if (this.cancellationHandle != null) {	1!
64	this.cancellationHandle.apply();	1✔
65	}
66	};	1✔
67	}	1✔
68
69	Duration getDelay() {
70	return delay;	1✔
71	}
72
73	Runnable getCallback() {
74	return callback;	1✔
75	}
76
77	String getTaskInfo() {
78	return taskInfo;	1✔
79	}
80
81	public void setCancellationHandle(Functions.Proc cancellationHandle) {
82	this.cancellationHandle = cancellationHandle;	1✔
83	}	1✔
84
85	public Functions.Proc getCancellationHandle() {
86	return cancellationHandleWrapper;	1✔
87	}
88	}
89
90	static final class TimerLockChange {
91	private final String caller;
92
93	/** +1 or -1 */
94	private final int change;
95
96	TimerLockChange(String caller, int change) {	1✔
97	this.caller = Objects.requireNonNull(caller);	1✔
98	if (change != -1 && change != 1) {	1!
99	throw new IllegalArgumentException("Invalid change: " + change);	×
100	}
101	this.change = change;	1✔
102	}	1✔
103
104	public String getCaller() {
105	return caller;	1✔
106	}
107
108	public int getChange() {
109	return change;	1✔
110	}
111	}
112
113	private final LongSupplier clock;
114
115	private final ExecutionId executionId;
116
117	private final TestWorkflowMutableState workflowMutableState;
118
119	private final long initialEventId;
120
121	private final List<HistoryEvent> events = new ArrayList<>();	1✔
122	private final List<CommitCallback> commitCallbacks = new ArrayList<>();	1✔
123	// Contains a workflow task created by the updater that needs to be persisted into a task queue on
124	// a commit.
125	// If an eager dispatch was performed, it should be reset to null
126	private WorkflowTask workflowTaskForMatching;
127	private final List<ActivityTask> activityTasks = new ArrayList<>();	1✔
128	private final List<TestWorkflowStore.NexusTask> nexusTasks = new ArrayList<>();	1✔
129	private final List<Timer> timers = new ArrayList<>();	1✔
130	private long workflowCompletedAtEventId = -1;	1✔
131	private boolean needWorkflowTask;
132	// How many times call SelfAdvancedTimer#lockTimeSkipping.
133	// Negative means how many times to call SelfAdvancedTimer#unlockTimeSkipping.
134	private final List<TimerLockChange> timerLocks = new ArrayList<>();	1✔
135
136	// Contains an exception that may be published by the updater in case if updater needs to perform
137	// and commit the changes.
138	// The updater can't just throw the exception because it will prevent the changes to be committed.
139	// This exception should be thrown at the very end after performing all the commit actions
140	private RuntimeException exception;
141
142	/**
143	* Creates an instance of the RequestContext
144	*
145	* @param clock clock used to timestamp events and schedule timers.
146	* @param workflowMutableState state of the execution being updated
147	* @param initialEventId expected id of the next event added to the history
148	*/
149	RequestContext(
150	LongSupplier clock, TestWorkflowMutableState workflowMutableState, long initialEventId) {	1✔
151	this.clock = Objects.requireNonNull(clock);	1✔
152	this.workflowMutableState = Objects.requireNonNull(workflowMutableState);	1✔
153	this.executionId = Objects.requireNonNull(workflowMutableState.getExecutionId());	1✔
154	if (initialEventId <= 0) {	1!
155	throw new IllegalArgumentException("Invalid initialEventId: " + initialEventId);	×
156	}
157	this.initialEventId = initialEventId;	1✔
158	}	1✔
159
160	void add(RequestContext ctx) {
161	this.activityTasks.addAll(ctx.getActivityTasks());	1✔
162	this.nexusTasks.addAll(ctx.getNexusTasks());	1✔
163	this.timers.addAll(ctx.getTimers());	1✔
164	this.events.addAll(ctx.getEvents());	1✔
165	}	1✔
166
167	void lockTimer(String caller) {
168	timerLocks.add(new TimerLockChange(caller, +1));	1✔
169	}	1✔
170
171	void unlockTimer(String caller) {
172	timerLocks.add(new TimerLockChange(caller, -1));	1✔
173	}	1✔
174
175	List<TimerLockChange> getTimerLocks() {
176	return timerLocks;	1✔
177	}
178
179	void clearTimersAndLocks() {
180	timerLocks.clear();	1✔
181	timers.clear();	1✔
182	}	1✔
183
184	Timestamp currentTime() {
185	return Timestamps.fromMillis(clock.getAsLong());	1✔
186	}
187
188	/** Returns eventId of the added event; */
189	long addEvent(HistoryEvent event) {
190	if (workflowMutableState.isTerminalState()) {	1!
191	throw Status.NOT_FOUND	×
192	.withDescription("workflow execution already completed")	×
193	.asRuntimeException();	×
194	}
195	long eventId = initialEventId + events.size();	1✔
196	if (WorkflowExecutionUtils.isWorkflowExecutionClosedEvent(event)) {	1✔
197	workflowCompletedAtEventId = eventId;	1✔
198	} else {
199	if (workflowCompletedAtEventId > 0 && workflowCompletedAtEventId < eventId) {	1!
200	throw new IllegalStateException("Event added after the workflow completion event");	×
201	}
202	}
203	events.add(event);	1✔
204	return eventId;	1✔
205	}
206
207	WorkflowExecution getExecution() {
208	return executionId.getExecution();	1✔
209	}
210
211	public TestWorkflowMutableState getWorkflowMutableState() {
212	return workflowMutableState;	1✔
213	}
214
215	String getNamespace() {
216	return executionId.getNamespace();	1✔
217	}
218
219	public long getInitialEventId() {
220	return initialEventId;	1✔
221	}
222
223	public long getNextEventId() {
224	return initialEventId + events.size();	1✔
225	}
226
227	/**
228	* Command needed, but there is one already running. So initiate another one as soon as it
229	* completes.
230	*/
231	void setNeedWorkflowTask(boolean needWorkflowTask) {
232	this.needWorkflowTask = needWorkflowTask;	1✔
233	}	1✔
234
235	boolean isNeedWorkflowTask() {
236	return needWorkflowTask;	1✔
237	}
238
239	void setWorkflowTaskForMatching(@Nonnull WorkflowTask workflowTaskForMatching) {
240	this.workflowTaskForMatching = Objects.requireNonNull(workflowTaskForMatching);	1✔
241	}	1✔
242
243	@Nullable
244	WorkflowTask resetWorkflowTaskForMatching() {
245	WorkflowTask existingTask = this.workflowTaskForMatching;	1✔
246	this.workflowTaskForMatching = null;	1✔
247	return existingTask;	1✔
248	}
249
250	WorkflowTask getWorkflowTaskForMatching() {
251	return workflowTaskForMatching;	1✔
252	}
253
254	void addActivityTask(ActivityTask activityTask) {
255	this.activityTasks.add(activityTask);	1✔
256	}	1✔
257
258	void addNexusTask(TestWorkflowStore.NexusTask nexusTask) {
259	this.nexusTasks.add(nexusTask);	1✔
260	}	1✔
261
262	/**
263	* @return cancellation handle
264	*/
265	Functions.Proc addTimer(Duration delay, Runnable callback, String name) {
266	Timer timer = new Timer(delay, callback, name);	1✔
267	this.timers.add(timer);	1✔
268	return timer.getCancellationHandle();	1✔
269	}
270
271	public List<Timer> getTimers() {
272	return timers;	1✔
273	}
274
275	List<ActivityTask> getActivityTasks() {
276	return activityTasks;	1✔
277	}
278
279	List<TestWorkflowStore.NexusTask> getNexusTasks() {
280	return nexusTasks;	1✔
281	}
282
283	List<HistoryEvent> getEvents() {
284	return events;	1✔
285	}
286
287	void onCommit(CommitCallback callback) {
288	commitCallbacks.add(callback);	1✔
289	}	1✔
290
291	/**
292	* @return nextEventId
293	*/
294	long commitChanges(TestWorkflowStore store) {
295	return store.save(this);	1✔
296	}
297
298	/** Called by {@link TestWorkflowStore#save(RequestContext)} */
299	void fireCallbacks(int historySize) {
300	for (CommitCallback callback : commitCallbacks) {	1✔
301	callback.apply(historySize);	1✔
302	}	1✔
303	}	1✔
304
305	ExecutionId getExecutionId() {
306	return executionId;	1✔
307	}
308
309	public RuntimeException getException() {
310	return exception;	1✔
311	}
312
313	public void setExceptionIfEmpty(RuntimeException exception) {
314	this.exception = MoreObjects.firstNonNull(this.exception, exception);	1✔
315	}	1✔
316	}

temporalio / sdk-java / #343

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