Skip to content

Commit 8d2f35e

Browse files
Quinn-With-Two-NsQuinn-With-Two-Ns
authored
Add support for setting ActivityID (#3044)
Add support for setting Activity ID
1 parent b250876 commit 8d2f35e

15 files changed

Lines changed: 1290 additions & 30 deletions

‎temporal-sdk/src/main/java/io/temporal/common/interceptors/WorkflowOutboundCallsInterceptor.java‎

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -41,20 +41,37 @@ public interface WorkflowOutboundCallsInterceptor {
4141

4242
final class ActivityInput<R> {
4343
private final String activityName;
44+
private final @Nullable String activityId;
4445
private final Class<R> resultClass;
4546
private final Type resultType;
4647
private final Object[] args;
4748
private final ActivityOptions options;
4849
private final Header header;
4950

51+
/**
52+
* @deprecated Kept only for backward compatibility.
53+
*/
54+
@Deprecated
55+
public ActivityInput(
56+
String activityName,
57+
Class<R> resultClass,
58+
Type resultType,
59+
Object[] args,
60+
ActivityOptions options,
61+
Header header) {
62+
this(activityName, null, resultClass, resultType, args, options, header);
63+
}
64+
5065
public ActivityInput(
5166
String activityName,
67+
@Nullable String activityId,
5268
Class<R> resultClass,
5369
Type resultType,
5470
Object[] args,
5571
ActivityOptions options,
5672
Header header) {
5773
this.activityName = activityName;
74+
this.activityId = activityId;
5875
this.resultClass = resultClass;
5976
this.resultType = resultType;
6077
this.args = args;
@@ -66,6 +83,12 @@ public String getActivityName() {
6683
return activityName;
6784
}
6885

86+
/** Returns the caller-supplied Activity ID, or {@code null} if the SDK should generate one. */
87+
@Nullable
88+
public String getActivityId() {
89+
return activityId;
90+
}
91+
6992
public Class<R> getResultClass() {
7093
return resultClass;
7194
}
@@ -107,20 +130,37 @@ public Promise<R> getResult() {
107130

108131
final class LocalActivityInput<R> {
109132
private final String activityName;
133+
private final @Nullable String activityId;
110134
private final Class<R> resultClass;
111135
private final Type resultType;
112136
private final Object[] args;
113137
private final LocalActivityOptions options;
114138
private final Header header;
115139

140+
/**
141+
* @deprecated Kept only for backward compatibility.
142+
*/
143+
@Deprecated
144+
public LocalActivityInput(
145+
String activityName,
146+
Class<R> resultClass,
147+
Type resultType,
148+
Object[] args,
149+
LocalActivityOptions options,
150+
Header header) {
151+
this(activityName, null, resultClass, resultType, args, options, header);
152+
}
153+
116154
public LocalActivityInput(
117155
String activityName,
156+
@Nullable String activityId,
118157
Class<R> resultClass,
119158
Type resultType,
120159
Object[] args,
121160
LocalActivityOptions options,
122161
Header header) {
123162
this.activityName = activityName;
163+
this.activityId = activityId;
124164
this.resultClass = resultClass;
125165
this.resultType = resultType;
126166
this.args = args;
@@ -132,6 +172,12 @@ public String getActivityName() {
132172
return activityName;
133173
}
134174

175+
/** Returns the caller-supplied Activity ID, or {@code null} if the SDK should generate one. */
176+
@Nullable
177+
public String getActivityId() {
178+
return activityId;
179+
}
180+
135181
public Class<R> getResultClass() {
136182
return resultClass;
137183
}

‎temporal-sdk/src/main/java/io/temporal/internal/sync/ActivityInvocationHandler.java‎

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
import io.temporal.activity.ActivityOptions;
55
import io.temporal.common.MethodRetry;
66
import io.temporal.common.interceptors.WorkflowOutboundCallsInterceptor;
7+
import io.temporal.workflow.ActivityInvocationOptions;
78
import io.temporal.workflow.ActivityStub;
89
import io.temporal.workflow.Functions;
910
import java.lang.reflect.InvocationHandler;
@@ -46,7 +47,6 @@ private ActivityInvocationHandler(
4647
@Override
4748
protected Function<Object[], Object> getActivityFunc(
4849
Method method, MethodRetry methodRetry, String activityName) {
49-
Function<Object[], Object> function;
5050
ActivityOptions merged =
5151
ActivityOptions.newBuilder(options)
5252
.mergeActivityOptions(this.activityMethodOptions.get(activityName))
@@ -58,10 +58,15 @@ protected Function<Object[], Object> getActivityFunc(
5858
+ activityName
5959
+ " activity. Please set at least one of the above through the ActivityStub or WorkflowImplementationOptions.");
6060
}
61+
ActivityInvocationOptions invocationOptions = ActivityInvocationInternal.consumeOptions();
6162
ActivityStub stub = ActivityStubImpl.newInstance(merged, activityExecutor, assertReadOnly);
62-
function =
63-
(a) -> stub.execute(activityName, method.getReturnType(), method.getGenericReturnType(), a);
64-
return function;
63+
return (a) ->
64+
stub.execute(
65+
activityName,
66+
method.getReturnType(),
67+
method.getGenericReturnType(),
68+
invocationOptions,
69+
a);
6570
}
6671

6772
@Override

‎temporal-sdk/src/main/java/io/temporal/internal/sync/ActivityInvocationInternal.java‎

Lines changed: 112 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,112 @@
1+
package io.temporal.internal.sync;
2+
3+
import io.temporal.workflow.ActivityInvocationOptions;
4+
import io.temporal.workflow.Functions;
5+
import io.temporal.workflow.Promise;
6+
import java.util.Objects;
7+
8+
/** Applies options to one typed Activity proxy invocation. */
9+
final class ActivityInvocationInternal {
10+
11+
private static final ThreadLocal<State> invocation = new ThreadLocal<>();
12+
13+
private ActivityInvocationInternal() {}
14+
15+
static <R> R invoke(ActivityInvocationOptions options, Functions.Func<R> invocationFunction) {
16+
State state = startInvocation(options, false);
17+
try {
18+
R result = invocationFunction.apply();
19+
state.verifyConsumed();
20+
return result;
21+
} finally {
22+
invocation.remove();
23+
}
24+
}
25+
26+
static void invoke(ActivityInvocationOptions options, Functions.Proc invocationFunction) {
27+
State state = startInvocation(options, false);
28+
try {
29+
invocationFunction.apply();
30+
state.verifyConsumed();
31+
} finally {
32+
invocation.remove();
33+
}
34+
}
35+
36+
static <R> Promise<R> invokeAsync(
37+
ActivityInvocationOptions options, Functions.Proc invocationFunction) {
38+
State state = startInvocation(options, true);
39+
try {
40+
invocationFunction.apply();
41+
return state.getResult();
42+
} finally {
43+
invocation.remove();
44+
}
45+
}
46+
47+
private static State startInvocation(ActivityInvocationOptions options, boolean async) {
48+
if (invocation.get() != null) {
49+
throw new IllegalStateException("Already invoking an Activity with invocation options");
50+
}
51+
52+
State state = new State(Objects.requireNonNull(options, "options"), async);
53+
invocation.set(state);
54+
return state;
55+
}
56+
57+
static ActivityInvocationOptions consumeOptions() {
58+
State state = invocation.get();
59+
if (state == null) {
60+
return ActivityInvocationOptions.getDefaultInstance();
61+
}
62+
if (state.consumed) {
63+
throw new IllegalStateException("ActivityInvocationOptions can apply to only one invocation");
64+
}
65+
state.consumed = true;
66+
return state.options;
67+
}
68+
69+
static <R> boolean captureResult(Promise<R> result) {
70+
State state = invocation.get();
71+
if (state == null || !state.async) {
72+
return false;
73+
}
74+
if (state.result != null) {
75+
throw new IllegalStateException("ActivityInvocationOptions can apply to only one invocation");
76+
}
77+
state.result = Objects.requireNonNull(result, "result");
78+
return true;
79+
}
80+
81+
private static final class State {
82+
private final ActivityInvocationOptions options;
83+
private final boolean async;
84+
private boolean consumed;
85+
private Promise<?> result;
86+
87+
private State(ActivityInvocationOptions options, boolean async) {
88+
this.options = options;
89+
this.async = async;
90+
}
91+
92+
private void verifyConsumed() {
93+
if (!consumed) {
94+
throw invalidInvocation();
95+
}
96+
}
97+
98+
@SuppressWarnings("unchecked")
99+
private <R> Promise<R> getResult() {
100+
if (!consumed || result == null) {
101+
throw invalidInvocation();
102+
}
103+
return (Promise<R>) result;
104+
}
105+
106+
private IllegalArgumentException invalidInvocation() {
107+
return new IllegalArgumentException(
108+
"activityMethod must invoke an Activity stub created through Workflow.newActivityStub "
109+
+ "or Workflow.newLocalActivityStub");
110+
}
111+
}
112+
}

‎temporal-sdk/src/main/java/io/temporal/internal/sync/ActivityStubBase.java‎

Lines changed: 59 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -2,11 +2,12 @@
22

33
import com.google.common.base.Defaults;
44
import io.temporal.failure.ActivityFailure;
5+
import io.temporal.workflow.ActivityInvocationOptions;
56
import io.temporal.workflow.ActivityStub;
67
import io.temporal.workflow.Promise;
78
import java.lang.reflect.Type;
89

9-
/** Supports calling activity by name and arguments without its strongly typed interface. */
10+
/** Supports calling an activity by name and arguments without its strongly typed interface. */
1011
abstract class ActivityStubBase implements ActivityStub {
1112

1213
@Override
@@ -16,7 +17,63 @@ public <T> T execute(String activityName, Class<T> resultClass, Object... args)
1617

1718
@Override
1819
public <T> T execute(String activityName, Class<T> resultClass, Type resultType, Object... args) {
19-
Promise<T> result = executeAsync(activityName, resultClass, resultType, args);
20+
return getResult(executeAsync(activityName, resultClass, resultType, args), resultClass);
21+
}
22+
23+
@Override
24+
public <R> R execute(
25+
String activityName,
26+
Class<R> resultClass,
27+
ActivityInvocationOptions options,
28+
Object... args) {
29+
return execute(activityName, resultClass, resultClass, options, args);
30+
}
31+
32+
@Override
33+
public abstract <R> R execute(
34+
String activityName,
35+
Class<R> resultClass,
36+
Type resultType,
37+
ActivityInvocationOptions options,
38+
Object... args);
39+
40+
@Override
41+
public <R> Promise<R> executeAsync(String activityName, Class<R> resultClass, Object... args) {
42+
return executeAsync(activityName, resultClass, resultClass, args);
43+
}
44+
45+
@Override
46+
public <R> Promise<R> executeAsync(
47+
String activityName, Class<R> resultClass, Type resultType, Object... args) {
48+
return executeAsync(
49+
activityName,
50+
resultClass,
51+
resultType,
52+
ActivityInvocationOptions.getDefaultInstance(),
53+
args);
54+
}
55+
56+
@Override
57+
public <R> Promise<R> executeAsync(
58+
String activityName,
59+
Class<R> resultClass,
60+
ActivityInvocationOptions options,
61+
Object... args) {
62+
return executeAsync(activityName, resultClass, resultClass, options, args);
63+
}
64+
65+
@Override
66+
public abstract <R> Promise<R> executeAsync(
67+
String activityName,
68+
Class<R> resultClass,
69+
Type resultType,
70+
ActivityInvocationOptions options,
71+
Object... args);
72+
73+
protected <R> R getResult(Promise<R> result, Class<R> resultClass) {
74+
if (ActivityInvocationInternal.captureResult(result)) {
75+
return Defaults.defaultValue(resultClass);
76+
}
2077
if (AsyncInternal.isAsync()) {
2178
AsyncInternal.setAsyncResult(result);
2279
return Defaults.defaultValue(resultClass);
@@ -31,13 +88,4 @@ public <T> T execute(String activityName, Class<T> resultClass, Type resultType,
3188
throw e;
3289
}
3390
}
34-
35-
@Override
36-
public <R> Promise<R> executeAsync(String activityName, Class<R> resultClass, Object... args) {
37-
return executeAsync(activityName, resultClass, resultClass, args);
38-
}
39-
40-
@Override
41-
public abstract <R> Promise<R> executeAsync(
42-
String activityName, Class<R> resultClass, Type resultType, Object... args);
4391
}

0 commit comments

Comments
 (0)