default activities (#1642)

Author: THardy98

packages/test/src/activities/default-and-defined.ts

@@ -0,0 +1,15 @@
+import { activityInfo } from '@temporalio/activity';
+
+export interface NameAndArgs {
+  name: string;
+  activityName?: string;
+  args: any[];
+}
+
+export async function definedActivity(...args: unknown[]): Promise<NameAndArgs> {
+  return { name: 'definedActivity', args };
+}
+
+export default async function (...args: unknown[]): Promise<NameAndArgs> {
+  return { name: 'default', activityName: activityInfo().activityType, args };
+}

packages/test/src/test-default-activity.ts

@@ -0,0 +1,35 @@
+import * as activities from './activities/default-and-defined';
+import { helpers, makeTestFunction } from './helpers-integration';
+import { workflowWithMaybeDefinedActivity } from './workflows/default-activity-wf';
+
+const test = makeTestFunction({ workflowsPath: require.resolve('./workflows/default-activity-wf') });
+
+test('Uses default activity if no matching activity exists', async (t) => {
+  const { executeWorkflow, createWorker } = helpers(t);
+  const worker = await createWorker({
+    activities,
+  });
+
+  await worker.runUntil(async () => {
+    const activityArgs = ['test', 'args'];
+    const res = await executeWorkflow(workflowWithMaybeDefinedActivity, {
+      args: [false, activityArgs],
+    });
+    t.deepEqual(res, { name: 'default', activityName: 'nonExistentActivity', args: activityArgs });
+  });
+});
+
+test('Does not use default activity if matching activity exists', async (t) => {
+  const { executeWorkflow, createWorker } = helpers(t);
+  const worker = await createWorker({
+    activities,
+  });
+
+  await worker.runUntil(async () => {
+    const activityArgs = ['test', 'args'];
+    const res = await executeWorkflow(workflowWithMaybeDefinedActivity, {
+      args: [true, activityArgs],
+    });
+    t.deepEqual(res, { name: 'definedActivity', args: activityArgs });
+  });
+});

packages/test/src/workflows/default-activity-wf.ts

@@ -0,0 +1,19 @@
+import { proxyActivities } from '@temporalio/workflow';
+import { NameAndArgs } from '../activities/default-and-defined';
+import type * as activities from '../activities/default-and-defined';
+
+const { definedActivity, nonExistentActivity } = proxyActivities<
+  typeof activities & { nonExistentActivity: (...args: unknown[]) => Promise<NameAndArgs> }
+>({
+  startToCloseTimeout: '30 seconds',
+});
+
+export async function workflowWithMaybeDefinedActivity(
+  useDefinedActivity: boolean,
+  activityArgs: unknown[]
+): Promise<NameAndArgs> {
+  if (useDefinedActivity) {
+    return await definedActivity(...activityArgs);
+  }
+  return await nonExistentActivity(...activityArgs);
+}

packages/worker/src/worker.ts

@@ -926,7 +926,8 @@ export class Worker {
                     );
 
                     const { activityType } = info;
-                    const fn = this.options.activities.get(activityType);
+                    // Use the corresponding activity if it exists, otherwise, fallback to default activity function (if exists)
+                    const fn = this.options.activities.get(activityType) ?? this.options.activities.get('default');
                     if (typeof fn !== 'function') {
                       throw ApplicationFailure.create({
                         type: 'NotFoundError',