Nexus handler near complete implementation #1708
Conversation
- This PR is blocked on publishing an initial version of the `nexus-rpc` package. - There is a TODO to figure out HandlerError and OperationError message rehydration, pending discussion. - Interceptors not yet implemented. - `WorkflowRunOperation` and `getClient()` not implemented for the `@temporalio/nexus` package. - Tests use the HTTP API directly in lieu of a workflow caller or strongly typed client, we can refactor those later.
| @@ -91,6 +91,17 @@ export async function decodeArrayFromPayloads( | |||
| return arrayFromPayloads(payloadConverter, await decodeOptional(payloadCodecs, payloads)); | |||
| } | |||
|
|
|||
| /** | |||
| * Decode `payloads` and then return {@link arrayFromPayloads}`. | |||
| if (nexusServices.has(s.name)) { | ||
| throw new TypeError(`Duplicate registration of nexus service ${s.name}`); | ||
| } | ||
| const ops = new Map<string, nexus.OperationHandler<any, any> | nexus.SyncOperationHandler<any, any>>(); |
There was a problem hiding this comment.
Should that be part of the Nexus SDK? Logic is trivial at present, but there I expect there will be some normalization work to be done at a later point (e.g. once we add decorators) which we wouldn't to diverge from one Nexus implementation to another. This also seems to be what we did in Java, and kind-of what we did in Go.
There was a problem hiding this comment.
Yeah, we discussed moving this to the Nexus SDK with Chad yesterday. I will do that.
There was a problem hiding this comment.
Oh, I missed that part of the conversation.
I will do that.
I only wanted to confirm that we agree on direction. I can do it myself if you don't have time/prefer not to do it (same apply to all suggestions I make on this PR).
| import getPort from 'get-port'; | ||
| import * as nexus from 'nexus-rpc'; | ||
| import * as protoJsonSerializer from 'proto3-json-serializer'; | ||
| import * as temporalnexus from '@temporalio/nexus'; |
There was a problem hiding this comment.
Dependency is missing in package.json and tsconfig.json
| }); | ||
|
|
||
| Runtime.install({ logger }); | ||
| t.context.httpPort = await getPort(); |
There was a problem hiding this comment.
TODO(JWH): This will be flaky in CI due to port collisions, and users will often need to do that themselves. Move this to Core.
| test.beforeEach(async (t) => { | ||
| const taskQueue = t.title + randomUUID(); | ||
| const { env } = t.context; | ||
| const response = await env.connection.operatorService.createNexusEndpoint({ |
There was a problem hiding this comment.
I understand this is the way for now, but what's the intent regarding this? Do we expect users to have to do the same in their own Temporal+Nexus tests?
| const input = await decodeFromPayload(this.dataConverter, payload); | ||
| if (typeof this.handler === 'function') { | ||
| const handler = this.handler as nexus.SyncOperationHandler<unknown, unknown>; | ||
| const output = await this.invokeUserCode('startOperation', handler.bind(undefined, input, options)); |
There was a problem hiding this comment.
Would be preferable to add some defensive checks here to prevent a user from returning a HandlerStartOperationResultSync or HandlerStartOperationResultAsync from their SyncOperationHandler. Sounds like an easy error to make. But I don't think there's currently a safe way of doing this.
nexus-rpcpackage.WorkflowRunOperationandgetClient()not implemented for the@temporalio/nexuspackage.