Skip to content

Commit f9a9cad

Browse files
Quinn-With-Two-NsQuinn-With-Two-Ns
authored
Rename WorkflowClientOperationHandlers->WorkflowRunOperation (#2405)
1 parent 32fbf02 commit f9a9cad

14 files changed

+155
-157
lines changed

‎temporal-opentracing/src/test/java/io/temporal/opentracing/NexusOperationTest.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,7 @@
3535
import io.temporal.client.WorkflowClientOptions;
3636
import io.temporal.client.WorkflowOptions;
3737
import io.temporal.nexus.Nexus;
38-
import io.temporal.nexus.WorkflowClientOperationHandlers;
38+
import io.temporal.nexus.WorkflowRunOperation;
3939
import io.temporal.testing.internal.SDKTestWorkflowRule;
4040
import io.temporal.worker.WorkerFactoryOptions;
4141
import io.temporal.workflow.*;
@@ -83,7 +83,7 @@ public interface TestNexusService {
8383
public class TestNexusServiceImpl {
8484
@OperationImpl
8585
public OperationHandler<String, String> operation() {
86-
return WorkflowClientOperationHandlers.fromWorkflowMethod(
86+
return WorkflowRunOperation.fromWorkflowMethod(
8787
(context, details, input) ->
8888
Nexus.getOperationContext()
8989
.getWorkflowClient()

‎temporal-sdk/src/main/java/io/temporal/nexus/WorkflowClientOperationHandlers.java

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

‎temporal-sdk/src/main/java/io/temporal/nexus/WorkflowHandleFactory.java

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -26,9 +26,8 @@
2626
import javax.annotation.Nullable;
2727

2828
/**
29-
* Function interface for {@link
30-
* WorkflowClientOperationHandlers#fromWorkflowHandle(WorkflowHandleFactory)} representing the
31-
* workflow to associate with each operation call.
29+
* Function interface for {@link WorkflowRunOperation#fromWorkflowHandle(WorkflowHandleFactory)}
30+
* representing the workflow to associate with each operation call.
3231
*/
3332
@FunctionalInterface
3433
public interface WorkflowHandleFactory<T, R> {

‎temporal-sdk/src/main/java/io/temporal/nexus/WorkflowMethodFactory.java

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -28,16 +28,15 @@
2828
import javax.annotation.Nullable;
2929

3030
/**
31-
* Function interface for {@link
32-
* WorkflowClientOperationHandlers#fromWorkflowMethod(WorkflowMethodFactory)} representing the
33-
* workflow method to invoke for every operation call.
31+
* Function interface for {@link WorkflowRunOperation#fromWorkflowMethod(WorkflowMethodFactory)}
32+
* representing the workflow method to invoke for every operation call.
3433
*/
3534
@FunctionalInterface
3635
public interface WorkflowMethodFactory<T, R> {
3736
/**
3837
* Invoked every operation start call and expected to return a workflow method reference to a
3938
* proxy created through {@link WorkflowClient#newWorkflowStub(Class, WorkflowOptions)} using the
40-
* provided {@link WorkflowClient}.
39+
* provided {@link WorkflowClient} form {@link Nexus#getOperationContext()}.
4140
*/
4241
@Nullable
4342
Functions.Func1<T, R> apply(OperationContext context, OperationStartDetails details, T input);

‎temporal-sdk/src/main/java/io/temporal/nexus/WorkflowRunOperation.java

Lines changed: 27 additions & 76 deletions
Original file line numberDiff line numberDiff line change
@@ -20,86 +20,37 @@
2020

2121
package io.temporal.nexus;
2222

23-
import static io.temporal.internal.common.LinkConverter.workflowEventToNexusLink;
24-
import static io.temporal.internal.common.NexusUtil.nexusProtoLinkToLink;
25-
26-
import io.nexusrpc.OperationInfo;
2723
import io.nexusrpc.handler.*;
2824
import io.nexusrpc.handler.OperationHandler;
29-
import io.temporal.api.common.v1.Link;
30-
import io.temporal.api.common.v1.WorkflowExecution;
31-
import io.temporal.api.enums.v1.EventType;
32-
import io.temporal.client.WorkflowClient;
33-
import io.temporal.internal.client.NexusStartWorkflowRequest;
34-
import io.temporal.internal.nexus.CurrentNexusOperationContext;
35-
import io.temporal.internal.nexus.InternalNexusOperationContext;
36-
import java.net.URISyntaxException;
37-
38-
class RunWorkflowOperation<T, R> implements OperationHandler<T, R> {
39-
private final WorkflowHandleFactory<T, R> handleFactory;
40-
41-
RunWorkflowOperation(WorkflowHandleFactory<T, R> handleFactory) {
42-
this.handleFactory = handleFactory;
43-
}
44-
45-
@Override
46-
public OperationStartResult<R> start(
47-
OperationContext ctx, OperationStartDetails operationStartDetails, T input) {
48-
InternalNexusOperationContext nexusCtx = CurrentNexusOperationContext.get();
49-
50-
WorkflowHandle handle = handleFactory.apply(ctx, operationStartDetails, input);
51-
52-
NexusStartWorkflowRequest nexusRequest =
53-
new NexusStartWorkflowRequest(
54-
operationStartDetails.getRequestId(),
55-
operationStartDetails.getCallbackUrl(),
56-
operationStartDetails.getCallbackHeaders(),
57-
nexusCtx.getTaskQueue(),
58-
operationStartDetails.getLinks());
59-
60-
WorkflowExecution workflowExec = handle.getInvoker().invoke(nexusRequest);
61-
62-
// Create the link information about the new workflow and return to the caller.
63-
Link.WorkflowEvent workflowEventLink =
64-
Link.WorkflowEvent.newBuilder()
65-
.setNamespace(nexusCtx.getNamespace())
66-
.setWorkflowId(workflowExec.getWorkflowId())
67-
.setRunId(workflowExec.getRunId())
68-
.setEventRef(
69-
Link.WorkflowEvent.EventReference.newBuilder()
70-
.setEventType(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED))
71-
.build();
72-
io.temporal.api.nexus.v1.Link nexusLink = workflowEventToNexusLink(workflowEventLink);
73-
try {
74-
OperationStartResult.Builder<R> result =
75-
OperationStartResult.newAsyncBuilder(workflowExec.getWorkflowId());
76-
if (nexusLink != null) {
77-
result.addLink(nexusProtoLinkToLink(nexusLink));
78-
}
79-
return result.build();
80-
} catch (URISyntaxException e) {
81-
// Not expected as the link is constructed by the SDK.
82-
throw new OperationHandlerException(
83-
OperationHandlerException.ErrorType.INTERNAL, "failed to construct result URL", e);
84-
}
25+
import io.temporal.common.Experimental;
26+
27+
/** WorkflowRunOperation can be used to map a workflow run to a Nexus operation */
28+
@Experimental
29+
public final class WorkflowRunOperation {
30+
/**
31+
* Maps a workflow method to an {@link io.nexusrpc.handler.OperationHandler}.
32+
*
33+
* @param startMethod returns the workflow method reference to call
34+
* @return Operation handler to be used as an {@link OperationImpl}
35+
*/
36+
public static <T, R> OperationHandler<T, R> fromWorkflowMethod(
37+
WorkflowMethodFactory<T, R> startMethod) {
38+
return new WorkflowRunOperationImpl<>(
39+
(OperationContext context, OperationStartDetails details, T input) ->
40+
WorkflowHandle.fromWorkflowMethod(startMethod.apply(context, details, input), input));
8541
}
8642

87-
@Override
88-
public R fetchResult(
89-
OperationContext operationContext, OperationFetchResultDetails operationFetchResultDetails) {
90-
throw new UnsupportedOperationException("Not implemented");
43+
/**
44+
* Maps a workflow handle to an {@link io.nexusrpc.handler.OperationHandler}.
45+
*
46+
* @param handleFactory returns the workflow handle that will be mapped to the call
47+
* @return Operation handler to be used as an {@link OperationImpl}
48+
*/
49+
public static <T, R> OperationHandler<T, R> fromWorkflowHandle(
50+
WorkflowHandleFactory<T, R> handleFactory) {
51+
return new WorkflowRunOperationImpl<>(handleFactory);
9152
}
9253

93-
@Override
94-
public OperationInfo fetchInfo(
95-
OperationContext operationContext, OperationFetchInfoDetails operationFetchInfoDetails) {
96-
throw new UnsupportedOperationException("Not implemented");
97-
}
98-
99-
@Override
100-
public void cancel(
101-
OperationContext operationContext, OperationCancelDetails operationCancelDetails) {
102-
WorkflowClient client = CurrentNexusOperationContext.get().getWorkflowClient();
103-
client.newUntypedWorkflowStub(operationCancelDetails.getOperationId()).cancel();
104-
}
54+
/** Prohibits instantiation. */
55+
private WorkflowRunOperation() {}
10556
}

‎temporal-sdk/src/main/java/io/temporal/nexus/WorkflowRunOperationImpl.java

Lines changed: 105 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,105 @@
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.nexus;
22+
23+
import static io.temporal.internal.common.LinkConverter.workflowEventToNexusLink;
24+
import static io.temporal.internal.common.NexusUtil.nexusProtoLinkToLink;
25+
26+
import io.nexusrpc.OperationInfo;
27+
import io.nexusrpc.handler.*;
28+
import io.nexusrpc.handler.OperationHandler;
29+
import io.temporal.api.common.v1.Link;
30+
import io.temporal.api.common.v1.WorkflowExecution;
31+
import io.temporal.api.enums.v1.EventType;
32+
import io.temporal.client.WorkflowClient;
33+
import io.temporal.internal.client.NexusStartWorkflowRequest;
34+
import io.temporal.internal.nexus.CurrentNexusOperationContext;
35+
import io.temporal.internal.nexus.InternalNexusOperationContext;
36+
import java.net.URISyntaxException;
37+
38+
class WorkflowRunOperationImpl<T, R> implements OperationHandler<T, R> {
39+
private final WorkflowHandleFactory<T, R> handleFactory;
40+
41+
WorkflowRunOperationImpl(WorkflowHandleFactory<T, R> handleFactory) {
42+
this.handleFactory = handleFactory;
43+
}
44+
45+
@Override
46+
public OperationStartResult<R> start(
47+
OperationContext ctx, OperationStartDetails operationStartDetails, T input) {
48+
InternalNexusOperationContext nexusCtx = CurrentNexusOperationContext.get();
49+
50+
WorkflowHandle handle = handleFactory.apply(ctx, operationStartDetails, input);
51+
52+
NexusStartWorkflowRequest nexusRequest =
53+
new NexusStartWorkflowRequest(
54+
operationStartDetails.getRequestId(),
55+
operationStartDetails.getCallbackUrl(),
56+
operationStartDetails.getCallbackHeaders(),
57+
nexusCtx.getTaskQueue(),
58+
operationStartDetails.getLinks());
59+
60+
WorkflowExecution workflowExec = handle.getInvoker().invoke(nexusRequest);
61+
62+
// Create the link information about the new workflow and return to the caller.
63+
Link.WorkflowEvent workflowEventLink =
64+
Link.WorkflowEvent.newBuilder()
65+
.setNamespace(nexusCtx.getNamespace())
66+
.setWorkflowId(workflowExec.getWorkflowId())
67+
.setRunId(workflowExec.getRunId())
68+
.setEventRef(
69+
Link.WorkflowEvent.EventReference.newBuilder()
70+
.setEventType(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED))
71+
.build();
72+
io.temporal.api.nexus.v1.Link nexusLink = workflowEventToNexusLink(workflowEventLink);
73+
try {
74+
OperationStartResult.Builder<R> result =
75+
OperationStartResult.newAsyncBuilder(workflowExec.getWorkflowId());
76+
if (nexusLink != null) {
77+
result.addLink(nexusProtoLinkToLink(nexusLink));
78+
}
79+
return result.build();
80+
} catch (URISyntaxException e) {
81+
// Not expected as the link is constructed by the SDK.
82+
throw new OperationHandlerException(
83+
OperationHandlerException.ErrorType.INTERNAL, "failed to construct result URL", e);
84+
}
85+
}
86+
87+
@Override
88+
public R fetchResult(
89+
OperationContext operationContext, OperationFetchResultDetails operationFetchResultDetails) {
90+
throw new UnsupportedOperationException("Not implemented");
91+
}
92+
93+
@Override
94+
public OperationInfo fetchInfo(
95+
OperationContext operationContext, OperationFetchInfoDetails operationFetchInfoDetails) {
96+
throw new UnsupportedOperationException("Not implemented");
97+
}
98+
99+
@Override
100+
public void cancel(
101+
OperationContext operationContext, OperationCancelDetails operationCancelDetails) {
102+
WorkflowClient client = CurrentNexusOperationContext.get().getWorkflowClient();
103+
client.newUntypedWorkflowStub(operationCancelDetails.getOperationId()).cancel();
104+
}
105+
}

‎temporal-sdk/src/test/java/io/temporal/workflow/nexus/AsyncWorkflowOperationTest.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@
2525
import io.nexusrpc.handler.ServiceImpl;
2626
import io.temporal.client.WorkflowOptions;
2727
import io.temporal.nexus.Nexus;
28-
import io.temporal.nexus.WorkflowClientOperationHandlers;
28+
import io.temporal.nexus.WorkflowRunOperation;
2929
import io.temporal.testing.WorkflowReplayer;
3030
import io.temporal.testing.internal.SDKTestWorkflowRule;
3131
import io.temporal.workflow.*;
@@ -129,7 +129,7 @@ public void unblock() {
129129
public class TestNexusServiceImpl {
130130
@OperationImpl
131131
public OperationHandler<String, String> operation() {
132-
return WorkflowClientOperationHandlers.fromWorkflowMethod(
132+
return WorkflowRunOperation.fromWorkflowMethod(
133133
(context, details, input) ->
134134
Nexus.getOperationContext()
135135
.getWorkflowClient()

‎temporal-sdk/src/test/java/io/temporal/workflow/nexus/CancelAsyncOperationTest.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@
3030
import io.temporal.failure.CanceledFailure;
3131
import io.temporal.failure.NexusOperationFailure;
3232
import io.temporal.nexus.Nexus;
33-
import io.temporal.nexus.WorkflowClientOperationHandlers;
33+
import io.temporal.nexus.WorkflowRunOperation;
3434
import io.temporal.testing.internal.SDKTestWorkflowRule;
3535
import io.temporal.testing.internal.TracingWorkerInterceptor;
3636
import io.temporal.workflow.*;
@@ -135,7 +135,7 @@ public String execute(String input) {
135135
public class TestNexusServiceImpl {
136136
@OperationImpl
137137
public OperationHandler<String, String> operation() {
138-
return WorkflowClientOperationHandlers.fromWorkflowMethod(
138+
return WorkflowRunOperation.fromWorkflowMethod(
139139
(context, details, input) ->
140140
Nexus.getOperationContext()
141141
.getWorkflowClient()

‎temporal-sdk/src/test/java/io/temporal/workflow/nexus/ParallelWorkflowOperationTest.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@
2525
import io.nexusrpc.handler.ServiceImpl;
2626
import io.temporal.client.WorkflowOptions;
2727
import io.temporal.nexus.Nexus;
28-
import io.temporal.nexus.WorkflowClientOperationHandlers;
28+
import io.temporal.nexus.WorkflowRunOperation;
2929
import io.temporal.testing.WorkflowReplayer;
3030
import io.temporal.testing.internal.SDKTestWorkflowRule;
3131
import io.temporal.workflow.*;
@@ -111,7 +111,7 @@ public String execute(String arg) {
111111
public class TestNexusServiceImpl {
112112
@OperationImpl
113113
public OperationHandler<String, String> operation() {
114-
return WorkflowClientOperationHandlers.fromWorkflowMethod(
114+
return WorkflowRunOperation.fromWorkflowMethod(
115115
(context, details, input) ->
116116
Nexus.getOperationContext()
117117
.getWorkflowClient()

‎temporal-sdk/src/test/java/io/temporal/workflow/nexus/TerminateWorkflowAsyncOperationTest.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@
3131
import io.temporal.failure.NexusOperationFailure;
3232
import io.temporal.failure.TerminatedFailure;
3333
import io.temporal.nexus.Nexus;
34-
import io.temporal.nexus.WorkflowClientOperationHandlers;
34+
import io.temporal.nexus.WorkflowRunOperation;
3535
import io.temporal.testing.internal.SDKTestWorkflowRule;
3636
import io.temporal.workflow.*;
3737
import io.temporal.workflow.shared.TestWorkflows;
@@ -109,7 +109,7 @@ public String execute(String input) {
109109
public class TestNexusServiceImpl {
110110
@OperationImpl
111111
public OperationHandler<String, String> operation() {
112-
return WorkflowClientOperationHandlers.fromWorkflowMethod(
112+
return WorkflowRunOperation.fromWorkflowMethod(
113113
(context, details, input) ->
114114
Nexus.getOperationContext()
115115
.getWorkflowClient()

0 commit comments

Comments
 (0)