1056 lines
52 KiB
JavaScript
1056 lines
52 KiB
JavaScript
"use strict";
|
|
var __decorate = (this && this.__decorate) || function (decorators, target, key, desc) {
|
|
var c = arguments.length, r = c < 3 ? target : desc === null ? desc = Object.getOwnPropertyDescriptor(target, key) : desc, d;
|
|
if (typeof Reflect === "object" && typeof Reflect.decorate === "function") r = Reflect.decorate(decorators, target, key, desc);
|
|
else for (var i = decorators.length - 1; i >= 0; i--) if (d = decorators[i]) r = (c < 3 ? d(r) : c > 3 ? d(target, key, r) : d(target, key)) || r;
|
|
return c > 3 && r && Object.defineProperty(target, key, r), r;
|
|
};
|
|
Object.defineProperty(exports, "__esModule", { value: true });
|
|
exports.QueryNotRegisteredError = exports.QueryRejectedError = exports.WorkflowClient = exports.WithStartWorkflowOperation = void 0;
|
|
const grpc_js_1 = require("@grpc/grpc-js");
|
|
const uuid_1 = require("uuid");
|
|
const common_1 = require("@temporalio/common");
|
|
const codec_helpers_1 = require("@temporalio/common/lib/internal-non-workflow/codec-helpers");
|
|
const payload_search_attributes_1 = require("@temporalio/common/lib/converter/payload-search-attributes");
|
|
const interceptors_1 = require("@temporalio/common/lib/interceptors");
|
|
const type_helpers_1 = require("@temporalio/common/lib/type-helpers");
|
|
const internal_non_workflow_1 = require("@temporalio/common/lib/internal-non-workflow");
|
|
const internal_workflow_1 = require("@temporalio/common/lib/internal-workflow");
|
|
const proto_1 = require("@temporalio/proto");
|
|
const errors_1 = require("./errors");
|
|
const types_1 = require("./types");
|
|
const workflow_options_1 = require("./workflow-options");
|
|
const helpers_1 = require("./helpers");
|
|
const base_client_1 = require("./base-client");
|
|
const iterators_utils_1 = require("./iterators-utils");
|
|
const workflow_update_stage_1 = require("./workflow-update-stage");
|
|
const internal_1 = require("./internal");
|
|
const interceptor_adapters_1 = require("./interceptor-adapters");
|
|
const UpdateWorkflowExecutionLifecycleStage = proto_1.temporal.api.enums.v1.UpdateWorkflowExecutionLifecycleStage;
|
|
function defaultWorkflowClientOptions() {
|
|
return {
|
|
...(0, base_client_1.defaultBaseClientOptions)(),
|
|
interceptors: [],
|
|
queryRejectCondition: 'NONE',
|
|
};
|
|
}
|
|
function assertRequiredWorkflowOptions(opts) {
|
|
if (!opts.taskQueue) {
|
|
throw new TypeError('Missing WorkflowOptions.taskQueue');
|
|
}
|
|
if (!opts.workflowId) {
|
|
throw new TypeError('Missing WorkflowOptions.workflowId');
|
|
}
|
|
}
|
|
function ensureArgs(opts) {
|
|
const { args, ...rest } = opts;
|
|
return { args: args ?? [], ...rest };
|
|
}
|
|
const withStartWorkflowOperationResolve = Symbol();
|
|
const withStartWorkflowOperationReject = Symbol();
|
|
const withStartWorkflowOperationUsed = Symbol();
|
|
/**
|
|
* Define how to start a workflow when using {@link WorkflowClient.startUpdateWithStart} and
|
|
* {@link WorkflowClient.executeUpdateWithStart}. `workflowIdConflictPolicy` is required in the options.
|
|
*/
|
|
class WithStartWorkflowOperation {
|
|
workflowTypeOrFunc;
|
|
options;
|
|
[withStartWorkflowOperationUsed] = false;
|
|
[withStartWorkflowOperationResolve] = undefined;
|
|
[withStartWorkflowOperationReject] = undefined;
|
|
workflowHandlePromise;
|
|
constructor(workflowTypeOrFunc, options) {
|
|
this.workflowTypeOrFunc = workflowTypeOrFunc;
|
|
this.options = options;
|
|
this.workflowHandlePromise = new Promise((resolve, reject) => {
|
|
this[withStartWorkflowOperationResolve] = resolve;
|
|
this[withStartWorkflowOperationReject] = reject;
|
|
});
|
|
// Suppress unhandled rejection: if executeUpdateWithStart fails before any
|
|
// response, we reject this promise AND re-throw to the caller. Without this
|
|
// no-op catch the rejected promise has no handler and Node 15+ terminates.
|
|
this.workflowHandlePromise.catch(() => undefined);
|
|
}
|
|
async workflowHandle() {
|
|
return await this.workflowHandlePromise;
|
|
}
|
|
}
|
|
exports.WithStartWorkflowOperation = WithStartWorkflowOperation;
|
|
/**
|
|
* Client for starting Workflow executions and creating Workflow handles.
|
|
*
|
|
* Typically this client should not be instantiated directly, instead create the high level {@link Client} and use
|
|
* {@link Client.workflow} to interact with Workflows.
|
|
*/
|
|
class WorkflowClient extends base_client_1.BaseClient {
|
|
options;
|
|
constructor(options) {
|
|
super(options);
|
|
this.options = {
|
|
...defaultWorkflowClientOptions(),
|
|
...(0, internal_workflow_1.filterNullAndUndefined)(options ?? {}),
|
|
loadedDataConverter: this.dataConverter,
|
|
};
|
|
}
|
|
/**
|
|
* Raw gRPC access to the Temporal service.
|
|
*
|
|
* **NOTE**: The namespace provided in {@link options} is **not** automatically set on requests made via this service
|
|
* object.
|
|
*/
|
|
get workflowService() {
|
|
return this.connection.workflowService;
|
|
}
|
|
async _start(workflowTypeOrFunc, options, interceptors) {
|
|
const workflowType = (0, common_1.extractWorkflowType)(workflowTypeOrFunc);
|
|
assertRequiredWorkflowOptions(options);
|
|
const compiledOptions = (0, workflow_options_1.compileWorkflowOptions)(ensureArgs(options));
|
|
const adaptedInterceptors = interceptors.map((i) => (0, interceptor_adapters_1.adaptWorkflowClientInterceptor)(i));
|
|
const startWithDetails = (0, interceptors_1.composeInterceptors)(adaptedInterceptors, 'startWithDetails', this._startWorkflowHandler.bind(this));
|
|
return startWithDetails({
|
|
options: compiledOptions,
|
|
headers: {},
|
|
workflowType,
|
|
});
|
|
}
|
|
async _signalWithStart(workflowTypeOrFunc, options, interceptors) {
|
|
const workflowType = (0, common_1.extractWorkflowType)(workflowTypeOrFunc);
|
|
const { signal, signalArgs, ...rest } = options;
|
|
assertRequiredWorkflowOptions(rest);
|
|
const compiledOptions = (0, workflow_options_1.compileWorkflowOptions)(ensureArgs(rest));
|
|
const signalWithStart = (0, interceptors_1.composeInterceptors)(interceptors, 'signalWithStart', this._signalWithStartWorkflowHandler.bind(this));
|
|
return signalWithStart({
|
|
options: compiledOptions,
|
|
headers: {},
|
|
workflowType,
|
|
signalName: typeof signal === 'string' ? signal : signal.name,
|
|
signalArgs: signalArgs ?? [],
|
|
});
|
|
}
|
|
/**
|
|
* Start a new Workflow execution.
|
|
*
|
|
* @returns a {@link WorkflowHandle} to the started Workflow
|
|
*/
|
|
async start(workflowTypeOrFunc, options) {
|
|
const { workflowId } = options;
|
|
const interceptors = this.getOrMakeInterceptors(workflowId);
|
|
const wfStartOutput = await this._start(workflowTypeOrFunc, { ...options, workflowId }, interceptors);
|
|
// runId is not used in handles created with `start*` calls because these
|
|
// handles should allow interacting with the workflow if it continues as new.
|
|
const baseHandle = this._createWorkflowHandle({
|
|
workflowId,
|
|
runId: undefined,
|
|
firstExecutionRunId: wfStartOutput.runId,
|
|
runIdForResult: wfStartOutput.runId,
|
|
interceptors,
|
|
followRuns: options.followRuns ?? true,
|
|
});
|
|
return {
|
|
...baseHandle,
|
|
firstExecutionRunId: wfStartOutput.runId,
|
|
eagerlyStarted: wfStartOutput.eagerlyStarted,
|
|
};
|
|
}
|
|
/**
|
|
* Start a new Workflow Execution and immediately send a Signal to that Workflow.
|
|
*
|
|
* The behavior of Signal-with-Start in the case where there is already a running Workflow with
|
|
* the given Workflow ID depends on the {@link WorkflowIDConflictPolicy}. That is, if the policy
|
|
* is `USE_EXISTING`, then the Signal is issued against the already existing Workflow Execution;
|
|
* however, if the policy is `FAIL`, then an error is thrown. If no policy is specified,
|
|
* Signal-with-Start defaults to `USE_EXISTING`.
|
|
*
|
|
* @returns a {@link WorkflowHandle} to the started Workflow
|
|
*/
|
|
async signalWithStart(workflowTypeOrFunc, options) {
|
|
const { workflowId } = options;
|
|
const interceptors = this.getOrMakeInterceptors(workflowId);
|
|
const runId = await this._signalWithStart(workflowTypeOrFunc, options, interceptors);
|
|
// runId is not used in handles created with `start*` calls because these
|
|
// handles should allow interacting with the workflow if it continues as new.
|
|
const handle = this._createWorkflowHandle({
|
|
workflowId,
|
|
runId: undefined,
|
|
firstExecutionRunId: undefined, // We don't know if this runId is first in the chain or not
|
|
runIdForResult: runId,
|
|
interceptors,
|
|
followRuns: options.followRuns ?? true,
|
|
}); // Cast is safe because we know we add the signaledRunId below
|
|
handle /* readonly */.signaledRunId = runId;
|
|
return handle;
|
|
}
|
|
/**
|
|
* Start a new Workflow Execution and immediately send an Update to that Workflow,
|
|
* then await and return the Update's result.
|
|
*
|
|
* The `updateOptions` object must contain a {@link WithStartWorkflowOperation}, which defines
|
|
* the options for the Workflow execution to start (e.g. the Workflow's type, task queue, input
|
|
* arguments, etc.)
|
|
*
|
|
* The behavior of Update-with-Start in the case where there is already a running Workflow with
|
|
* the given Workflow ID depends on the specified {@link WorkflowIDConflictPolicy}. That is, if
|
|
* the policy is `USE_EXISTING`, then the Update is issued against the already existing Workflow
|
|
* Execution; however, if the policy is `FAIL`, then an error is thrown. Caller MUST specify
|
|
* the desired WorkflowIDConflictPolicy.
|
|
*
|
|
* This call will block until the Update has completed. The Workflow handle can be retrieved by
|
|
* awaiting on {@link WithStartWorkflowOperation.workflowHandle}, whether or not the Update
|
|
* succeeds.
|
|
*
|
|
* @returns the Update result
|
|
*/
|
|
async executeUpdateWithStart(updateDef, updateOptions) {
|
|
const handle = await this._startUpdateWithStart(updateDef, {
|
|
...updateOptions,
|
|
waitForStage: workflow_update_stage_1.WorkflowUpdateStage.COMPLETED,
|
|
});
|
|
return await handle.result();
|
|
}
|
|
/**
|
|
* Start a new Workflow Execution and immediately send an Update to that Workflow,
|
|
* then return a {@link WorkflowUpdateHandle} for that Update.
|
|
*
|
|
* The `updateOptions` object must contain a {@link WithStartWorkflowOperation}, which defines
|
|
* the options for the Workflow execution to start (e.g. the Workflow's type, task queue, input
|
|
* arguments, etc.)
|
|
*
|
|
* The behavior of Update-with-Start in the case where there is already a running Workflow with
|
|
* the given Workflow ID depends on the specified {@link WorkflowIDConflictPolicy}. That is, if
|
|
* the policy is `USE_EXISTING`, then the Update is issued against the already existing Workflow
|
|
* Execution; however, if the policy is `FAIL`, then an error is thrown. Caller MUST specify
|
|
* the desired WorkflowIDConflictPolicy.
|
|
*
|
|
* This call will block until the Update has reached the specified {@link WorkflowUpdateStage}.
|
|
* Note that this means that the call will not return successfully until the Update has
|
|
* been delivered to a Worker. The Workflow handle can be retrieved by awaiting on
|
|
* {@link WithStartWorkflowOperation.workflowHandle}, whether or not the Update succeeds.
|
|
*
|
|
* @returns a {@link WorkflowUpdateHandle} to the started Update
|
|
*/
|
|
async startUpdateWithStart(updateDef, updateOptions) {
|
|
return this._startUpdateWithStart(updateDef, updateOptions);
|
|
}
|
|
async _startUpdateWithStart(updateDef, updateWithStartOptions) {
|
|
const { waitForStage, args, startWorkflowOperation, ...updateOptions } = updateWithStartOptions;
|
|
const { workflowTypeOrFunc, options: workflowOptions } = startWorkflowOperation;
|
|
const { workflowId } = workflowOptions;
|
|
if (startWorkflowOperation[withStartWorkflowOperationUsed]) {
|
|
throw new Error('This WithStartWorkflowOperation instance has already been executed.');
|
|
}
|
|
startWorkflowOperation[withStartWorkflowOperationUsed] = true;
|
|
assertRequiredWorkflowOptions(workflowOptions);
|
|
const startUpdateWithStartInput = {
|
|
workflowType: (0, common_1.extractWorkflowType)(workflowTypeOrFunc),
|
|
workflowStartOptions: (0, workflow_options_1.compileWorkflowOptions)(ensureArgs(workflowOptions)),
|
|
workflowStartHeaders: {},
|
|
updateName: typeof updateDef === 'string' ? updateDef : updateDef.name,
|
|
updateArgs: args ?? [],
|
|
updateOptions,
|
|
updateHeaders: {},
|
|
};
|
|
const interceptors = this.getOrMakeInterceptors(workflowId);
|
|
const onStart = (startResponse) => startWorkflowOperation[withStartWorkflowOperationResolve](this._createWorkflowHandle({
|
|
workflowId,
|
|
firstExecutionRunId: startResponse.runId ?? undefined,
|
|
interceptors,
|
|
followRuns: workflowOptions.followRuns ?? true,
|
|
}));
|
|
const onStartError = (err) => {
|
|
startWorkflowOperation[withStartWorkflowOperationReject](err);
|
|
};
|
|
const fn = (0, interceptors_1.composeInterceptors)(interceptors, 'startUpdateWithStart', this._updateWithStartHandler.bind(this, waitForStage, onStart, onStartError));
|
|
const updateOutput = await fn(startUpdateWithStartInput);
|
|
let outcome = updateOutput.updateOutcome;
|
|
if (!outcome && waitForStage === workflow_update_stage_1.WorkflowUpdateStage.COMPLETED) {
|
|
outcome = await this._pollForUpdateOutcome(updateOutput.updateId, {
|
|
workflowId,
|
|
runId: updateOutput.workflowExecution.runId,
|
|
});
|
|
}
|
|
return this.createWorkflowUpdateHandle(updateOutput.updateId, workflowId, updateOutput.workflowExecution.runId, outcome);
|
|
}
|
|
/**
|
|
* Start a new Workflow execution, then await for its completion and return that Workflow's result.
|
|
*
|
|
* @returns the result of the Workflow execution
|
|
*/
|
|
async execute(workflowTypeOrFunc, options) {
|
|
const { workflowId } = options;
|
|
const interceptors = this.getOrMakeInterceptors(workflowId);
|
|
await this._start(workflowTypeOrFunc, options, interceptors);
|
|
return await this.result(workflowId, undefined, {
|
|
...options,
|
|
followRuns: options.followRuns ?? true,
|
|
});
|
|
}
|
|
/**
|
|
* Get the result of a Workflow execution.
|
|
*
|
|
* Follow the chain of execution in case Workflow continues as new, or has a cron schedule or retry policy.
|
|
*/
|
|
async result(workflowId, runId, opts) {
|
|
const followRuns = opts?.followRuns ?? true;
|
|
const execution = { workflowId, runId };
|
|
const req = {
|
|
namespace: this.options.namespace,
|
|
execution,
|
|
skipArchival: true,
|
|
waitNewEvent: true,
|
|
historyEventFilterType: proto_1.temporal.api.enums.v1.HistoryEventFilterType.HISTORY_EVENT_FILTER_TYPE_CLOSE_EVENT,
|
|
};
|
|
let ev;
|
|
for (;;) {
|
|
let res;
|
|
try {
|
|
res = await this.workflowService.getWorkflowExecutionHistory(req);
|
|
}
|
|
catch (err) {
|
|
this.rethrowGrpcError(err, 'Failed to get Workflow execution history', { workflowId, runId });
|
|
}
|
|
const events = res.history?.events;
|
|
if (events == null || events.length === 0) {
|
|
req.nextPageToken = res.nextPageToken;
|
|
continue;
|
|
}
|
|
if (events.length !== 1) {
|
|
throw new Error(`Expected at most 1 close event(s), got: ${events.length}`);
|
|
}
|
|
// getWorkflowExecutionHistory should never return an array of undefined events
|
|
ev = events[0];
|
|
if (ev.workflowExecutionCompletedEventAttributes) {
|
|
if (followRuns && ev.workflowExecutionCompletedEventAttributes.newExecutionRunId) {
|
|
execution.runId = ev.workflowExecutionCompletedEventAttributes.newExecutionRunId;
|
|
req.nextPageToken = undefined;
|
|
continue;
|
|
}
|
|
// Note that we can only return one value from our workflow function in JS.
|
|
// Ignore any other payloads in result
|
|
const [result] = await (0, internal_non_workflow_1.decodeArrayFromPayloads)(this.dataConverter, ev.workflowExecutionCompletedEventAttributes.result?.payloads);
|
|
return result;
|
|
}
|
|
else if (ev.workflowExecutionFailedEventAttributes) {
|
|
if (followRuns && ev.workflowExecutionFailedEventAttributes.newExecutionRunId) {
|
|
execution.runId = ev.workflowExecutionFailedEventAttributes.newExecutionRunId;
|
|
req.nextPageToken = undefined;
|
|
continue;
|
|
}
|
|
const { failure, retryState } = ev.workflowExecutionFailedEventAttributes;
|
|
throw new errors_1.WorkflowFailedError('Workflow execution failed', await (0, internal_non_workflow_1.decodeOptionalFailureToOptionalError)(this.dataConverter, failure), (0, common_1.decodeRetryState)(retryState));
|
|
}
|
|
else if (ev.workflowExecutionCanceledEventAttributes) {
|
|
const failure = new common_1.CancelledFailure('Workflow canceled', await (0, internal_non_workflow_1.decodeArrayFromPayloads)(this.dataConverter, ev.workflowExecutionCanceledEventAttributes.details?.payloads));
|
|
failure.stack = '';
|
|
throw new errors_1.WorkflowFailedError('Workflow execution cancelled', failure, common_1.RetryState.NON_RETRYABLE_FAILURE);
|
|
}
|
|
else if (ev.workflowExecutionTerminatedEventAttributes) {
|
|
const failure = new common_1.TerminatedFailure(ev.workflowExecutionTerminatedEventAttributes.reason || 'Workflow execution terminated');
|
|
failure.stack = '';
|
|
throw new errors_1.WorkflowFailedError(ev.workflowExecutionTerminatedEventAttributes.reason || 'Workflow execution terminated', failure, common_1.RetryState.NON_RETRYABLE_FAILURE);
|
|
}
|
|
else if (ev.workflowExecutionTimedOutEventAttributes) {
|
|
if (followRuns && ev.workflowExecutionTimedOutEventAttributes.newExecutionRunId) {
|
|
execution.runId = ev.workflowExecutionTimedOutEventAttributes.newExecutionRunId;
|
|
req.nextPageToken = undefined;
|
|
continue;
|
|
}
|
|
const failure = new common_1.TimeoutFailure('Workflow execution timed out', undefined, common_1.TimeoutType.START_TO_CLOSE);
|
|
failure.stack = '';
|
|
throw new errors_1.WorkflowFailedError('Workflow execution timed out', failure, (0, common_1.decodeRetryState)(ev.workflowExecutionTimedOutEventAttributes.retryState));
|
|
}
|
|
else if (ev.workflowExecutionContinuedAsNewEventAttributes) {
|
|
const { newExecutionRunId } = ev.workflowExecutionContinuedAsNewEventAttributes;
|
|
if (!newExecutionRunId) {
|
|
throw new TypeError('Expected service to return newExecutionRunId for WorkflowExecutionContinuedAsNewEvent');
|
|
}
|
|
if (!followRuns) {
|
|
throw new errors_1.WorkflowContinuedAsNewError('Workflow execution continued as new', newExecutionRunId);
|
|
}
|
|
execution.runId = newExecutionRunId;
|
|
req.nextPageToken = undefined;
|
|
continue;
|
|
}
|
|
}
|
|
}
|
|
rethrowUpdateGrpcError(err, fallbackMessage, workflowExecution) {
|
|
if ((0, errors_1.isGrpcServiceError)(err)) {
|
|
if (err.code === grpc_js_1.status.DEADLINE_EXCEEDED || err.code === grpc_js_1.status.CANCELLED) {
|
|
throw new errors_1.WorkflowUpdateRPCTimeoutOrCancelledError(err.details ?? 'Workflow update call timeout or cancelled', {
|
|
cause: err,
|
|
});
|
|
}
|
|
}
|
|
if (err instanceof common_1.CancelledFailure) {
|
|
throw new errors_1.WorkflowUpdateRPCTimeoutOrCancelledError(err.message ?? 'Workflow update call timeout or cancelled', {
|
|
cause: err,
|
|
});
|
|
}
|
|
this.rethrowGrpcError(err, fallbackMessage, workflowExecution);
|
|
}
|
|
rethrowGrpcError(err, fallbackMessage, workflowExecution) {
|
|
if ((0, errors_1.isGrpcServiceError)(err)) {
|
|
(0, helpers_1.rethrowKnownErrorTypes)(err);
|
|
if (err.code === grpc_js_1.status.NOT_FOUND) {
|
|
throw new common_1.WorkflowNotFoundError(err.details ?? 'Workflow not found', workflowExecution?.workflowId ?? '', workflowExecution?.runId);
|
|
}
|
|
throw new errors_1.ServiceError(fallbackMessage, { cause: err });
|
|
}
|
|
throw new errors_1.ServiceError('Unexpected error while making gRPC request', { cause: err });
|
|
}
|
|
/**
|
|
* Use given input to make a queryWorkflow call to the service
|
|
*
|
|
* Used as the final function of the query interceptor chain
|
|
*/
|
|
async _queryWorkflowHandler(input) {
|
|
const req = {
|
|
queryRejectCondition: input.queryRejectCondition,
|
|
namespace: this.options.namespace,
|
|
execution: input.workflowExecution,
|
|
query: {
|
|
queryType: input.queryType,
|
|
queryArgs: { payloads: await (0, internal_non_workflow_1.encodeToPayloads)(this.dataConverter, ...input.args) },
|
|
header: { fields: input.headers },
|
|
},
|
|
};
|
|
let response;
|
|
try {
|
|
response = await this.workflowService.queryWorkflow(req);
|
|
}
|
|
catch (err) {
|
|
if ((0, errors_1.isGrpcServiceError)(err)) {
|
|
(0, helpers_1.rethrowKnownErrorTypes)(err);
|
|
if (err.code === grpc_js_1.status.INVALID_ARGUMENT) {
|
|
throw new QueryNotRegisteredError(err.message.replace(/^3 INVALID_ARGUMENT: /, ''), err.code);
|
|
}
|
|
}
|
|
this.rethrowGrpcError(err, 'Failed to query Workflow', input.workflowExecution);
|
|
}
|
|
if (response.queryRejected) {
|
|
if (response.queryRejected.status === undefined || response.queryRejected.status === null) {
|
|
throw new TypeError('Received queryRejected from server with no status');
|
|
}
|
|
throw new QueryRejectedError(response.queryRejected.status);
|
|
}
|
|
if (!response.queryResult) {
|
|
throw new TypeError('Invalid response from server');
|
|
}
|
|
// We ignore anything but the first result
|
|
return await (0, internal_non_workflow_1.decodeFromPayloadsAtIndex)(this.dataConverter, 0, response.queryResult?.payloads);
|
|
}
|
|
async _createUpdateWorkflowRequest(lifecycleStage, input) {
|
|
const updateId = input.options?.updateId ?? (0, uuid_1.v4)();
|
|
return {
|
|
namespace: this.options.namespace,
|
|
workflowExecution: input.workflowExecution,
|
|
firstExecutionRunId: input.firstExecutionRunId,
|
|
waitPolicy: {
|
|
lifecycleStage,
|
|
},
|
|
request: {
|
|
meta: {
|
|
updateId,
|
|
identity: this.options.identity,
|
|
},
|
|
input: {
|
|
header: { fields: input.headers },
|
|
name: input.updateName,
|
|
args: { payloads: await (0, internal_non_workflow_1.encodeToPayloads)(this.dataConverter, ...input.args) },
|
|
},
|
|
},
|
|
};
|
|
}
|
|
/**
|
|
* Start the Update.
|
|
*
|
|
* Used as the final function of the interceptor chain during startUpdate and executeUpdate.
|
|
*/
|
|
async _startUpdateHandler(waitForStage, input) {
|
|
let waitForStageProto = (0, workflow_update_stage_1.encodeWorkflowUpdateStage)(waitForStage) ??
|
|
UpdateWorkflowExecutionLifecycleStage.UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_ACCEPTED;
|
|
waitForStageProto =
|
|
waitForStageProto >= UpdateWorkflowExecutionLifecycleStage.UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_ACCEPTED
|
|
? waitForStageProto
|
|
: UpdateWorkflowExecutionLifecycleStage.UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_ACCEPTED;
|
|
const request = await this._createUpdateWorkflowRequest(waitForStageProto, input);
|
|
// Repeatedly send UpdateWorkflowExecution until update is durable (if the server receives a request with
|
|
// an update ID that already exists, it responds with information for the existing update). If the
|
|
// requested wait stage is COMPLETED, further polling is done before returning the UpdateHandle.
|
|
let response;
|
|
try {
|
|
do {
|
|
response = await this.workflowService.updateWorkflowExecution(request);
|
|
} while (response.stage < UpdateWorkflowExecutionLifecycleStage.UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_ACCEPTED);
|
|
}
|
|
catch (err) {
|
|
this.rethrowUpdateGrpcError(err, 'Workflow Update failed', input.workflowExecution);
|
|
}
|
|
return {
|
|
updateId: request.request.meta.updateId,
|
|
workflowRunId: response.updateRef.workflowExecution.runId,
|
|
outcome: response.outcome ?? undefined,
|
|
};
|
|
}
|
|
/**
|
|
* Send the Update-With-Start MultiOperation request.
|
|
*
|
|
* Used as the final function of the interceptor chain during
|
|
* startUpdateWithStart and executeUpdateWithStart.
|
|
*/
|
|
async _updateWithStartHandler(waitForStage, onStart, onStartError, input) {
|
|
const startInput = {
|
|
workflowType: input.workflowType,
|
|
options: input.workflowStartOptions,
|
|
headers: input.workflowStartHeaders,
|
|
};
|
|
const updateInput = {
|
|
updateName: input.updateName,
|
|
args: input.updateArgs,
|
|
workflowExecution: {
|
|
workflowId: input.workflowStartOptions.workflowId,
|
|
},
|
|
options: input.updateOptions,
|
|
headers: input.updateHeaders,
|
|
};
|
|
let seenStart = false;
|
|
try {
|
|
const startRequest = await this.createStartWorkflowRequest(startInput);
|
|
const waitForStageProto = (0, workflow_update_stage_1.encodeWorkflowUpdateStage)(waitForStage);
|
|
const updateRequest = await this._createUpdateWorkflowRequest(waitForStageProto, updateInput);
|
|
const multiOpReq = {
|
|
namespace: this.options.namespace,
|
|
operations: [
|
|
{
|
|
startWorkflow: startRequest,
|
|
},
|
|
{
|
|
updateWorkflow: updateRequest,
|
|
},
|
|
],
|
|
};
|
|
let multiOpResp;
|
|
let startResp;
|
|
let updateResp;
|
|
let reachedStage;
|
|
// Repeatedly send ExecuteMultiOperation until update is durable (if the server receives a request with
|
|
// an update ID that already exists, it responds with information for the existing update). If the
|
|
// requested wait stage is COMPLETED, further polling is done before returning the UpdateHandle.
|
|
do {
|
|
multiOpResp = await this.workflowService.executeMultiOperation(multiOpReq);
|
|
startResp = multiOpResp.responses?.[0]
|
|
?.startWorkflow;
|
|
if (!seenStart) {
|
|
onStart(startResp);
|
|
seenStart = true;
|
|
}
|
|
updateResp = multiOpResp.responses?.[1]
|
|
?.updateWorkflow;
|
|
reachedStage =
|
|
updateResp.stage ??
|
|
UpdateWorkflowExecutionLifecycleStage.UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_UNSPECIFIED;
|
|
} while (reachedStage < UpdateWorkflowExecutionLifecycleStage.UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_ACCEPTED);
|
|
return {
|
|
workflowExecution: {
|
|
workflowId: updateResp.updateRef.workflowExecution.workflowId,
|
|
runId: updateResp.updateRef.workflowExecution.runId,
|
|
},
|
|
updateId: updateRequest.request.meta.updateId,
|
|
updateOutcome: updateResp.outcome ?? undefined,
|
|
};
|
|
}
|
|
catch (thrownError) {
|
|
let err = thrownError;
|
|
if ((0, errors_1.isGrpcServiceError)(err) && err.code === grpc_js_1.status.ALREADY_EXISTS) {
|
|
err = new common_1.WorkflowExecutionAlreadyStartedError('Workflow execution already started', input.workflowStartOptions.workflowId, input.workflowType);
|
|
}
|
|
if (!seenStart) {
|
|
onStartError(err);
|
|
}
|
|
if ((0, errors_1.isGrpcServiceError)(err)) {
|
|
this.rethrowUpdateGrpcError(err, 'Update-With-Start failed', updateInput.workflowExecution);
|
|
}
|
|
throw err;
|
|
}
|
|
}
|
|
createWorkflowUpdateHandle(updateId, workflowId, workflowRunId, outcome) {
|
|
return {
|
|
updateId,
|
|
workflowId,
|
|
workflowRunId,
|
|
result: async () => {
|
|
const completedOutcome = outcome ?? (await this._pollForUpdateOutcome(updateId, { workflowId, runId: workflowRunId }));
|
|
if (completedOutcome.failure) {
|
|
throw new errors_1.WorkflowUpdateFailedError('Workflow Update failed', await (0, internal_non_workflow_1.decodeOptionalFailureToOptionalError)(this.dataConverter, completedOutcome.failure));
|
|
}
|
|
else {
|
|
return await (0, internal_non_workflow_1.decodeFromPayloadsAtIndex)(this.dataConverter, 0, completedOutcome.success?.payloads);
|
|
}
|
|
},
|
|
};
|
|
}
|
|
/**
|
|
* Poll Update until a response with an outcome is received; return that outcome.
|
|
* This is used directly; no interceptor is available.
|
|
*/
|
|
async _pollForUpdateOutcome(updateId, workflowExecution) {
|
|
const req = {
|
|
namespace: this.options.namespace,
|
|
updateRef: { workflowExecution, updateId },
|
|
identity: this.options.identity,
|
|
waitPolicy: {
|
|
lifecycleStage: (0, workflow_update_stage_1.encodeWorkflowUpdateStage)(workflow_update_stage_1.WorkflowUpdateStage.COMPLETED),
|
|
},
|
|
};
|
|
for (;;) {
|
|
try {
|
|
const response = await this.workflowService.pollWorkflowExecutionUpdate(req);
|
|
if (response.outcome) {
|
|
return response.outcome;
|
|
}
|
|
}
|
|
catch (err) {
|
|
const wE = typeof workflowExecution.workflowId === 'string' ? workflowExecution : undefined;
|
|
this.rethrowUpdateGrpcError(err, 'Workflow Update Poll failed', wE);
|
|
}
|
|
}
|
|
}
|
|
/**
|
|
* Use given input to make a signalWorkflowExecution call to the service
|
|
*
|
|
* Used as the final function of the signal interceptor chain
|
|
*/
|
|
async _signalWorkflowHandler(input) {
|
|
const req = {
|
|
identity: this.options.identity,
|
|
namespace: this.options.namespace,
|
|
workflowExecution: input.workflowExecution,
|
|
requestId: (0, uuid_1.v4)(),
|
|
// control is unused,
|
|
signalName: input.signalName,
|
|
header: { fields: input.headers },
|
|
input: { payloads: await (0, internal_non_workflow_1.encodeToPayloads)(this.dataConverter, ...input.args) },
|
|
};
|
|
try {
|
|
await this.workflowService.signalWorkflowExecution(req);
|
|
}
|
|
catch (err) {
|
|
this.rethrowGrpcError(err, 'Failed to signal Workflow', input.workflowExecution);
|
|
}
|
|
}
|
|
/**
|
|
* Use given input to make a signalWithStartWorkflowExecution call to the service
|
|
*
|
|
* Used as the final function of the signalWithStart interceptor chain
|
|
*/
|
|
async _signalWithStartWorkflowHandler(input) {
|
|
const { identity } = this.options;
|
|
const { options, workflowType, signalName, signalArgs, headers } = input;
|
|
const req = {
|
|
namespace: this.options.namespace,
|
|
identity,
|
|
requestId: (0, uuid_1.v4)(),
|
|
workflowId: options.workflowId,
|
|
workflowIdReusePolicy: (0, common_1.encodeWorkflowIdReusePolicy)(options.workflowIdReusePolicy),
|
|
workflowIdConflictPolicy: (0, common_1.encodeWorkflowIdConflictPolicy)(options.workflowIdConflictPolicy),
|
|
workflowType: { name: workflowType },
|
|
input: { payloads: await (0, internal_non_workflow_1.encodeToPayloads)(this.dataConverter, ...options.args) },
|
|
signalName,
|
|
signalInput: { payloads: await (0, internal_non_workflow_1.encodeToPayloads)(this.dataConverter, ...signalArgs) },
|
|
taskQueue: {
|
|
kind: proto_1.temporal.api.enums.v1.TaskQueueKind.TASK_QUEUE_KIND_NORMAL,
|
|
name: options.taskQueue,
|
|
},
|
|
workflowExecutionTimeout: options.workflowExecutionTimeout,
|
|
workflowRunTimeout: options.workflowRunTimeout,
|
|
workflowTaskTimeout: options.workflowTaskTimeout,
|
|
workflowStartDelay: options.startDelay,
|
|
retryPolicy: options.retry ? (0, common_1.compileRetryPolicy)(options.retry) : undefined,
|
|
memo: options.memo ? { fields: await (0, internal_non_workflow_1.encodeMapToPayloads)(this.dataConverter, options.memo) } : undefined,
|
|
searchAttributes: options.searchAttributes || options.typedSearchAttributes // eslint-disable-line @typescript-eslint/no-deprecated
|
|
? {
|
|
indexedFields: (0, payload_search_attributes_1.encodeUnifiedSearchAttributes)(options.searchAttributes, options.typedSearchAttributes), // eslint-disable-line @typescript-eslint/no-deprecated
|
|
}
|
|
: undefined,
|
|
cronSchedule: options.cronSchedule,
|
|
header: { fields: headers },
|
|
userMetadata: await (0, codec_helpers_1.encodeUserMetadata)(this.dataConverter, options.staticSummary, options.staticDetails),
|
|
priority: options.priority ? (0, common_1.compilePriority)(options.priority) : undefined,
|
|
versioningOverride: options.versioningOverride ?? undefined,
|
|
};
|
|
try {
|
|
return (await this.workflowService.signalWithStartWorkflowExecution(req)).runId;
|
|
}
|
|
catch (err) {
|
|
if (err.code === grpc_js_1.status.ALREADY_EXISTS) {
|
|
throw new common_1.WorkflowExecutionAlreadyStartedError('Workflow execution already started', options.workflowId, workflowType);
|
|
}
|
|
this.rethrowGrpcError(err, 'Failed to signalWithStart Workflow', { workflowId: options.workflowId });
|
|
}
|
|
}
|
|
/**
|
|
* Use given input to make startWorkflowExecution call to the service
|
|
*
|
|
* Used as the final function of the start interceptor chain
|
|
*/
|
|
async _startWorkflowHandler(input) {
|
|
const req = await this.createStartWorkflowRequest(input);
|
|
const { options: opts, workflowType } = input;
|
|
const internalOptions = opts[internal_1.InternalWorkflowStartOptionsSymbol];
|
|
try {
|
|
const response = await this.workflowService.startWorkflowExecution(req);
|
|
if (internalOptions != null) {
|
|
internalOptions.backLink = response.link ?? undefined;
|
|
}
|
|
return {
|
|
runId: response.runId,
|
|
eagerlyStarted: response.eagerWorkflowTask != null,
|
|
};
|
|
}
|
|
catch (err) {
|
|
if (err.code === grpc_js_1.status.ALREADY_EXISTS) {
|
|
throw new common_1.WorkflowExecutionAlreadyStartedError('Workflow execution already started', opts.workflowId, workflowType);
|
|
}
|
|
this.rethrowGrpcError(err, 'Failed to start Workflow', { workflowId: opts.workflowId });
|
|
}
|
|
}
|
|
async createStartWorkflowRequest(input) {
|
|
const { options: opts, workflowType, headers } = input;
|
|
const { identity, namespace } = this.options;
|
|
const internalOptions = opts[internal_1.InternalWorkflowStartOptionsSymbol];
|
|
const supportsEagerStart = this.connection?.[types_1.InternalConnectionLikeSymbol]
|
|
?.supportsEagerStart;
|
|
if (opts.requestEagerStart && !supportsEagerStart) {
|
|
throw new Error('Eager workflow start requires a NativeConnection shared between client and worker. ' +
|
|
'Pass a NativeConnection via ClientOptions.connection, or disable requestEagerStart.');
|
|
}
|
|
return {
|
|
namespace,
|
|
identity,
|
|
requestId: internalOptions?.requestId ?? (0, uuid_1.v4)(),
|
|
workflowId: opts.workflowId,
|
|
workflowIdReusePolicy: (0, common_1.encodeWorkflowIdReusePolicy)(opts.workflowIdReusePolicy),
|
|
workflowIdConflictPolicy: (0, common_1.encodeWorkflowIdConflictPolicy)(opts.workflowIdConflictPolicy),
|
|
workflowType: { name: workflowType },
|
|
input: { payloads: await (0, internal_non_workflow_1.encodeToPayloads)(this.dataConverter, ...opts.args) },
|
|
taskQueue: {
|
|
kind: proto_1.temporal.api.enums.v1.TaskQueueKind.TASK_QUEUE_KIND_NORMAL,
|
|
name: opts.taskQueue,
|
|
},
|
|
workflowExecutionTimeout: opts.workflowExecutionTimeout,
|
|
workflowRunTimeout: opts.workflowRunTimeout,
|
|
workflowTaskTimeout: opts.workflowTaskTimeout,
|
|
workflowStartDelay: opts.startDelay,
|
|
retryPolicy: opts.retry ? (0, common_1.compileRetryPolicy)(opts.retry) : undefined,
|
|
memo: opts.memo ? { fields: await (0, internal_non_workflow_1.encodeMapToPayloads)(this.dataConverter, opts.memo) } : undefined,
|
|
searchAttributes: opts.searchAttributes || opts.typedSearchAttributes // eslint-disable-line @typescript-eslint/no-deprecated
|
|
? {
|
|
indexedFields: (0, payload_search_attributes_1.encodeUnifiedSearchAttributes)(opts.searchAttributes, opts.typedSearchAttributes), // eslint-disable-line @typescript-eslint/no-deprecated
|
|
}
|
|
: undefined,
|
|
cronSchedule: opts.cronSchedule,
|
|
header: { fields: headers },
|
|
userMetadata: await (0, codec_helpers_1.encodeUserMetadata)(this.dataConverter, opts.staticSummary, opts.staticDetails),
|
|
priority: opts.priority ? (0, common_1.compilePriority)(opts.priority) : undefined,
|
|
versioningOverride: opts.versioningOverride ?? undefined,
|
|
requestEagerExecution: opts.requestEagerStart,
|
|
...(0, internal_workflow_1.filterNullAndUndefined)(internalOptions ?? {}),
|
|
};
|
|
}
|
|
/**
|
|
* Use given input to make terminateWorkflowExecution call to the service
|
|
*
|
|
* Used as the final function of the terminate interceptor chain
|
|
*/
|
|
async _terminateWorkflowHandler(input) {
|
|
const req = {
|
|
namespace: this.options.namespace,
|
|
identity: this.options.identity,
|
|
...input,
|
|
details: {
|
|
payloads: input.details ? await (0, internal_non_workflow_1.encodeToPayloads)(this.dataConverter, ...input.details) : undefined,
|
|
},
|
|
firstExecutionRunId: input.firstExecutionRunId,
|
|
};
|
|
try {
|
|
return await this.workflowService.terminateWorkflowExecution(req);
|
|
}
|
|
catch (err) {
|
|
this.rethrowGrpcError(err, 'Failed to terminate Workflow', input.workflowExecution);
|
|
}
|
|
}
|
|
/**
|
|
* Uses given input to make requestCancelWorkflowExecution call to the service
|
|
*
|
|
* Used as the final function of the cancel interceptor chain
|
|
*/
|
|
async _cancelWorkflowHandler(input) {
|
|
try {
|
|
return await this.workflowService.requestCancelWorkflowExecution({
|
|
namespace: this.options.namespace,
|
|
identity: this.options.identity,
|
|
requestId: (0, uuid_1.v4)(),
|
|
workflowExecution: input.workflowExecution,
|
|
firstExecutionRunId: input.firstExecutionRunId,
|
|
});
|
|
}
|
|
catch (err) {
|
|
this.rethrowGrpcError(err, 'Failed to cancel workflow', input.workflowExecution);
|
|
}
|
|
}
|
|
/**
|
|
* Uses given input to make describeWorkflowExecution call to the service
|
|
*
|
|
* Used as the final function of the describe interceptor chain
|
|
*/
|
|
async _describeWorkflowHandler(input) {
|
|
try {
|
|
return await this.workflowService.describeWorkflowExecution({
|
|
namespace: this.options.namespace,
|
|
execution: input.workflowExecution,
|
|
});
|
|
}
|
|
catch (err) {
|
|
this.rethrowGrpcError(err, 'Failed to describe workflow', input.workflowExecution);
|
|
}
|
|
}
|
|
/**
|
|
* Create a new workflow handle for new or existing Workflow execution
|
|
*/
|
|
_createWorkflowHandle({ workflowId, runId, firstExecutionRunId, interceptors, runIdForResult, ...resultOptions }) {
|
|
const _startUpdate = async (def, waitForStage, options) => {
|
|
const next = this._startUpdateHandler.bind(this, waitForStage);
|
|
const fn = (0, interceptors_1.composeInterceptors)(interceptors, 'startUpdate', next);
|
|
const { args, ...opts } = options ?? {};
|
|
const input = {
|
|
workflowExecution: { workflowId, runId },
|
|
firstExecutionRunId,
|
|
updateName: typeof def === 'string' ? def : def.name,
|
|
args: args ?? [],
|
|
waitForStage,
|
|
headers: {},
|
|
options: opts,
|
|
};
|
|
const output = await fn(input);
|
|
const handle = this.createWorkflowUpdateHandle(output.updateId, input.workflowExecution.workflowId, output.workflowRunId, output.outcome);
|
|
if (!output.outcome && waitForStage === workflow_update_stage_1.WorkflowUpdateStage.COMPLETED) {
|
|
await this._pollForUpdateOutcome(handle.updateId, input.workflowExecution);
|
|
}
|
|
return handle;
|
|
};
|
|
return {
|
|
client: this,
|
|
workflowId,
|
|
async result() {
|
|
return this.client.result(workflowId, runIdForResult, resultOptions);
|
|
},
|
|
async terminate(reason) {
|
|
const next = this.client._terminateWorkflowHandler.bind(this.client);
|
|
const fn = (0, interceptors_1.composeInterceptors)(interceptors, 'terminate', next);
|
|
return await fn({
|
|
workflowExecution: { workflowId, runId },
|
|
reason,
|
|
firstExecutionRunId,
|
|
});
|
|
},
|
|
async cancel() {
|
|
const next = this.client._cancelWorkflowHandler.bind(this.client);
|
|
const fn = (0, interceptors_1.composeInterceptors)(interceptors, 'cancel', next);
|
|
return await fn({
|
|
workflowExecution: { workflowId, runId },
|
|
firstExecutionRunId,
|
|
});
|
|
},
|
|
async describe() {
|
|
const next = this.client._describeWorkflowHandler.bind(this.client);
|
|
const fn = (0, interceptors_1.composeInterceptors)(interceptors, 'describe', next);
|
|
const raw = await fn({
|
|
workflowExecution: { workflowId, runId },
|
|
});
|
|
const info = await (0, helpers_1.executionInfoFromRaw)(raw.workflowExecutionInfo ?? {}, this.client.dataConverter, raw);
|
|
const userMetadata = raw.executionConfig?.userMetadata;
|
|
return {
|
|
...info,
|
|
staticDetails: async () => (await (0, internal_non_workflow_1.decodeOptionalSinglePayload)(this.client.dataConverter, userMetadata?.details)) ?? undefined,
|
|
staticSummary: async () => (await (0, internal_non_workflow_1.decodeOptionalSinglePayload)(this.client.dataConverter, userMetadata?.summary)) ?? undefined,
|
|
raw,
|
|
};
|
|
},
|
|
async fetchHistory() {
|
|
let nextPageToken = undefined;
|
|
const events = Array();
|
|
for (;;) {
|
|
const response = await this.client.workflowService.getWorkflowExecutionHistory({
|
|
nextPageToken,
|
|
namespace: this.client.options.namespace,
|
|
execution: { workflowId, runId },
|
|
});
|
|
events.push(...(response.history?.events ?? []));
|
|
nextPageToken = response.nextPageToken;
|
|
if (nextPageToken == null || nextPageToken.length === 0)
|
|
break;
|
|
}
|
|
return proto_1.temporal.api.history.v1.History.create({ events });
|
|
},
|
|
async startUpdate(def, options) {
|
|
return await _startUpdate(def, options.waitForStage, options);
|
|
},
|
|
async executeUpdate(def, options) {
|
|
const handle = await _startUpdate(def, workflow_update_stage_1.WorkflowUpdateStage.COMPLETED, options);
|
|
return await handle.result();
|
|
},
|
|
getUpdateHandle(updateId) {
|
|
return this.client.createWorkflowUpdateHandle(updateId, workflowId, runId);
|
|
},
|
|
async signal(def, ...args) {
|
|
const next = this.client._signalWorkflowHandler.bind(this.client);
|
|
const fn = (0, interceptors_1.composeInterceptors)(interceptors, 'signal', next);
|
|
await fn({
|
|
workflowExecution: { workflowId, runId },
|
|
signalName: typeof def === 'string' ? def : def.name,
|
|
args,
|
|
headers: {},
|
|
});
|
|
},
|
|
async query(def, ...args) {
|
|
const next = this.client._queryWorkflowHandler.bind(this.client);
|
|
const fn = (0, interceptors_1.composeInterceptors)(interceptors, 'query', next);
|
|
return fn({
|
|
workflowExecution: { workflowId, runId },
|
|
queryRejectCondition: (0, types_1.encodeQueryRejectCondition)(this.client.options.queryRejectCondition),
|
|
queryType: typeof def === 'string' ? def : def.name,
|
|
args,
|
|
headers: {},
|
|
});
|
|
},
|
|
};
|
|
}
|
|
/**
|
|
* Create a handle to an existing Workflow.
|
|
*
|
|
* - If only `workflowId` is passed, and there are multiple Workflow Executions with that ID, the handle will refer to
|
|
* the most recent one.
|
|
* - If `workflowId` and `runId` are passed, the handle will refer to the specific Workflow Execution with that Run
|
|
* ID.
|
|
* - If `workflowId` and {@link GetWorkflowHandleOptions.firstExecutionRunId} are passed, the handle will refer to the
|
|
* most recent Workflow Execution in the *Chain* that started with `firstExecutionRunId`.
|
|
*
|
|
* A *Chain* is a series of Workflow Executions that share the same Workflow ID and are connected by:
|
|
* - Being part of the same {@link https://docs.temporal.io/typescript/clients#scheduling-cron-workflows | Cron}
|
|
* - {@link https://docs.temporal.io/typescript/workflows#continueasnew | Continue As New}
|
|
* - {@link https://typescript.temporal.io/api/interfaces/client.workflowoptions/#retry | Retries}
|
|
*
|
|
* This method does not validate `workflowId`. If there is no Workflow Execution with the given `workflowId`, handle
|
|
* methods like `handle.describe()` will throw a {@link WorkflowNotFoundError} error.
|
|
*/
|
|
getHandle(workflowId, runId, options) {
|
|
const interceptors = this.getOrMakeInterceptors(workflowId, runId);
|
|
return this._createWorkflowHandle({
|
|
workflowId,
|
|
runId,
|
|
firstExecutionRunId: options?.firstExecutionRunId,
|
|
runIdForResult: runId ?? options?.firstExecutionRunId,
|
|
interceptors,
|
|
followRuns: options?.followRuns ?? true,
|
|
});
|
|
}
|
|
async *_list(options) {
|
|
let nextPageToken = Buffer.alloc(0);
|
|
for (;;) {
|
|
let response;
|
|
try {
|
|
response = await this.workflowService.listWorkflowExecutions({
|
|
namespace: this.options.namespace,
|
|
query: options?.query,
|
|
nextPageToken,
|
|
pageSize: options?.pageSize,
|
|
});
|
|
}
|
|
catch (e) {
|
|
this.rethrowGrpcError(e, 'Failed to list workflows', undefined);
|
|
}
|
|
// Not decoding memo payloads concurrently even though we could have to keep the lazy nature of this iterator.
|
|
// Decoding is done for `memo` fields which tend to be small.
|
|
// We might decide to change that based on user feedback.
|
|
for (const raw of response.executions) {
|
|
yield await (0, helpers_1.executionInfoFromRaw)(raw, this.dataConverter, raw);
|
|
}
|
|
nextPageToken = response.nextPageToken;
|
|
if (nextPageToken == null || nextPageToken.length === 0)
|
|
break;
|
|
}
|
|
}
|
|
/**
|
|
* Return a list of Workflow Executions matching the given `query`.
|
|
*
|
|
* Note that the list of Workflow Executions returned is approximate and eventually consistent.
|
|
*
|
|
* More info on the concept of "visibility" and the query syntax on the Temporal documentation site:
|
|
* https://docs.temporal.io/visibility
|
|
*/
|
|
list(options) {
|
|
return {
|
|
[Symbol.asyncIterator]: () => this._list(options)[Symbol.asyncIterator](),
|
|
intoHistories: (intoHistoriesOptions) => {
|
|
return (0, iterators_utils_1.mapAsyncIterable)(this._list(options), async ({ workflowId, runId }) => ({
|
|
workflowId,
|
|
history: await this.getHandle(workflowId, runId).fetchHistory(),
|
|
}), { concurrency: intoHistoriesOptions?.concurrency ?? 5 });
|
|
},
|
|
};
|
|
}
|
|
/**
|
|
* Return the number of Workflow Executions matching the given `query`. If no `query` is provided, then return the
|
|
* total number of Workflow Executions for this namespace.
|
|
*
|
|
* Note that the number of Workflow Executions returned is approximate and eventually consistent.
|
|
*
|
|
* More info on the concept of "visibility" and the query syntax on the Temporal documentation site:
|
|
* https://docs.temporal.io/visibility
|
|
*/
|
|
async count(query) {
|
|
let response;
|
|
try {
|
|
response = await this.workflowService.countWorkflowExecutions({
|
|
namespace: this.options.namespace,
|
|
query,
|
|
});
|
|
}
|
|
catch (e) {
|
|
this.rethrowGrpcError(e, 'Failed to count workflows');
|
|
}
|
|
return (0, helpers_1.decodeCountWorkflowExecutionsResponse)(response);
|
|
}
|
|
getOrMakeInterceptors(workflowId, runId) {
|
|
if (typeof this.options.interceptors === 'object' && 'calls' in this.options.interceptors) {
|
|
// eslint-disable-next-line @typescript-eslint/no-deprecated
|
|
const factories = this.options.interceptors.calls ?? [];
|
|
return factories.map((ctor) => ctor({ workflowId, runId }));
|
|
}
|
|
return Array.isArray(this.options.interceptors) ? this.options.interceptors : [];
|
|
}
|
|
}
|
|
exports.WorkflowClient = WorkflowClient;
|
|
let QueryRejectedError = class QueryRejectedError extends Error {
|
|
status;
|
|
constructor(status) {
|
|
super('Query rejected');
|
|
this.status = status;
|
|
}
|
|
};
|
|
exports.QueryRejectedError = QueryRejectedError;
|
|
exports.QueryRejectedError = QueryRejectedError = __decorate([
|
|
(0, type_helpers_1.SymbolBasedInstanceOfError)('QueryRejectedError')
|
|
], QueryRejectedError);
|
|
let QueryNotRegisteredError = class QueryNotRegisteredError extends Error {
|
|
code;
|
|
constructor(message, code) {
|
|
super(message);
|
|
this.code = code;
|
|
}
|
|
};
|
|
exports.QueryNotRegisteredError = QueryNotRegisteredError;
|
|
exports.QueryNotRegisteredError = QueryNotRegisteredError = __decorate([
|
|
(0, type_helpers_1.SymbolBasedInstanceOfError)('QueryNotRegisteredError')
|
|
], QueryNotRegisteredError);
|
|
//# sourceMappingURL=workflow-client.js.map
|