#284

Committed 23 Jul 2024 10:09PM UTC coverage: 77.304% (-0.06%) from 77.364%

Build # #284

Build Type

push

github

Committed by

web-flow

Commit Message

Reintroduce slot supplier & add many tests (#2143)

Run Details

593 of 752 new or added lines in 37 files covered. (78.86%)

22 existing lines in 10 files now uncovered.

19554 of 25295 relevant lines covered (77.3%)

0.77 hits per line

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

85.71

/temporal-sdk/src/main/java/io/temporal/worker/WorkflowTaskDispatchHandle.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.worker;

import com.google.common.base.Preconditions;
import io.temporal.api.workflowservice.v1.PollWorkflowTaskQueueResponse;
import io.temporal.internal.worker.TrackingSlotSupplier;
import io.temporal.internal.worker.WorkflowTask;
import io.temporal.worker.tuning.SlotPermit;
import io.temporal.worker.tuning.SlotReleaseReason;
import io.temporal.worker.tuning.WorkflowSlotInfo;
import java.io.Closeable;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
import javax.annotation.Nonnull;

public class WorkflowTaskDispatchHandle implements Closeable {
  private final AtomicBoolean completed = new AtomicBoolean();
  private final Function<WorkflowTask, Boolean> dispatchCallback;
  private final TrackingSlotSupplier<WorkflowSlotInfo> slotSupplier;
  private final SlotPermit permit;

  /**
   * @param dispatchCallback callback into a {@code WorkflowWorker} to dispatch a workflow task.
   * @param slotSupplier slot supplier that was used to reserve a slot for this workflow task
   */
  public WorkflowTaskDispatchHandle(
      DispatchCallback dispatchCallback,
      TrackingSlotSupplier<WorkflowSlotInfo> slotSupplier,
      SlotPermit permit) {
    this.dispatchCallback = dispatchCallback;
    this.slotSupplier = slotSupplier;
    this.permit = permit;
  }

  /**
   * @param workflowTask to be fed directly into the workflow worker
   * @return true is the workflow task was successfully dispatched
   * @throws IllegalArgumentException if the workflow task doesn't belong to the task queue of the
   *     worker provided this {@link WorkflowTaskDispatchHandle}
   */
  public boolean dispatch(@Nonnull PollWorkflowTaskQueueResponse workflowTask) {
    Preconditions.checkNotNull(workflowTask, "workflowTask");
    if (completed.compareAndSet(false, true)) {
      return dispatchCallback.apply(
          new WorkflowTask(workflowTask, (rr) -> slotSupplier.releaseSlot(rr, permit)));
    } else {
      return false;
    }
  }

  @Override
  public void close() {
    if (completed.compareAndSet(false, true)) {
      slotSupplier.releaseSlot(SlotReleaseReason.neverUsed(), permit);
    }
  }

  /** A callback into a {@code WorkflowWorker} to dispatch a workflow task */
  @FunctionalInterface
  public interface DispatchCallback extends Function<WorkflowTask, Boolean> {

    /**
     * Should dispatch the Workflow Task to the Workflow Worker. Shouldn't block the thread.
     *
     * @param workflowTask WorkflowTask to be dispatched
     * @return true if the dispatch was successful and false otherwise
     * @throws IllegalArgumentException if {@code workflowTask} doesn't belong to the task queue of the Worker that provided the {@link WorkflowTaskDispatchHandle
     */
    @Override
    Boolean apply(WorkflowTask workflowTask) throws IllegalArgumentException;
  }
}

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.worker;
22
23	import com.google.common.base.Preconditions;
24	import io.temporal.api.workflowservice.v1.PollWorkflowTaskQueueResponse;
25	import io.temporal.internal.worker.TrackingSlotSupplier;
26	import io.temporal.internal.worker.WorkflowTask;
27	import io.temporal.worker.tuning.SlotPermit;
28	import io.temporal.worker.tuning.SlotReleaseReason;
29	import io.temporal.worker.tuning.WorkflowSlotInfo;
30	import java.io.Closeable;
31	import java.util.concurrent.atomic.AtomicBoolean;
32	import java.util.function.Function;
33	import javax.annotation.Nonnull;
34
35	public class WorkflowTaskDispatchHandle implements Closeable {
36	private final AtomicBoolean completed = new AtomicBoolean();	1✔
37	private final Function<WorkflowTask, Boolean> dispatchCallback;
38	private final TrackingSlotSupplier<WorkflowSlotInfo> slotSupplier;
39	private final SlotPermit permit;
40
41	/**
42	* @param dispatchCallback callback into a {@code WorkflowWorker} to dispatch a workflow task.
43	* @param slotSupplier slot supplier that was used to reserve a slot for this workflow task
44	*/
45	public WorkflowTaskDispatchHandle(
46	DispatchCallback dispatchCallback,
47	TrackingSlotSupplier<WorkflowSlotInfo> slotSupplier,
48	SlotPermit permit) {	1✔
49	this.dispatchCallback = dispatchCallback;	1✔
50	this.slotSupplier = slotSupplier;	1✔
51	this.permit = permit;	1✔
52	}	1✔
53
54	/**
55	* @param workflowTask to be fed directly into the workflow worker
56	* @return true is the workflow task was successfully dispatched
57	* @throws IllegalArgumentException if the workflow task doesn't belong to the task queue of the
58	* worker provided this {@link WorkflowTaskDispatchHandle}
59	*/
60	public boolean dispatch(@Nonnull PollWorkflowTaskQueueResponse workflowTask) {
61	Preconditions.checkNotNull(workflowTask, "workflowTask");	1✔
62	if (completed.compareAndSet(false, true)) {	1✔
63	return dispatchCallback.apply(	1✔
64	new WorkflowTask(workflowTask, (rr) -> slotSupplier.releaseSlot(rr, permit)));	1✔
65	} else {
66	return false;	×
67	}
68	}
69
70	@Override
71	public void close() {
72	if (completed.compareAndSet(false, true)) {	1✔
NEW 73	slotSupplier.releaseSlot(SlotReleaseReason.neverUsed(), permit);	×
74	}
75	}	1✔
76
77	/** A callback into a {@code WorkflowWorker} to dispatch a workflow task */
78	@FunctionalInterface
79	public interface DispatchCallback extends Function<WorkflowTask, Boolean> {
80
81	/**
82	* Should dispatch the Workflow Task to the Workflow Worker. Shouldn't block the thread.
83	*
84	* @param workflowTask WorkflowTask to be dispatched
85	* @return true if the dispatch was successful and false otherwise
86	* @throws IllegalArgumentException if {@code workflowTask} doesn't belong to the task queue of the Worker that provided the {@link WorkflowTaskDispatchHandle
87	*/
88	@Override
89	Boolean apply(WorkflowTask workflowTask) throws IllegalArgumentException;
90	}
91	}

temporalio / sdk-java / #284

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