Skip to content

Commit 52dab8a

Browse files
committed
1 parent ef92eff commit 52dab8a

10 files changed

Lines changed: 160 additions & 156 deletions

‎src/main/java/com/uber/cadence/activity/Activity.java‎

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -234,25 +234,25 @@ public final class Activity {
234234
* for an "external system finished" signal containing the final result.
235235
*/
236236
public static void doNotCompleteOnReturn() {
237-
ActivityInternal.doNotCompleteOnReturn();
237+
getExecutionContext().doNotCompleteOnReturn();
238238
}
239239

240240
/**
241241
* @return task token that is required to report task completion when manual activity completion
242242
* is used.
243243
*/
244244
public static byte[] getTaskToken() {
245-
return ActivityInternal.getTask().getTaskToken();
245+
return getExecutionContext().getTaskToken();
246246
}
247247

248248
/** @return workfow execution that requested the activity execution */
249249
public static com.uber.cadence.WorkflowExecution getWorkflowExecution() {
250-
return ActivityInternal.getTask().getWorkflowExecution();
250+
return getExecutionContext().getWorkflowExecution();
251251
}
252252

253253
/** @return task that caused activity execution */
254254
public static ActivityTask getTask() {
255-
return ActivityInternal.getTask();
255+
return getExecutionContext().getInfo();
256256
}
257257

258258
/**
@@ -265,7 +265,7 @@ public static ActivityTask getTask() {
265265
* exception.
266266
*/
267267
public static <V> void heartbeat(V details) throws ActivityCompletionException {
268-
ActivityInternal.recordActivityHeartbeat(details);
268+
getExecutionContext().heartbeat(details);
269269
}
270270

271271
/**
@@ -281,7 +281,7 @@ public static <V> void heartbeat(V details) throws ActivityCompletionException {
281281
* @param detailsClass type of the heartbeat details
282282
*/
283283
public static <V> Optional<V> getHeartbeatDetails(Class<V> detailsClass) {
284-
return ActivityInternal.getHeartbeatDetails(detailsClass, detailsClass);
284+
return getExecutionContext().getHeartbeatDetails(detailsClass);
285285
}
286286

287287
/**
@@ -291,7 +291,7 @@ public static <V> Optional<V> getHeartbeatDetails(Class<V> detailsClass) {
291291
* @param detailsType type including generic information of the heartbeat details.
292292
*/
293293
public static <V> Optional<V> getHeartbeatDetails(Class<V> detailsClass, Type detailsType) {
294-
return ActivityInternal.getHeartbeatDetails(detailsClass, detailsType);
294+
return getExecutionContext().getHeartbeatDetails(detailsClass, detailsType);
295295
}
296296

297297
/**
@@ -301,19 +301,19 @@ public static <V> Optional<V> getHeartbeatDetails(Class<V> detailsClass, Type de
301301
* hide the service.
302302
*/
303303
public static IWorkflowService getService() {
304-
return ActivityInternal.getService();
304+
return getExecutionContext().getService();
305305
}
306306

307307
public static String getDomain() {
308-
return ActivityInternal.getDomain();
308+
return getExecutionContext().getDomain();
309309
}
310310

311311
/**
312312
* @return the activity execution context for the currently executing activity. This can be used
313313
* to access activity task metadata and use manual completion.
314314
*/
315-
public static com.uber.cadence.internal.sync.ActivityExecutionContext getExecutionContext() {
316-
return ActivityInternal.getContext();
315+
public static ActivityExecutionContext getExecutionContext() {
316+
return ActivityInternal.getExecutionContext();
317317
}
318318

319319
/**
Lines changed: 111 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,111 @@
1+
/*
2+
* Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
3+
*
4+
* Modifications copyright (C) 2017 Uber Technologies, Inc.
5+
*
6+
* Licensed under the Apache License, Version 2.0 (the "License"). You may not
7+
* use this file except in compliance with the License. A copy of the License is
8+
* located at
9+
*
10+
* http://aws.amazon.com/apache2.0
11+
*
12+
* or in the "license" file accompanying this file. This file is distributed on
13+
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
14+
* express or implied. See the License for the specific language governing
15+
* permissions and limitations under the License.
16+
*/
17+
18+
package com.uber.cadence.activity;
19+
20+
import com.uber.cadence.WorkflowExecution;
21+
import com.uber.cadence.client.ActivityCompletionClient;
22+
import com.uber.cadence.client.ActivityCompletionException;
23+
import com.uber.cadence.serviceclient.IWorkflowService;
24+
import java.lang.reflect.Type;
25+
import java.util.Optional;
26+
27+
/**
28+
* Context object passed to an activity implementation. Use {@link Activity#getExecutionContext()}
29+
* from an activity implementation to access.
30+
*/
31+
public interface ActivityExecutionContext {
32+
33+
/** Information about activity invocation and the caller workflow */
34+
ActivityTask getInfo();
35+
36+
/**
37+
* Use to notify Cadence service that activity execution is alive.
38+
*
39+
* @param details In case of activity timeout details are returned as a field of the exception
40+
* thrown.
41+
* @throws ActivityCompletionException Indicates that activity cancellation was requested by the
42+
* workflow or any other reason for activity to stop execution. Should be rethrown from
43+
* activity implementation to indicate successful cancellation.
44+
*/
45+
<V> void heartbeat(V details) throws ActivityCompletionException;
46+
47+
/**
48+
* Extracts heartbeat details from the last failed attempt. This is used in combination with retry
49+
* options. An activity could be scheduled with an optional {@link
50+
* com.uber.cadence.common.RetryOptions} on {@link ActivityOptions}. If an activity failed then
51+
* the server would attempt to dispatch another activity task to retry according to the retry
52+
* options. If there was heartbeat details reported by the activity from the failed attempt, the
53+
* details would be delivered along with the activity task for the retry attempt. The activity
54+
* could extract the details by {@link #getHeartbeatDetails(Class)}() and resume from the
55+
* progress.
56+
*
57+
* @param detailsClass type of the heartbeat details
58+
*/
59+
<V> Optional<V> getHeartbeatDetails(Class<V> detailsClass);
60+
61+
/**
62+
* Extracts heartbeat details from the last failed attempt. This is used in combination with retry
63+
* options. An activity could be scheduled with an optional {@link
64+
* com.uber.cadence.common.RetryOptions} on {@link ActivityOptions}. If an activity failed then
65+
* the server would attempt to dispatch another activity task to retry according to the retry
66+
* options. If there was heartbeat details reported by the activity from the failed attempt, the
67+
* details would be delivered along with the activity task for the retry attempt. The activity
68+
* could extract the details by {@link #getHeartbeatDetails(Class)}() and resume from the
69+
* progress.
70+
*
71+
* @param detailsClass type of the heartbeat details
72+
* @param detailsType type including generic information of the heartbeat details
73+
*/
74+
<V> Optional<V> getHeartbeatDetails(Class<V> detailsClass, Type detailsType);
75+
76+
/**
77+
* A correlation token that can be used to complete the activity asynchronously through {@link
78+
* com.uber.cadence.client.ActivityCompletionClient#complete(byte[], Object)}.
79+
*/
80+
byte[] getTaskToken();
81+
82+
/**
83+
* If this method is called during an activity execution then activity is not going to complete
84+
* when its method returns. It is expected to be completed asynchronously using {@link
85+
* com.uber.cadence.client.ActivityCompletionClient}.
86+
*/
87+
void doNotCompleteOnReturn();
88+
89+
boolean isDoNotCompleteOnReturn();
90+
91+
/**
92+
* Use for asynchronous completion within the activity worker process. Calling this method
93+
* prevents automatic activity completion on return and returns a client that can be used to
94+
* complete the activity asynchronously.
95+
*/
96+
ActivityCompletionClient useLocalManualCompletion();
97+
98+
boolean isUseLocalManualCompletion();
99+
100+
/** @return workfow execution that requested the activity execution */
101+
WorkflowExecution getWorkflowExecution();
102+
103+
/**
104+
* @return an instance of the Cadence service client that is the same used by the invoked activity
105+
* worker.
106+
*/
107+
IWorkflowService getService();
108+
109+
/** @return the domain of the activity execution */
110+
String getDomain();
111+
}

‎src/main/java/com/uber/cadence/internal/sync/ActivityExecutionContext.java‎

Lines changed: 0 additions & 101 deletions
This file was deleted.

‎src/main/java/com/uber/cadence/internal/sync/ActivityExecutionContextImpl.java‎

Lines changed: 14 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -42,12 +42,12 @@
4242
import org.slf4j.LoggerFactory;
4343

4444
/**
45-
* Base implementation of an {@link ActivityExecutionContext}.
45+
* Base implementation of an {@link com.uber.cadence.activity.ActivityExecutionContext}.
4646
*
4747
* @author fateev, suskin
48-
* @see ActivityExecutionContext
48+
* @see com.uber.cadence.activity.ActivityExecutionContext
4949
*/
50-
class ActivityExecutionContextImpl implements ActivityExecutionContext {
50+
class ActivityExecutionContextImpl implements com.uber.cadence.activity.ActivityExecutionContext {
5151

5252
private static final Logger log = LoggerFactory.getLogger(ActivityExecutionContextImpl.class);
5353
private static final long HEARTBEAT_RETRY_WAIT_MILLIS = 1000;
@@ -91,9 +91,13 @@ class ActivityExecutionContextImpl implements ActivityExecutionContext {
9191
new ManualActivityCompletionClientFactoryImpl(service, domain, dataConverter, metricsScope);
9292
}
9393

94-
/** @see ActivityExecutionContext#recordActivityHeartbeat(Object) */
9594
@Override
96-
public <V> void recordActivityHeartbeat(V details) throws ActivityCompletionException {
95+
public ActivityTask getInfo() {
96+
return task;
97+
}
98+
99+
@Override
100+
public <V> void heartbeat(V details) throws ActivityCompletionException {
97101
if (heartbeatExecutor.isShutdown()) {
98102
throw new ActivityWorkerShutdownException(task);
99103
}
@@ -114,6 +118,11 @@ public <V> void recordActivityHeartbeat(V details) throws ActivityCompletionExce
114118
}
115119
}
116120

121+
@Override
122+
public <V> Optional<V> getHeartbeatDetails(Class<V> detailsClass) {
123+
return getHeartbeatDetails(detailsClass, detailsClass);
124+
}
125+
117126
@Override
118127
public <V> Optional<V> getHeartbeatDetails(Class<V> detailsClass, Type detailsType) {
119128
lock.lock();
@@ -234,7 +243,6 @@ public ActivityCompletionClient useLocalManualCompletion() {
234243
}
235244

236245
/** @see ActivityExecutionContext#getTask() */
237-
@Override
238246
public ActivityTask getTask() {
239247
return task;
240248
}

‎src/main/java/com/uber/cadence/internal/sync/ActivityInternal.java‎

Lines changed: 2 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -17,40 +17,13 @@
1717

1818
package com.uber.cadence.internal.sync;
1919

20-
import com.uber.cadence.activity.ActivityTask;
21-
import com.uber.cadence.serviceclient.IWorkflowService;
22-
import java.lang.reflect.Type;
23-
import java.util.Optional;
20+
import com.uber.cadence.activity.ActivityExecutionContext;
2421

2522
public final class ActivityInternal {
2623

2724
private ActivityInternal() {}
2825

29-
public static ActivityExecutionContext getContext() {
26+
public static ActivityExecutionContext getExecutionContext() {
3027
return CurrentActivityExecutionContext.get();
3128
}
32-
33-
public static <V> void recordActivityHeartbeat(V details) {
34-
getContext().recordActivityHeartbeat(details);
35-
}
36-
37-
public static ActivityTask getTask() {
38-
return getContext().getTask();
39-
}
40-
41-
public static String getDomain() {
42-
return getContext().getDomain();
43-
}
44-
45-
public static IWorkflowService getService() {
46-
return getContext().getService();
47-
}
48-
49-
public static void doNotCompleteOnReturn() {
50-
getContext().doNotCompleteOnReturn();
51-
}
52-
53-
public static <V> Optional<V> getHeartbeatDetails(Class<V> detailsClass, Type detailsType) {
54-
return getContext().getHeartbeatDetails(detailsClass, detailsType);
55-
}
5629
}

‎src/main/java/com/uber/cadence/internal/sync/CurrentActivityExecutionContext.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,8 @@
1717

1818
package com.uber.cadence.internal.sync;
1919

20+
import com.uber.cadence.activity.ActivityExecutionContext;
21+
2022
/**
2123
* Thread local store of the context object passed to an activity implementation. Avoid using this
2224
* class directly.

0 commit comments

Comments
 (0)