Temporal Typescript SDK 學習筆記

import { Worker } from '@temporalio/worker';
import * as activities from './activities';

const workers = []

async function run() {
  const worker = await Worker.create({
    workflowsPath: require.resolve('./workflows'),
    activities,
    taskQueue: 'default',
  });
  workers.push(worker)

  const activitiesWorker = await Worker.create({
    activities,
    taskQuene: 'hello-world' // proxyActivities 時指定為 hello-world，這個 worker 就會去處理～
  });
  workers.push(worker)

  await Promise.all(workers)
}

run().catch((err) => {
  console.error(err);
  workers.forEach((w) => w.shutdown());
  process.exit(1);
});

// client.ts
import { Workflow } from '@temporalio/workflow';
import { Connection, WorkflowClient } from '@temporalio/client';

let client: WorkflowClient;

export async function getClient() {
  if (!client) {
    const connection = await Connection.connect({
      // Connect to localhost with default ConnectionOptions.
      // In production, pass options to the Connection constructor to configure TLS and other settings:
      // address: 'foo.bar.tmprl.cloud', // as provisioned
      // tls: {} // as provisioned
    });

    client = new WorkflowClient({
      connection,
      namespace: 'default', // change if you have a different namespace
    });
  }

  return client;
}

import { getClient } from './client'
import { example } from './workflows'

async function run() {
  const client = await getClient();
  const handle = await client.start(example, {
    args: ['Temporal'], // 參數傳遞給 example workflow
    taskQueue: 'hello-world', // 加入的 taskQuene 名稱，對應 worker 必須能處理此 workflow
    workflowId: 'example-workflow-id',
    // workflow id reuse policy 參考結尾列表
    workflowIdReusePolicy: WorkflowIdReusePolicy.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE,
    // retry policy
    retry: {
      maximumAttempts: 20
    }
  })
  const res = await handle.result()
}

run()

// workflows.ts
import { defineSignal, setHandler, sleep } from '@temporalio/worker'

export const cancelSubscription = defineSignal('cancelSignal'); // new

export async function SubscriptionWorkflow(
  email: string,
  trialPeriod: string | number
) {
  let isCanceled = false; // internal variable to track cancel state

  setHandler(cancelSubscription, () => void (isCanceled = true)); // new

  await acts.sendWelcomeEmail(email);
  await sleep(trialPeriod); // sleep will wait a fixed time
  if (isCanceled) {
    await acts.sendCancellationEmailDuringTrialPeriod(email); // new
  } else {
    await acts.sendSubscriptionOverEmail(email);
  }
}

// other-client.ts
import { cancelSubscription } from './workflows';
import client from './client';

function signalWorkflow(workflowId, signalName, ...args) {
  const handle = client.getHandle(workflowId);
  await handle.signal(signalName, ...args);
}

signalWorkflow('my-workflow-id', cancelSubscription)

// workflows.ts
export const childCompletedSignal = defineSignal(childCompletedSignal)

export async function checkDataWorkflow() {
  let completedChildWorkflows = 0;
  let totalChildWorkflows = 0;

  setHandler(childCompletedSignal, () => {
    completedChildWorkflows += 1;
  });

  const { workflowId } = workflowInfo();
  const dataList = await getChildData(); // activity to call rest api to get childData

  const childHandles = dataList.map((data) => {
    return startChild(checkPartialData, {
      args: [{
        parentWorkflowId: workflowId,
        data
      }],
      workflowId: `check-child-${Date.now()}`
    })
  })

  totalChildWorkflows = childHandles.length;

  // 這邊其實可以用 executeChild 搭配 Promise.all 更直覺，但這只是一個示範
  // wait until all child complete
  await condition(() => completedChildWorkflows === totalChildWorkflows)

  return true
}

export async function checkPartialData({
  parentWorkflowId,
  data
}) {
  // check data...
  // send signal to parent by parentWorkflowId
  await signalWorkflow(parentWorkflowId, childCompletedSignal)
}

export const userInfoStatusQuery = defineQuery('userInfoStatusQuery');

export async function updateUserInfoWorkflow() {
  let completedAt = null

  // return workflow status when being queried
  setHandle(userInfoStatusQuery, () => ({
    completed: !!completedAt,
    completedAt,
  }))

  await updateUserInfo() // activity to update user info

  completedAt = new Date();

  return true; // 即使 workflow completed 後，query 此 workflowId 仍然能取得他的狀態
}

// other-client.ts
import { userInfoStatusQuery } from './workflows';
import client from './client';

async function queryWorkflow(workflowId, signalName, ...args) {
  const handle = client.getHandle(workflowId);
  return await handle.query(signalName, ...args);
}

queryWorkflow('my-workflow-id', userInfoStatusQuery).then(
  (status) => console.log(status.completedAt)
)

// workflows.ts
export async function fetchAllData({
  startPage,
  pageSize
}) {
  let hasNextPage = true;
  let page = startPage;

  while (hasNextPage) {
    hasNextPage = await executeChild(fetchPartialData, {
      args: [page, pageSize],
      workflowId: `fetch-partial-${page}`,
    });

    page += 1;
  }

  return true
}

export async function fetchPartialData(
  page = 0,
  pageSize = 1000
) {
  // getUsers could be your activity to call rest api to get users
  const { userIds, pageInfo } = await getUsers(page, pageSize);

  await Promise.all(
    userIds.map(async (userId) => {
      await executeChild(fetchData, {
        args: [userId],
        workflowId: `fetch-user-${userId}`,
      });
    })
  )

  return pageInfo.hasNextPage
}

export async function fetchData(userId) {
  // do something with userId
}

const wrapTimeout = (targetHandle, maxWaitSeconds = 30) => {
  return Promise.race([
    targetHandle,
    new Promise((r) => setTimeout(r, maxWaitSeconds * 1000)),
  ]);
}

const runWorkflow = async () => {
  const wfHandle = await client.start(myWorkflow, {
    args: [],
    workflowId: 'my-workflow-id'
  });
  // workflow 跑超過 30s 就直接往後不等，避免卡住
  await wrapTimeout(wfHandle.result(), 30)

  const status = await wfHandle.query(checkStatusQuery);
  return status;
};