Add Workflow result interceptor to Client

Author: lorensr

packages/client/src/interceptors.ts

@@ -4,15 +4,16 @@
  * @module
  */
 
-import { Next, Headers } from '@temporalio/internal-workflow-common';
+import { Headers, Next, Workflow, WorkflowResultType } from '@temporalio/internal-workflow-common';
 import { temporal } from '@temporalio/proto';
-import { CompiledWorkflowOptions } from './workflow-options';
 import {
   DescribeWorkflowExecutionResponse,
   RequestCancelWorkflowExecutionResponse,
   TerminateWorkflowExecutionResponse,
-  WorkflowExecution
+  WorkflowExecution,
 } from './types';
+import { WorkflowResultOptions } from './workflow-client';
+import { CompiledWorkflowOptions } from './workflow-options';
 
 export { Next, Headers };
 
@@ -65,7 +66,13 @@ export interface WorkflowCancelInput {
 /** Input for WorkflowClientCallsInterceptor.describe */
 export interface WorkflowDescribeInput {
   readonly workflowExecution: WorkflowExecution;
-  readonly firstExecutionRunId?: string;
+}
+
+/** Input for WorkflowClientCallsInterceptor.result */
+export interface WorkflowResultInput {
+  readonly workflowExecution: WorkflowExecution;
+  readonly runIdForResult?: string;
+  readonly resultOptions?: WorkflowResultOptions;
 }
 
 /**
@@ -109,6 +116,13 @@ export interface WorkflowClientCallsInterceptor {
    * Intercept a service call to describeWorkflowExecution
    */
   describe?: (input: WorkflowDescribeInput, next: Next<this, 'describe'>) => Promise<DescribeWorkflowExecutionResponse>;
+  /**
+   * Intercept a call to @{link WorkflowClient.result}
+   */
+  result?: <T extends Workflow>(
+    input: WorkflowResultInput,
+    next: Next<this, 'result'>
+  ) => Promise<WorkflowResultType<T>>;
 }
 
 interface WorkflowClientCallsInterceptorFactoryInput {

packages/client/src/workflow-client.ts

@@ -44,6 +44,7 @@ import {
   WorkflowClientInterceptors,
   WorkflowDescribeInput,
   WorkflowQueryInput,
+  WorkflowResultInput,
   WorkflowSignalInput,
   WorkflowSignalWithStartInput,
   WorkflowStartInput,
@@ -727,6 +728,18 @@ export class WorkflowClient {
     }
   }
 
+  /**
+   * Uses given input to call `WorkflowClient.result`
+   *
+   * Used as the final function of the result interceptor chain
+   */
+  protected async _resultWorkflowHandler<T extends Workflow>(
+    input: WorkflowResultInput
+  ): Promise<WorkflowResultType<T>> {
+    return this.result(input.workflowExecution.workflowId, input.runIdForResult, input.resultOptions);
+    // this.rethrowGrpcError(err, input.workflowExecution, 'Failed to describe workflow');
+  }
+
   /**
    * Create a new workflow handle for new or existing Workflow execution
    */
@@ -742,7 +755,13 @@ export class WorkflowClient {
       client: this,
       workflowId,
       async result(): Promise<WorkflowResultType<T>> {
-        return this.client.result(workflowId, runIdForResult, resultOptions);
+        const next = this.client._resultWorkflowHandler.bind(this.client);
+        const fn = interceptors.length ? composeInterceptors(interceptors, 'result', next) : next;
+        return await fn({
+          workflowExecution: { workflowId, runId },
+          runIdForResult,
+          resultOptions,
+        });
       },
       async terminate(reason?: string) {
         const next = this.client._terminateWorkflowHandler.bind(this.client);
@@ -766,7 +785,6 @@ export class WorkflowClient {
         const fn = interceptors.length ? composeInterceptors(interceptors, 'describe', next) : next;
         return await fn({
           workflowExecution: { workflowId, runId },
-          firstExecutionRunId,
         });
       },
       async signal<Args extends any[]>(def: SignalDefinition<Args> | string, ...args: Args): Promise<void> {

packages/test/src/test-interceptors.ts

@@ -5,8 +5,21 @@
  * @module
  */
 
-import { Connection, WorkflowClient, WorkflowFailedError } from '@temporalio/client';
-import { ApplicationFailure, TerminatedFailure, toPayload } from '@temporalio/common';
+import {
+  Connection,
+  WorkflowClient,
+  WorkflowClientCallsInterceptor,
+  WorkflowFailedError,
+  WorkflowResultInput,
+} from '@temporalio/client';
+import {
+  ApplicationFailure,
+  Next,
+  TerminatedFailure,
+  toPayload,
+  Workflow,
+  WorkflowResultType,
+} from '@temporalio/common';
 import { Core, DefaultLogger, Worker } from '@temporalio/worker';
 import { defaultPayloadConverter, WorkflowInfo } from '@temporalio/workflow';
 import test from 'ava';
@@ -83,6 +96,13 @@ if (RUN_INTEGRATION_TESTS) {
               const result: string = (await next(input)) as any;
               return [...result].reverse().join('');
             },
+            async result<T extends Workflow>(
+              input: WorkflowResultInput,
+              next: Next<WorkflowClientCallsInterceptor, 'result'>
+            ) {
+              const result = await next(input);
+              return (result + '-intercepted') as WorkflowResultType<T>;
+            },
           }),
         ],
       },
@@ -99,7 +119,7 @@ if (RUN_INTEGRATION_TESTS) {
           wf.query(getSecretQuery).then((result) => t.is(result, '12345')),
         ]);
         const result = await wf.result();
-        t.is(result, message);
+        t.is(result, message + '-intercepted');
       }
       {
         const wf = await client.signalWithStart(interceptorExample, {
@@ -109,7 +129,7 @@ if (RUN_INTEGRATION_TESTS) {
           signalArgs: ['12345'],
         });
         const result = await wf.result();
-        t.is(result, message);
+        t.is(result, message + '-intercepted');
       }
     } finally {
       worker.shutdown();