|
3 | 3 | import static org.junit.Assume.assumeTrue; |
4 | 4 |
|
5 | 5 | import io.nexusrpc.OperationException; |
| 6 | +import io.nexusrpc.handler.OperationCancelDetails; |
| 7 | +import io.nexusrpc.handler.OperationContext; |
6 | 8 | import io.nexusrpc.handler.OperationHandler; |
7 | 9 | import io.nexusrpc.handler.OperationImpl; |
| 10 | +import io.nexusrpc.handler.OperationStartDetails; |
| 11 | +import io.nexusrpc.handler.OperationStartResult; |
8 | 12 | import io.nexusrpc.handler.ServiceImpl; |
9 | 13 | import io.temporal.api.nexus.v1.Endpoint; |
10 | 14 | import io.temporal.client.NexusClient; |
11 | 15 | import io.temporal.client.NexusOperationExecutionDescription; |
| 16 | +import io.temporal.client.NexusOperationFailedException; |
12 | 17 | import io.temporal.client.NexusOperationHandle; |
13 | 18 | import io.temporal.client.StartNexusOperationOptions; |
14 | 19 | import io.temporal.client.UntypedNexusOperationHandle; |
@@ -72,33 +77,84 @@ public void describeWithoutRunIdTargetsLatest() { |
72 | 77 |
|
73 | 78 | @Test |
74 | 79 | public void cancelSucceedsForStartedOperation() { |
75 | | - startOperation().cancel(); |
76 | | - // No exception — server accepted the cancel request. |
| 80 | + UntypedNexusOperationHandle handle = startPendingOperation(); |
| 81 | + handle.cancel(); |
| 82 | + assertCancellationRecorded(handle, /* expectedReason= */ null); |
77 | 83 | } |
78 | 84 |
|
79 | 85 | @Test |
80 | 86 | public void cancelWithReasonSucceedsForStartedOperation() { |
81 | | - startOperation().cancel("test-cancel-reason"); |
| 87 | + UntypedNexusOperationHandle handle = startPendingOperation(); |
| 88 | + handle.cancel("test-cancel-reason"); |
| 89 | + assertCancellationRecorded(handle, "test-cancel-reason"); |
82 | 90 | } |
83 | 91 |
|
84 | 92 | @Test |
85 | 93 | public void cancelWithNullReasonSucceeds() { |
86 | | - startOperation().cancel(null); |
| 94 | + UntypedNexusOperationHandle handle = startPendingOperation(); |
| 95 | + handle.cancel(null); |
| 96 | + assertCancellationRecorded(handle, /* expectedReason= */ null); |
87 | 97 | } |
88 | 98 |
|
89 | 99 | @Test |
90 | 100 | public void terminateSucceedsForStartedOperation() { |
91 | | - startOperation().terminate(); |
| 101 | + UntypedNexusOperationHandle handle = startPendingOperation(); |
| 102 | + handle.terminate(); |
| 103 | + assertTerminatedFailure(handle); |
92 | 104 | } |
93 | 105 |
|
94 | 106 | @Test |
95 | 107 | public void terminateWithReasonSucceedsForStartedOperation() { |
96 | | - startOperation().terminate("test-terminate-reason"); |
| 108 | + UntypedNexusOperationHandle handle = startPendingOperation(); |
| 109 | + handle.terminate("test-terminate-reason"); |
| 110 | + assertTerminatedFailure(handle); |
97 | 111 | } |
98 | 112 |
|
99 | 113 | @Test |
100 | 114 | public void terminateWithNullReasonSucceeds() { |
101 | | - startOperation().terminate(null); |
| 115 | + UntypedNexusOperationHandle handle = startPendingOperation(); |
| 116 | + handle.terminate(null); |
| 117 | + assertTerminatedFailure(handle); |
| 118 | + } |
| 119 | + |
| 120 | + /** |
| 121 | + * Async operations stay in RUNNING state until something external transitions them — used by |
| 122 | + * cancel/terminate tests so the lifecycle RPC has a non-terminal operation to act on. |
| 123 | + */ |
| 124 | + private UntypedNexusOperationHandle startPendingOperation() { |
| 125 | + return startOperation(TestNexusServiceImpl.ASYNC_PREFIX + UUID.randomUUID()); |
| 126 | + } |
| 127 | + |
| 128 | + /** |
| 129 | + * Terminate is forceful and immediate per the proto contract; the server transitions the |
| 130 | + * operation to TERMINATED regardless of handler state, so {@code getResult} promptly throws |
| 131 | + * {@link NexusOperationFailedException}. |
| 132 | + */ |
| 133 | + private static void assertTerminatedFailure(UntypedNexusOperationHandle handle) { |
| 134 | + try { |
| 135 | + handle.getResult(15, java.util.concurrent.TimeUnit.SECONDS, String.class); |
| 136 | + Assert.fail("expected getResult to throw after the operation was terminated"); |
| 137 | + } catch (NexusOperationFailedException expected) { |
| 138 | + // The TerminatedFailure shows up either on this exception's message or via getCause(). |
| 139 | + } catch (java.util.concurrent.TimeoutException e) { |
| 140 | + Assert.fail("getResult timed out — terminate should have produced a terminal outcome"); |
| 141 | + } |
| 142 | + } |
| 143 | + |
| 144 | + /** |
| 145 | + * Cancel does NOT auto-transition the operation per the proto contract (handler cooperation is |
| 146 | + * required). Assert via {@code describe()} that the cancellation request was at least recorded |
| 147 | + * server-side. |
| 148 | + */ |
| 149 | + private static void assertCancellationRecorded( |
| 150 | + UntypedNexusOperationHandle handle, @javax.annotation.Nullable String expectedReason) { |
| 151 | + NexusOperationExecutionDescription description = handle.describe(); |
| 152 | + Assert.assertNotNull( |
| 153 | + "expected cancellation_info to be recorded after cancel()", |
| 154 | + description.getCancellationInfo()); |
| 155 | + if (expectedReason != null) { |
| 156 | + Assert.assertEquals(expectedReason, description.getCancellationInfo().getReason()); |
| 157 | + } |
102 | 158 | } |
103 | 159 |
|
104 | 160 | @Test |
@@ -165,17 +221,40 @@ public static class TestNexusServiceImpl { |
165 | 221 | /** Inputs starting with this prefix make the handler throw, exercising the failure path. */ |
166 | 222 | static final String FAIL_PREFIX = "FAIL:"; |
167 | 223 |
|
| 224 | + /** |
| 225 | + * Inputs starting with this prefix make the handler return an async-started result without ever |
| 226 | + * completing the operation. Used by cancel/terminate tests so the operation stays in RUNNING |
| 227 | + * state long enough for the lifecycle RPC to be observed. |
| 228 | + */ |
| 229 | + static final String ASYNC_PREFIX = "ASYNC:"; |
| 230 | + |
168 | 231 | @OperationImpl |
169 | 232 | public OperationHandler<String, String> operation() { |
170 | | - return OperationHandler.sync( |
171 | | - (context, details, input) -> { |
172 | | - if (input != null && input.startsWith(FAIL_PREFIX)) { |
173 | | - // OperationException.failed = definitive failure (no retries) so the caller's |
174 | | - // getResult surfaces the failure instead of timing out. |
175 | | - throw OperationException.failed("intentional failure: " + input); |
176 | | - } |
177 | | - return "echo:" + (input == null ? "<null>" : input); |
178 | | - }); |
| 233 | + return new OperationHandler<String, String>() { |
| 234 | + @Override |
| 235 | + public OperationStartResult<String> start( |
| 236 | + OperationContext context, OperationStartDetails details, String input) |
| 237 | + throws OperationException { |
| 238 | + if (input != null && input.startsWith(FAIL_PREFIX)) { |
| 239 | + // OperationException.failed = definitive failure (no retries) so the caller's |
| 240 | + // getResult surfaces the failure instead of timing out. |
| 241 | + throw OperationException.failed("intentional failure: " + input); |
| 242 | + } |
| 243 | + if (input != null && input.startsWith(ASYNC_PREFIX)) { |
| 244 | + // Async-started: server keeps the operation in RUNNING state until something |
| 245 | + // external (terminate, cancellation that takes effect, or schedule-to-close) |
| 246 | + // transitions it. |
| 247 | + return OperationStartResult.async("token-" + UUID.randomUUID()); |
| 248 | + } |
| 249 | + return OperationStartResult.sync("echo:" + (input == null ? "<null>" : input)); |
| 250 | + } |
| 251 | + |
| 252 | + @Override |
| 253 | + public void cancel(OperationContext context, OperationCancelDetails details) { |
| 254 | + // No-op. Tests assert cancellation visibility via describe()'s CancellationInfo |
| 255 | + // rather than driving the operation to a terminal state from the handler. |
| 256 | + } |
| 257 | + }; |
179 | 258 | } |
180 | 259 | } |
181 | 260 |
|
|
0 commit comments