Developer's guide - Features
The Features section of the Temporal Developer's guide provides basic implementation guidance on how to use many of the development features available to Workflows and Activities in the Temporal Platform.
This guide is a work in progress. Some sections may be incomplete or missing for some languages. Information may change at any time.
If you can't find what you are looking for in the Developer's guide, it could be in older docs for SDKs.
In this section you can find the following:
- How to develop Signals
- How to develop Queries
- How to start a Child Workflow Execution
- How to start a Temporal Cron Job
- How to use Continue-As-New
- How to set Workflow timeouts & retries
- How to set Activity timeouts & retries
- How to Heartbeat an Activity
- How to Asynchronously complete an Activity
- How to register Namespaces
- How to use custom payload conversion
Signals
A SignalWhat is a Signal?
A Signal is an asynchronous request to a Workflow Execution.
Learn more is a message sent to a running Workflow Execution.
Signals are defined in your code and handled in your Workflow Definition. Signals can be sent to Workflow Executions from a Temporal Client or from another Workflow Execution.
Define Signal
A Signal has a name and can have arguments.
- The name, also called a Signal type, is a string.
- The arguments must be serializable
What is a Data Converter?
A Data Converter is a Temporal SDK component that serializes and encodes data entering and exiting a Temporal Cluster.
Learn more.
- Go
- Java
- PHP
- Python
- TypeScript
Structs should be used to define Signals and carry data, as long as the struct is serializable via the Data Converter.
The Receive()
method on the Data Converter decodes the data into the Struct within the Workflow.
Only public fields are serializable.
MySignal struct {
Message string // serializable
message string // not serializable
}
The @SignalMethod
annotation indicates that the method is used to handle and react to external Signals.
@SignalMethod
void mySignal(String signalName);
The method can have parameters that contain the Signal payload and must be serializable by the default Jackson JSON Payload Converter.
void mySignal(String signalName, Object... args);
This method does not return a value and must have a void
return type.
Things to consider when defining Signals:
- Use Workflow object constructors and initialization blocks to initialize the internal data structures if possible.
- Signals might be received by a Workflow before the Workflow method is executed. When implementing Signals in scenarios where this can occur, assume that no parts of Workflow code ran. In some cases, Signal method implementation might require some initialization to be performed by the Workflow method code first—for example, when the Signal processing depends on, and is defined by the Workflow input. In this case, you can use a flag to determine whether the Workflow method is already triggered; if not, persist the Signal data into a collection for delayed processing by the Workflow method.
Workflows can answer synchronous QueriesHow to develop with Queries
A Query is a synchronous operation that is used to get the state of a Workflow Execution.
Learn more and receive SignalsHow to develop with Signals
A Signal is a message sent to a running Workflow Execution
Learn more.
All interface methods must have one of the following annotations:
- #[WorkflowMethod] indicates an entry point to a Workflow.
It contains parameters that specify timeouts and a Task Queue name.
Required parameters (such as
executionStartToCloseTimeoutSeconds
) that are not specified through the annotation must be provided at runtime. - #[SignalMethod] indicates a method that reacts to external signals. It must have a
void
return type. - #[QueryMethod] indicates a method that reacts to synchronous query requests. It must have a non
void
return type.
It is possible (though not recommended for usability reasons) to annotate concrete class implementation.
You can have more than one method with the same annotation (except #[WorkflowMethod]).
For example:
use Temporal\Workflow\WorkflowInterface;
use Temporal\Workflow\WorkflowMethod;
use Temporal\Workflow\SignalMethod;
use Temporal\Workflow\QueryMethod;
#[WorkflowInterface]
interface FileProcessingWorkflow
{
#[WorkflowMethod]
public function processFile(Argument $args);
#[QueryMethod("history")]
public function getHistory(): array;
#[QueryMethod("status")]
public function getStatus(): string;
#[SignalMethod]
public function retryNow(): void;
#[SignalMethod]
public function abandon(): void;
}
Note that name parameter of Workflow method annotations can be used to specify name of Workflow, Signal and Query types. If name is not specified the short name of the Workflow interface is used.
In the above code the #[WorkflowMethod(name)]
is not specified, thus the Workflow Type defaults to "FileProcessingWorkflow"
.
To define a Signal, set the Signal decorator @workflow.signal
on the Signal function inside your Workflow.
@workflow.signal
def your_signal(self, value: str) -> None:
self._signal = value
The @workflow.signal
decorator defines a method as a Signal. Signals can be asynchronous or synchronous methods and can be inherited; however, if a method is overridden, the override must also be decorated.
Dynamic Signals
You can use @workflow.signal(dynamic=True)
, which means all other unhandled Signals fall through to this.
Your method parameters must be self
, a string Signal name, and a *args
variable argument parameter.
@workflow.signal(dynamic=True)
def signal_dynamic(self, name: str, *args: Any) -> None:
self._last_event = f"signal_dynamic {name}: {args[0]}"
Customize name
Non-dynamic methods can only have positional arguments. Temporal suggests taking a single argument that is an object or data class of fields that can be added to as needed.
Return values from Signal methods are ignored.
You can have a name parameter to customize the Signal's name, otherwise it defaults to the unqualified method __name__
.
The following example sets a custom Signal name.
@workflow.signal(name="Custom-Name")
def signal(self, arg: str) -> None:
self._last_event = f"signal: {arg}"
You can either set the name
or the dynamic
parameter in a Signal's decorator, but not both.
- TypeScript
- JavaScript
import { defineSignal } from '@temporalio/workflow';
interface JoinInput {
userId: string;
groupId: string;
}
export const joinSignal = defineSignal<[JoinInput]>('join');
"use strict";
exports.__esModule = true;
exports.joinSignal = void 0;
// @ts-nocheck
var workflow_1 = require("@temporalio/workflow");
exports.joinSignal = (0, workflow_1.defineSignal)('join');
Handle Signal
Workflows listen for Signals by the Signal's name.
- Go
- Java
- PHP
- Python
- TypeScript
Use the GetSignalChannel()
API from the go.temporal.io/sdk/workflow
package to get the Signal Channel.
func YourWorkflowDefinition(ctx workflow.Context, param YourWorkflowParam) error {
// ...
var signal MySignal
signalChan := workflow.GetSignalChannel(ctx, "your-signal-name")
signalChan.Receive(ctx, &signal)
if len(signal.Message) > 0 && signal.Message != "SOME_VALUE" {
return errors.New("signal")
}
// ...
}
In the example above, the Workflow code uses workflow.GetSignalChannel
to open a workflow.Channel
for the Signal type (identified by the Signal name).
Before completing the Workflow or using Continue-As-New, make sure to do an asynchronous drain on the Signal channel. Otherwise, the Signals will be lost.
Use the @SignalMethod
annotation to handle Signals in the Workflow interface.
The Signal type defaults to the name of the method. In the following example, the Signal type defaults to retryNow
.
@WorkflowInterface
public interface FileProcessingWorkflow {
@WorkflowMethod
String processFile(Arguments args);
@SignalMethod
void retryNow();
}
To overwrite this default naming and assign a custom Signal type, use the @SignalMethod
annotation with the name
parameter.
In the following example, the Signal type is set to retrysignal
.
@WorkflowInterface
public interface FileProcessingWorkflow {
@WorkflowMethod
String processFile(Arguments args);
@SignalMethod(name = "retrysignal")
void retryNow();
}
A Workflow interface can define any number of methods annotated with @SignalMethod
, but the method names or the name
parameters for each must be unique.
In the following example, we define a Signal method updateGreeting
to update the greeting in the Workflow.
We set a Workflow.await
in the Workflow implementation to block the current Workflow Execution until the provided unblock condition is evaluated to true
.
In this case, the unblocking condition is evaluated to true
when the Signal to update the greeting is received.
@WorkflowInterface
public interface HelloWorld {
@WorkflowMethod
void sayHello(String name);
@SignalMethod
void updateGreeting(String greeting);
}
public class HelloWorldImpl implements HelloWorld {
private final Logger workflowLogger = Workflow.getLogger(HelloWorldImpl.class);
private String greeting;
@Override
public void sayHello(String name) {
int count = 0;
while (!"Bye".equals(greeting)) {
String oldGreeting = greeting;
Workflow.await(() -> !Objects.equals(greeting, oldGreeting));
}
workflowLogger.info(++count + ": " + greeting + " " + name + "!");
}
@Override
public void updateGreeting(String greeting) {
this.greeting = greeting;
}
}
This Workflow completes when the Signal updates the greeting to Bye
.
Dynamic Signal Handler You can also implement Signal handlers dynamically. This is useful for library-level code and implementation of DSLs.
Use Workflow.registerListener(Object)
to register an implementation of the DynamicSignalListener
in the Workflow implementation code.
Workflow.registerListener(
(DynamicSignalHandler)
(signalName, encodedArgs) -> name = encodedArgs.get(0, String.class));
When registered, any Signals sent to the Workflow without a defined handler will be delivered to the DynamicSignalHandler
.
Note that you can only register one Workflow.registerListener(Object)
per Workflow Execution.
DynamicSignalHandler
can be implemented in both regular and dynamic Workflow implementations.
Use the #[SignalMethod]
annotation to handle Signals in the Workflow interface:
use Temporal\Workflow;
#[Workflow\WorkflowInterface]
class YourWorkflow
{
private bool $value;
#[Workflow\WorkflowMethod]
public function run()
{
yield Workflow::await(fn()=> $this->value);
return 'OK';
}
#[Workflow\SignalMethod]
public function setValue(bool $value)
{
$this->value = $value;
}
}
In the preceding example, the Workflow updates the protected value.
The main Workflow coroutine waits for the value to change by using the Workflow::await()
function.
To send a Signal to the Workflow, use the signal
method from the WorkflowHandle
class.
await handle.signal("some signal")
- TypeScript
- JavaScript
import { setHandler } from '@temporalio/workflow';
export async function yourWorkflow() {
const groups = new Map<string, Set<string>>();
setHandler(joinSignal, ({ userId, groupId }: JoinInput) => {
const group = groups.get(groupId);
if (group) {
group.add(userId);
} else {
groups.set(groupId, new Set([userId]));
}
});
}
"use strict";
var __awaiter = (this && this.__awaiter) || function (thisArg, _arguments, P, generator) {
function adopt(value) { return value instanceof P ? value : new P(function (resolve) { resolve(value); }); }
return new (P || (P = Promise))(function (resolve, reject) {
function fulfilled(value) { try { step(generator.next(value)); } catch (e) { reject(e); } }
function rejected(value) { try { step(generator["throw"](value)); } catch (e) { reject(e); } }
function step(result) { result.done ? resolve(result.value) : adopt(result.value).then(fulfilled, rejected); }
step((generator = generator.apply(thisArg, _arguments || [])).next());
});
};
var __generator = (this && this.__generator) || function (thisArg, body) {
var _ = { label: 0, sent: function() { if (t[0] & 1) throw t[1]; return t[1]; }, trys: [], ops: [] }, f, y, t, g;
return g = { next: verb(0), "throw": verb(1), "return": verb(2) }, typeof Symbol === "function" && (g[Symbol.iterator] = function() { return this; }), g;
function verb(n) { return function (v) { return step([n, v]); }; }
function step(op) {
if (f) throw new TypeError("Generator is already executing.");
while (g && (g = 0, op[0] && (_ = 0)), _) try {
if (f = 1, y && (t = op[0] & 2 ? y["return"] : op[0] ? y["throw"] || ((t = y["return"]) && t.call(y), 0) : y.next) && !(t = t.call(y, op[1])).done) return t;
if (y = 0, t) op = [op[0] & 2, t.value];
switch (op[0]) {
case 0: case 1: t = op; break;
case 4: _.label++; return { value: op[1], done: false };
case 5: _.label++; y = op[1]; op = [0]; continue;
case 7: op = _.ops.pop(); _.trys.pop(); continue;
default:
if (!(t = _.trys, t = t.length > 0 && t[t.length - 1]) && (op[0] === 6 || op[0] === 2)) { _ = 0; continue; }
if (op[0] === 3 && (!t || (op[1] > t[0] && op[1] < t[3]))) { _.label = op[1]; break; }
if (op[0] === 6 && _.label < t[1]) { _.label = t[1]; t = op; break; }
if (t && _.label < t[2]) { _.label = t[2]; _.ops.push(op); break; }
if (t[2]) _.ops.pop();
_.trys.pop(); continue;
}
op = body.call(thisArg, _);
} catch (e) { op = [6, e]; y = 0; } finally { f = t = 0; }
if (op[0] & 5) throw op[1]; return { value: op[0] ? op[1] : void 0, done: true };
}
};
exports.__esModule = true;
exports.yourWorkflow = void 0;
// @ts-nocheck
var workflow_1 = require("@temporalio/workflow");
function yourWorkflow() {
return __awaiter(this, void 0, void 0, function () {
var groups;
return __generator(this, function (_a) {
groups = new Map();
(0, workflow_1.setHandler)(joinSignal, function (_a) {
var userId = _a.userId, groupId = _a.groupId;
var group = groups.get(groupId);
if (group) {
group.add(userId);
}
else {
groups.set(groupId, new Set([userId]));
}
});
return [2 /*return*/];
});
});
}
exports.yourWorkflow = yourWorkflow;
Send Signal from Client
When a Signal is sent successfully from the Temporal Client, the WorkflowExecutionSignaled Event appears in the Event History of the Workflow that receives the Signal.
- Go
- Java
- PHP
- Python
- TypeScript
Use the SignalWorkflow()
method on an instance of the Go SDK Temporal Client to send a SignalWhat is a Signal?
A Signal is an asynchronous request to a Workflow Execution.
Learn more to a Workflow Execution.
Pass in both the Workflow IdWhat is a Workflow Id?
A Workflow Id is a customizable, application-level identifier for a Workflow Execution that is unique to an Open Workflow Execution within a Namespace.
Learn more and Run IdWhat is a Run Id?
A Run Id is a globally unique, platform-level identifier for a Workflow Execution.
Learn more to uniquely identify the Workflow Execution.
If only the Workflow Id is supplied (provide an empty string as the Run Id param), the Workflow Execution that is Running receives the Signal.
// ...
signal := MySignal {
Message: "Some important data",
}
err = temporalClient.SignalWorkflow(context.Background(), "your-workflow-id", runID, "your-signal-name", signal)
if err != nil {
log.Fatalln("Error sending the Signal", err)
return
}
// ...
Possible errors:
serviceerror.NotFound
serviceerror.Internal
serviceerror.Unavailable
To send a Signal to a Workflow Execution from a Client, call the Signal method, annotated with @SignalMethod
in the Workflow interface, from the Client code.
In the following Client code example, we start the Workflow greetCustomer
and call the Signal method addCustomer
that is handled in the Workflow.
// create a typed Workflow stub for GreetingsWorkflow
GreetingsWorkflow workflow = client.newWorkflowStub(GreetingsWorkflow.class,
WorkflowOptions.newBuilder()
// set the Task Queue
.setTaskQueue(taskQueue)
// Workflow Id is recommended but not required
.setWorkflowId(workflowId)
.build());
// start the Workflow
WorkflowClient.start(workflow::greetCustomer);
// send a Signal to the Workflow
Customer customer = new Customer("John", "Spanish", "john@john.com");
workflow.addCustomer(customer); //addCustomer is the Signal method defined in the greetCustomer Workflow.
See Handle SignalsHow to handle Signals in an Workflow in Java
Use the @SignalMethod
annotation to handle Signals within the Workflow interface.
Learn more for details on how to handle Signals in a Workflow.
To send a Signal to a Workflow Execution from a Client, call the Signal method, annotated with #[SignalMethod]
in the Workflow interface, from the Client code.
To send a Signal to a Workflow, use WorkflowClient->newWorkflowStub
or WorkflowClient->newUntypedWorkflowStub
:
$workflow = $workflowClient->newWorkflowStub(YourWorkflow::class);
$run = $workflowClient->start($workflow);
// do something
$workflow->setValue(true);
assert($run->getValue() === true);
Use WorkflowClient->newRunningWorkflowStub
or WorkflowClient->newUntypedRunningWorkflowStub
with Workflow Id to send Signals to already running Workflows.
$workflow = $workflowClient->newRunningWorkflowStub(YourWorkflow::class, 'workflowID');
$workflow->setValue(true);
See Handle SignalHow to handle a Signal
Workflows listen for Signals by the Signal's name.
Learn more for details on how to handle Signals in a Workflow.
To send a Signal from the Client, use the signal() function on the Workflow handle.
To get the Workflow handle, you can use any of the following options.
- Use the get_workflow_handle() method.
- Use the get_workflow_handle_for() method to get a type-safe Workflow handle by its Workflow Id.
- Use the start_workflow() to start a Workflow and return its handle.
async def your_function():
client = await Client.connect("localhost:7233")
handle = client.get_workflow_handle_for(
"your-workflow-id",
)
await handle.signal()
import { WorkflowClient } from '@temporalio/client';
import { joinSignal } from './workflows';
const client = new WorkflowClient();
const handle = client.getHandle('workflow-id-123');
await handle.signal(joinSignal, { userId: 'user-1', groupId: 'group-1' });
Send Signal from Workflow
A Workflow can send a Signal to another Workflow, in which case it's called an External Signal.
When an External Signal is sent:
- A SignalExternalWorkflowExecutionInitiated Event appears in the sender's Event History.
- A WorkflowExecutionSignaled Event appears in the recipient's Event History.
- Go
- Java
- PHP
- Python
- TypeScript
A Signal can be sent from within a Workflow to a different Workflow Execution using the SignalExternalWorkflow
API from the go.temporal.io/sdk/workflow
package.
// ...
func YourWorkflowDefinition(ctx workflow.Context, param YourWorkflowParam) error {
//...
signal := MySignal {
Message: "Some important data",
}
err := workflow.SignalExternalWorkflow(ctx, "some-workflow-id", "", "your-signal-name", signal).Get(ctx, nil)
if err != nil {
// ...
}
// ...
}
To send a Signal from within a Workflow to a different Workflow Execution, initiate an ExternalWorkflowStub
in the implementation of the current Workflow and call the Signal method defined in the other Workflow.
The following example shows how to use an untyped ExternalWorkflowStub
in the Workflow implementation to send a Signal to another Workflow.
public String sendGreeting(String name) {
// initiate ExternalWorkflowStub to call another Workflow by its Id "ReplyWF"
ExternalWorkflowStub callRespondWorkflow = Workflow.newUntypedExternalWorkflowStub("ReplyWF");
String responseTrigger = activity.greeting("Hello", name);
// send a Signal from this sendGreeting Workflow to the other Workflow
// by calling the Signal method name "getGreetCall" defined in that Workflow.
callRespondWorkflow.signal("getGreetCall", responseTrigger);
return responseTrigger;
To send signal to a Workflow use WorkflowClient
->newWorkflowStub
or WorkflowClient
->newUntypedWorkflowStub
:
$workflow = $workflowClient->newWorkflowStub(YourWorkflow::class);
$run = $workflowClient->start($workflow);
// do something
$workflow->setValue(true);
assert($run->getValue() === true);
Use WorkflowClient
->newRunningWorkflowStub
or WorkflowClient->newUntypedRunningWorkflowStub
with Workflow Id to send
Signals to a running Workflow.
$workflow = $workflowClient->newRunningWorkflowStub(YourWorkflow::class, 'workflowID');
$workflow->setValue(true);
Use get_external_workflow_handle_for
to get a typed Workflow handle to an existing Workflow by its identifier. Use get_external_workflow_handle
when you don't know the type of the other Workflow.
@workflow.defn
class MyWorkflow:
@workflow.run
async run(self) -> None:
handle = workflow.get_external_workflow_handle_for(OtherWorkflow.run, "other-workflow-id")
await handle.signal(OtherWorkflow.other_signal, "other signal arg")
The Workflow Type passed is only for type annotations and not for validation.
import { getExternalWorkflowHandle } from '@temporalio/workflow';
import { joinSignal } from './other-workflow';
export async function yourWorkflowThatSignals() {
const handle = getExternalWorkflowHandle('workflow-id-123');
await handle.signal(joinSignal, { userId: 'user-1', groupId: 'group-1' });
}
Signal-With-Start
Signal-With-Start is used from the Client. It takes a Workflow Id, Workflow arguments, a Signal name, and Signal arguments.
If there's a Workflow running with the given Workflow Id, it will be signaled. If there isn't, a new Workflow will be started and immediately signaled.
- Go
- Java
- PHP
- Python
- TypeScript
Use the SignalWithStartWorkflow()
API on the Go SDK Temporal Client to start a Workflow Execution (if not already running) and pass it the Signal at the same time.
Because the Workflow Execution might not exist, this API does not take a Run ID as a parameter
// ...
signal := MySignal {
Message: "Some important data",
}
err = temporalClient.SignalWithStartWorkflow(context.Background(), "your-workflow-id", "your-signal-name", signal)
if err != nil {
log.Fatalln("Error sending the Signal", err)
return
}
To send Signals to a Workflow Execution whose status is unknown, use SignalWithStart
with a WorkflowStub
in the Client code.
This method ensures that if the Workflow Execution is in a closed state, a new Workflow Execution is spawned and the Signal is delivered to the running Workflow Execution.
Note that when the SignalwithStart
spawns a new Workflow Execution, the Signal is delivered before the call to your @WorkflowMethod
.
This means that the Signal handler in your Workflow interface code will execute before the @WorkfowMethod
.
You must ensure that your code logic can deal with this.
In the following example, the Client code uses SignalwithStart
to send the Signal setCustomer
to the UntypedWorkflowStub
named GreetingWorkflow
.
If the GreetingWorkflow
Workflow Execution is not running, the SignalwithStart
starts the Workflow Execution.
...
public static void signalWithStart() {
// WorkflowStub is a client-side stub to a single Workflow instance
WorkflowStub untypedWorkflowStub = client.newUntypedWorkflowStub("GreetingWorkflow",
WorkflowOptions.newBuilder()
.setWorkflowId(workflowId)
.setTaskQueue(taskQueue)
.build());
untypedWorkflowStub.signalWithStart("setCustomer", new Object[] {customer2}, new Object[] {customer1});
printWorkflowStatus();
try {
String greeting = untypedWorkflowStub.getResult(String.class);
printWorkflowStatus();
System.out.println("Greeting: " + greeting);
} catch(WorkflowFailedException e) {
System.out.println("Workflow failed: " + e.getCause().getMessage());
printWorkflowStatus();
}
}
...
The following example shows the Workflow interface for the GreetingWorkflow
called in the previous example.
...
@WorkflowInterface
public interface GreetingWorkflow {
@WorkflowMethod
String greet(Customer customer);
@SignalMethod
void setCustomer(Customer customer);
@QueryMethod
Customer getCustomer();
...
}
Note that the Signal handler setCustomer
is executed before the @WorkflowMethod
greet
is called.
In cases where you may not know if a Workflow is running, and want to send a Signal to it, use startwithSignal
.
If a running Workflow exists, the startwithSignal
API sends the Signal.
If there is no running Workflow, the API starts a new Workflow Run and delivers the Signal to it.
$workflow = $workflowClient->newWorkflowStub(YourWorkflow::class);
$run = $workflowClient->startWithSignal(
$workflow,
'setValue',
[true], // signal arguments
[] // start arguments
);
To send a Signal-With-Start in Python, use the start_workflow()
method and pass the start_signal
argument with the name of your Signal, instead of using a traditional Workflow start.
async def main():
client = await Client.connect("localhost:7233", namespace="your-namespace")
handle = await client.start_workflow(
"your-workflow-name",
"some arg",
id="your-workflow-id",
task_queue="your-task-queue",
start_signal="your-signal-name",
)
WorkflowClient.signalWithStart
import { WorkflowClient } from '@temporalio/client';
import { joinSignal, yourWorkflow } from './workflows';
const client = new WorkflowClient();
await client.signalWithStart(yourWorkflow, {
workflowId: 'workflow-id-123',
args: [{ foo: 1 }],
signal: joinSignal,
signalArgs: [{ userId: 'user-1', groupId: 'group-1' }],
});
Queries
A QueryWhat is a Query?
A Query is a synchronous operation that is used to report the state of a Workflow Execution.
Learn more is a synchronous operation that is used to get the state of a Workflow Execution.
Define Query
A Query has a name and can have arguments.
- The name, also called a Query type, is a string.
- The arguments must be serializable
What is a Data Converter?
A Data Converter is a Temporal SDK component that serializes and encodes data entering and exiting a Temporal Cluster.
Learn more.
- Go
- Java
- PHP
- Python
- TypeScript
In Go, a Query type, also called a Query name, is a string
value.
queryType := "your_query_name"
To define a Query, define the method name and the result type of the Query.
query(String queryType, Class<R> resultClass, Type resultType, Object... args);
/* @param queryType name of the Query handler. Usually it is a method name.
* @param resultClass class of the Query result type
* @param args optional Query arguments
* @param <R> type of the Query result
*/
Query methods can take in any number of input parameters which can be used to limit the data that is returned.
Use the Query method names to send and receive Queries.
Query methods must never change any Workflow state including starting Activities or blocking threads in any way.
Workflows can answer synchronous QueriesHow to develop with Queries
A Query is a synchronous operation that is used to get the state of a Workflow Execution.
Learn more and receive SignalsHow to develop with Signals
A Signal is a message sent to a running Workflow Execution
Learn more.
All interface methods must have one of the following annotations:
- #[WorkflowMethod] indicates an entry point to a Workflow.
It contains parameters that specify timeouts and a Task Queue name.
Required parameters (such as
executionStartToCloseTimeoutSeconds
) that are not specified through the annotation must be provided at runtime. - #[SignalMethod] indicates a method that reacts to external signals. It must have a
void
return type. - #[QueryMethod] indicates a method that reacts to synchronous query requests. It must have a non
void
return type.
It is possible (though not recommended for usability reasons) to annotate concrete class implementation.
You can have more than one method with the same annotation (except #[WorkflowMethod]).
For example:
use Temporal\Workflow\WorkflowInterface;
use Temporal\Workflow\WorkflowMethod;
use Temporal\Workflow\SignalMethod;
use Temporal\Workflow\QueryMethod;
#[WorkflowInterface]
interface FileProcessingWorkflow
{
#[WorkflowMethod]
public function processFile(Argument $args);
#[QueryMethod("history")]
public function getHistory(): array;
#[QueryMethod("status")]
public function getStatus(): string;
#[SignalMethod]
public function retryNow(): void;
#[SignalMethod]
public function abandon(): void;
}
Note that name parameter of Workflow method annotations can be used to specify name of Workflow, Signal and Query types. If name is not specified the short name of the Workflow interface is used.
In the above code the #[WorkflowMethod(name)]
is not specified, thus the Workflow Type defaults to "FileProcessingWorkflow"
.
To define a Query, set the Query decorator @workflow.query
on the Query function inside your Workflow.
@workflow.query
async def current_greeting(self) -> str:
return self._current_greeting
The @workflow.query
decorator defines a method as a Query. Queries can be asynchronous or synchronous methods and can be inherited; however, if a method is overridden, the override must also be decorated. Queries should return a value.
Dynamic Queries
You can use @workflow.query(dynamic=True)
, which means all other unhandled Queries fall through to this.
@workflow.query(dynamic=True)
def query_dynamic(self, name: str, *args: Any) -> str:
return f"query_dynamic {name}: {args[0]}"
Customize names
You can have a name parameter to customize the Query's name, otherwise it defaults to the unqualified method __name__
.
The following example sets a custom Query name.
@workflow.query(name="Custom-Name")
def query(self, arg: str) -> None:
self._last_event = f"query: {arg}"
You can either set the name
or the dynamic
parameter in a Query's decorator, but not both.
Use defineQuery
to define the name, parameters, and return value of a Query.
- TypeScript
- JavaScript
import { defineQuery } from '@temporalio/workflow';
export const getValueQuery = defineQuery<number | undefined, [string]>(
'getValue',
);
"use strict";
exports.__esModule = true;
exports.getValueQuery = void 0;
// @ts-nocheck
var workflow_1 = require("@temporalio/workflow");
exports.getValueQuery = (0, workflow_1.defineQuery)('getValue');
Handle Query
Queries are handled by your Workflow.
Don’t include any logic that causes CommandWhat is a Command?
A Command is a requested action issued by a Worker to the Temporal Cluster after a Workflow Task Execution completes.
Learn more generation within a Query handler (such as executing Activities).
Including such logic causes unexpected behavior.
- Go
- Java
- PHP
- Python
- TypeScript
Use the SetQueryHandler
API from the go.temporal.io/sdk/workflow
package to set a Query Handler that listens for a Query by name.
The handler must be a function that returns two values:
- A serializable result
- An error
The handler function can receive any number of input parameters, but all input parameters must be serializable.
The following sample code sets up a Query Handler that handles the current_state
Query type:
func YourWorkflow(ctx workflow.Context, input string) error {
currentState := "started" // This could be any serializable struct.
queryType := "current_state"
err := workflow.SetQueryHandler(ctx, queryType, func() (string, error) {
return currentState, nil
})
if err != nil {
currentState = "failed to register query handler"
return err
}
// Your normal Workflow code begins here, and you update the currentState as the code makes progress.
currentState = "waiting timer"
err = NewTimer(ctx, time.Hour).Get(ctx, nil)
if err != nil {
currentState = "timer failed"
return err
}
currentState = "waiting activity"
ctx = WithActivityOptions(ctx, yourActivityOptions)
err = ExecuteActivity(ctx, YourActivity, "your_input").Get(ctx, nil)
if err != nil {
currentState = "activity failed"
return err
}
currentState = "done"
return nil
}
For example, suppose your query handler function takes two parameters:
err := workflow.SetQueryHandler(ctx, "current_state", func(prefix string, suffix string) (string, error) {
return prefix + currentState + suffix, nil
})
To handle a Query in the Workflow, create a Query handler using the @QueryMethod
annotation in the Workflow interface and define it in the Workflow implementation.
The @QueryMethod
annotation indicates that the method is used to handle a Query that is sent to the Workflow Execution.
The method can have parameters that can be used to filter data that the Query returns.
Because the method returns a value, it must have a return type that is not void
.
The Query name defaults to the name of the method.
In the following example, the Query name defaults to getStatus
.
@WorkflowInterface
public interface FileProcessingWorkflow {
@QueryMethod
String getStatus();
}
To overwrite this default naming and assign a custom Query name, use the @QueryMethod
annotation with the name
parameter. In the following example, the Query name is set to "history".
@WorkflowInterface
public interface FileProcessingWorkflow {
@QueryMethod(name = "history")
String getStatus();
}
A Workflow Definition interface can define multiple methods annotated with @QueryMethod
, but the method names or the name
parameters for each must be unique.
The following Workflow interface has a Query method getCount()
to handle Queries to this Workflow.
@WorkflowInterface
public interface HelloWorld {
@WorkflowMethod
void sayHello(String name);
@QueryMethod
int getCount();
}
The following example is the Workflow implementation with the Query method defined in the HelloWorld
Workflow interface from the previous example.
public static class HelloWorldImpl implements HelloWorld {
private String greeting = "Hello";
private int count = 0;
@Override
public void sayHello(String name) {
while (!"Bye".equals(greeting)) {
logger.info(++count + ": " + greeting + " " + name + "!");
String oldGreeting = greeting;
Workflow.await(() -> !Objects.equals(greeting, oldGreeting));
}
logger.info(++count + ": " + greeting + " " + name + "!");
}
@Override
public int getCount() {
return count;
}
}
Dynamic Query Handler You can also implement Query handlers dynamically. This is useful for library-level code and implementation of DSLs.
Use Workflow.registerListener(Object)
to register an implementation of the DynamicQueryListener
in the Workflow implementation code.
Workflow.registerListener(
(DynamicQueryHandler)
(queryName, encodedArgs) -> name = encodedArgs.get(0, String.class));
When registered, any Queries sent to the Workflow without a defined handler will be delivered to the DynamicQueryHandler
.
Note that you can only register one Workflow.registerListener(Object)
per Workflow Execution.
DynamicQueryHandler
can be implemented in both regular and dynamic Workflow implementations.
You can add custom Query types to handle Queries such as Querying the current state of a
Workflow, or Querying how many Activities the Workflow has completed. To do this, you need to set
up a Query handler using method attribute QueryMethod
or Workflow::registerQueryHandler
.
#[Workflow\WorkflowInterface]
class YourWorkflow
{
#[Workflow\QueryMethod]
public function getValue()
{
return 42;
}
#[Workflow\WorkflowMethod]
public function run()
{
// workflow code
}
}
The handler function can receive any number of input parameters, but all input parameters must be
serializable. The following sample code sets up a Query handler that handles the Query type of
currentState
:
#[Workflow\WorkflowInterface]
class YourWorkflow
{
private string $currentState;
#[Workflow\QueryMethod('current_state')]
public function getCurrentState(): string
{
return $this->currentState;
}
#[Workflow\WorkflowMethod]
public function run()
{
// Your normal Workflow code begins here, and you update the currentState
// as the code makes progress.
$this->currentState = 'waiting timer';
try{
yield Workflow::timer(DateInterval::createFromDateString('1 hour'));
} catch (\Throwable $e) {
$this->currentState = 'timer failed';
throw $e;
}
$yourActivity = Workflow::newActivityStub(
YourActivityInterface::class,
ActivityOptions::new()->withScheduleToStartTimeout(60)
);
$this->currentState = 'waiting activity';
try{
yield $yourActivity->doSomething('some input');
} catch (\Throwable $e) {
$this->currentState = 'activity failed';
throw $e;
}
$this->currentState = 'done';
return null;
}
}
You can also issue a Query from code using the QueryWorkflow()
API on a Temporal Client object.
Use WorkflowStub
to Query Workflow instances from your Client code (can be applied to both running and closed Workflows):
$workflow = $workflowClient->newWorkflowStub(
YourWorkflow::class,
WorkflowOptions::new()
);
$workflowClient->start($workflow);
var_dump($workflow->getCurrentState());
sleep(60);
var_dump($workflow->getCurrentState());
To send a Query to the Workflow, use the query
method from the WorkflowHandle
class.
await handle.query("some query")
Use handleQuery
to handle Queries inside a Workflow.
You make a Query with handle.query(query, ...args)
. A Query needs a return value, but can also take arguments.
- TypeScript
- JavaScript
export async function trackState(): Promise<void> {
const state = new Map<string, number>();
setHandler(setValueSignal, (key, value) => void state.set(key, value));
setHandler(getValueQuery, (key) => state.get(key));
await CancellationScope.current().cancelRequested;
}
"use strict";
var __awaiter = (this && this.__awaiter) || function (thisArg, _arguments, P, generator) {
function adopt(value) { return value instanceof P ? value : new P(function (resolve) { resolve(value); }); }
return new (P || (P = Promise))(function (resolve, reject) {
function fulfilled(value) { try { step(generator.next(value)); } catch (e) { reject(e); } }
function rejected(value) { try { step(generator["throw"](value)); } catch (e) { reject(e); } }
function step(result) { result.done ? resolve(result.value) : adopt(result.value).then(fulfilled, rejected); }
step((generator = generator.apply(thisArg, _arguments || [])).next());
});
};
var __generator = (this && this.__generator) || function (thisArg, body) {
var _ = { label: 0, sent: function() { if (t[0] & 1) throw t[1]; return t[1]; }, trys: [], ops: [] }, f, y, t, g;
return g = { next: verb(0), "throw": verb(1), "return": verb(2) }, typeof Symbol === "function" && (g[Symbol.iterator] = function() { return this; }), g;
function verb(n) { return function (v) { return step([n, v]); }; }
function step(op) {
if (f) throw new TypeError("Generator is already executing.");
while (g && (g = 0, op[0] && (_ = 0)), _) try {
if (f = 1, y && (t = op[0] & 2 ? y["return"] : op[0] ? y["throw"] || ((t = y["return"]) && t.call(y), 0) : y.next) && !(t = t.call(y, op[1])).done) return t;
if (y = 0, t) op = [op[0] & 2, t.value];
switch (op[0]) {
case 0: case 1: t = op; break;
case 4: _.label++; return { value: op[1], done: false };
case 5: _.label++; y = op[1]; op = [0]; continue;
case 7: op = _.ops.pop(); _.trys.pop(); continue;
default:
if (!(t = _.trys, t = t.length > 0 && t[t.length - 1]) && (op[0] === 6 || op[0] === 2)) { _ = 0; continue; }
if (op[0] === 3 && (!t || (op[1] > t[0] && op[1] < t[3]))) { _.label = op[1]; break; }
if (op[0] === 6 && _.label < t[1]) { _.label = t[1]; t = op; break; }
if (t && _.label < t[2]) { _.label = t[2]; _.ops.push(op); break; }
if (t[2]) _.ops.pop();
_.trys.pop(); continue;
}
op = body.call(thisArg, _);
} catch (e) { op = [6, e]; y = 0; } finally { f = t = 0; }
if (op[0] & 5) throw op[1]; return { value: op[0] ? op[1] : void 0, done: true };
}
};
exports.__esModule = true;
exports.trackState = void 0;
// @ts-nocheck
function trackState() {
return __awaiter(this, void 0, void 0, function () {
var state;
return __generator(this, function (_a) {
switch (_a.label) {
case 0:
state = new Map();
setHandler(setValueSignal, function (key, value) { return void state.set(key, value); });
setHandler(getValueQuery, function (key) { return state.get(key); });
return [4 /*yield*/, CancellationScope.current().cancelRequested];
case 1:
_a.sent();
return [2 /*return*/];
}
});
});
}
exports.trackState = trackState;
Send Query
Queries are sent from a Temporal Client.
- Go
- Java
- PHP
- Python
- TypeScript
Use the QueryWorkflow()
API or the QueryWorkflowWithOptions
API on the Temporal Client to send a Query to a Workflow Execution.
// ...
response, err := temporalClient.QueryWorkflow(context.Background(), workflowID, runID, queryType)
if err != nil {
// ...
}
// ...
You can pass an arbitrary number of arguments to the QueryWorkflow()
function.
// ...
response, err := temporalClient.QueryWorkflow(context.Background(), workflowID, runID, queryType, "foo", "baz")
if err != nil {
// ...
}
// ...
The QueryWorkflowWithOptions()
API provides similar functionality, but with the ability to set additional configurations through QueryWorkflowWithOptionsRequest.
When using this API, you will also receive a structured response of type QueryWorkflowWithOptionsResponse.
// ...
response, err := temporalClient.QueryWorkflowWithOptions(context.Background(), &client.QueryWorkflowWithOptionsRequest{
WorkflowID: workflowID,
RunID: runID,
QueryType: queryType,
Args: args,
})
if err != nil {
// ...
}
To send a Query to a Workflow Execution from an external process, call the Query method (defined in the Workflow) from a WorkflowStub
within the Client code.
For example, the following Client code calls a Query method queryGreeting()
defined in the GreetingWorkflow
Workflow interface.
// Create our workflow options
WorkflowOptions workflowOptions =
WorkflowOptions.newBuilder()
.setWorkflowId(WORKFLOW_ID)
.setTaskQueue(TASK_QUEUE).build();
// Create the Temporal client stub. It is used to start our workflow execution.
GreetingWorkflow workflow = client.newWorkflowStub(GreetingWorkflow.class, workflowOptions);
// Start our workflow asynchronously to not use another thread to query.
WorkflowClient.start(workflow::createGreeting, "World");
// Query the Workflow to get the current value of greeting and print it.
System.out.println(workflow.queryGreeting());
Content is planned but not yet available.
The information you are looking for may be found in the legacy docs.
To send a Query to a Workflow Execution from Client code, use the query()
method on the Workflow handle.
await my_workflow_handle.query(MyWorkflow.my_query, "my query arg")
Use WorkflowHandle.query
to query a running or completed Workflow.
- TypeScript
- JavaScript
import { Client } from '@temporalio/client';
import { getValueQuery } from './workflows';
async function run(): Promise<void> {
const client = new Client();
const handle = client.workflow.getHandle('state-id-0');
const meaning = await handle.query(getValueQuery, 'meaning-of-life');
console.log({ meaning });
}
"use strict";
var __awaiter = (this && this.__awaiter) || function (thisArg, _arguments, P, generator) {
function adopt(value) { return value instanceof P ? value : new P(function (resolve) { resolve(value); }); }
return new (P || (P = Promise))(function (resolve, reject) {
function fulfilled(value) { try { step(generator.next(value)); } catch (e) { reject(e); } }
function rejected(value) { try { step(generator["throw"](value)); } catch (e) { reject(e); } }
function step(result) { result.done ? resolve(result.value) : adopt(result.value).then(fulfilled, rejected); }
step((generator = generator.apply(thisArg, _arguments || [])).next());
});
};
var __generator = (this && this.__generator) || function (thisArg, body) {
var _ = { label: 0, sent: function() { if (t[0] & 1) throw t[1]; return t[1]; }, trys: [], ops: [] }, f, y, t, g;
return g = { next: verb(0), "throw": verb(1), "return": verb(2) }, typeof Symbol === "function" && (g[Symbol.iterator] = function() { return this; }), g;
function verb(n) { return function (v) { return step([n, v]); }; }
function step(op) {
if (f) throw new TypeError("Generator is already executing.");
while (g && (g = 0, op[0] && (_ = 0)), _) try {
if (f = 1, y && (t = op[0] & 2 ? y["return"] : op[0] ? y["throw"] || ((t = y["return"]) && t.call(y), 0) : y.next) && !(t = t.call(y, op[1])).done) return t;
if (y = 0, t) op = [op[0] & 2, t.value];
switch (op[0]) {
case 0: case 1: t = op; break;
case 4: _.label++; return { value: op[1], done: false };
case 5: _.label++; y = op[1]; op = [0]; continue;
case 7: op = _.ops.pop(); _.trys.pop(); continue;
default:
if (!(t = _.trys, t = t.length > 0 && t[t.length - 1]) && (op[0] === 6 || op[0] === 2)) { _ = 0; continue; }
if (op[0] === 3 && (!t || (op[1] > t[0] && op[1] < t[3]))) { _.label = op[1]; break; }
if (op[0] === 6 && _.label < t[1]) { _.label = t[1]; t = op; break; }
if (t && _.label < t[2]) { _.label = t[2]; _.ops.push(op); break; }
if (t[2]) _.ops.pop();
_.trys.pop(); continue;
}
op = body.call(thisArg, _);
} catch (e) { op = [6, e]; y = 0; } finally { f = t = 0; }
if (op[0] & 5) throw op[1]; return { value: op[0] ? op[1] : void 0, done: true };
}
};
exports.__esModule = true;
// @ts-nocheck
var client_1 = require("@temporalio/client");
var workflows_1 = require("./workflows");
function run() {
return __awaiter(this, void 0, void 0, function () {
var client, handle, meaning;
return __generator(this, function (_a) {
switch (_a.label) {
case 0:
client = new client_1.Client();
handle = client.workflow.getHandle('state-id-0');
return [4 /*yield*/, handle.query(workflows_1.getValueQuery, 'meaning-of-life')];
case 1:
meaning = _a.sent();
console.log({ meaning: meaning });
return [2 /*return*/];
}
});
});
}
Workflow timeouts
Each Workflow timeout controls the maximum duration of a different aspect of a Workflow Execution.
Workflow timeouts are set when starting the Workflow ExecutionWorkflow timeouts
Each Workflow timeout controls the maximum duration of a different aspect of a Workflow Execution.
Learn more.
- Workflow Execution Timeout
What is a Workflow Execution Timeout?
A Workflow Execution Timeout is the maximum time that a Workflow Execution can be executing (have an Open status) including retries and any usage of Continue As New.
Learn more - restricts the maximum amount of time that a single Workflow Execution can be executed. - Workflow Run Timeout
What is a Workflow Run Timeout?
This is the maximum amount of time that a single Workflow Run is restricted to.
Learn more: restricts the maximum amount of time that a single Workflow Run can last. - Workflow Task Timeout
What is a Workflow Task Timeout?
A Workflow Task Timeout is the maximum amount of time that the Temporal Server will wait for a Worker to start processing a Workflow Task after the Task has been pulled from the Task Queue.
Learn more: restricts the maximum amount of time that a Worker can execute a Workflow Task.
- Go
- Java
- PHP
- Python
- TypeScript
Create an instance of StartWorkflowOptions
from the go.temporal.io/sdk/client
package, set a timeout, and pass the instance to the ExecuteWorkflow
call.
Available timeouts are:
WorkflowExecutionTimeout
WorkflowRunTimeout
WorkflowTaskTimeout
workflowOptions := client.StartWorkflowOptions{
// ...
// Set Workflow Timeout duration
WorkflowExecutionTimeout: time.Hours * 24 * 365 * 10,
// WorkflowRunTimeout: time.Hours * 24 * 365 * 10,
// WorkflowTaskTimeout: time.Second * 10,
// ...
}
workflowRun, err := c.ExecuteWorkflow(context.Background(), workflowOptions, YourWorkflowDefinition)
if err != nil {
// ...
}
Create an instance of WorkflowStub
in the Client code and set your timeout.
Available timeouts are:
//create Workflow stub for YourWorkflowInterface
YourWorkflowInterface workflow1 =
WorkerGreet.greetclient.newWorkflowStub(
GreetWorkflowInterface.class,
WorkflowOptions.newBuilder()
.setWorkflowId("YourWorkflow")
.setTaskQueue(WorkerGreet.TASK_QUEUE)
// Set Workflow Timeout duration
.setWorkflowExecutionTimeout(Duration.ofSeconds(10))
// .setWorkflowRunTimeout(Duration.ofSeconds(10))
// .setWorkflowTaskTimeout(Duration.ofSeconds(10))
.build());
Create an instance of WorkflowOptions
in the Client code and set your timeout.
Available timeouts are:
withWorkflowExecutionTimeout()
withWorkflowRunTimeout()
withWorkflowTaskTimeout()
$workflow = $this->workflowClient->newWorkflowStub(
DynamicSleepWorkflowInterface::class,
WorkflowOptions::new()
->withWorkflowId(DynamicSleepWorkflow::WORKFLOW_ID)
->withWorkflowIdReusePolicy(WorkflowIdReusePolicy::WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE)
// Set Workflow Timeout duration
->withWorkflowExecutionTimeout(CarbonInterval::minutes(2))
// ->withWorkflowRunTimeout(CarbonInterval::minute(2))
// ->withWorkflowTaskTimeout(CarbonInterval::minute(2))
);
Set the timeout from either the start_workflow()
or execute_workflow()
asynchronous methods.
Available timeouts are:
execution_timeout
run_timeout
task_timeout
handle = await client.start_workflow(
"your-workflow-name",
"some arg",
id="your-workflow-id",
task_queue="your-task-queue",
start_signal="your-signal-name",
# Set Workflow Timeout duration
execution_timeout=timedelta(seconds=2),
# run_timeout=timedelta(seconds=2),
# task_timeout=timedelta(seconds=2),
)
handle = await client.execute_workflow(
"your-workflow-name",
"some arg",
id="your-workflow-id",
task_queue="your-task-queue",
start_signal="your-signal-name",
# Set Workflow Timeout duration
execution_timeout=timedelta(seconds=2),
# run_timeout=timedelta(seconds=2),
# task_timeout=timedelta(seconds=2),
)
Create an instance of WorkflowOptions
from the Client and set your Workflow Timeout.
Available timeouts are:
- TypeScript
- JavaScript
await client.workflow.start(example, {
taskQueue,
workflowId,
workflowExecutionTimeout: '1 day',
});
await client.workflow.start(example, {
taskQueue: taskQueue,
workflowId: workflowId,
workflowExecutionTimeout: '1 day'
});
- TypeScript
- JavaScript
await client.workflow.start(example, {
taskQueue,
workflowId,
workflowRunTimeout: '1 minute',
});
await client.workflow.start(example, {
taskQueue: taskQueue,
workflowId: workflowId,
workflowRunTimeout: '1 minute'
});
- TypeScript
- JavaScript
await client.workflow.start(example, {
taskQueue,
workflowId,
workflowTaskTimeout: '1 minute',
});
await client.workflow.start(example, {
taskQueue: taskQueue,
workflowId: workflowId,
workflowTaskTimeout: '1 minute'
});
Workflow retries
A Retry Policy can work in cooperation with the timeouts to provide fine controls to optimize the execution experience.
Use a Retry PolicyWhat is a Retry Policy?
A Retry Policy is a collection of attributes that instructs the Temporal Server how to retry a failure of a Workflow Execution or an Activity Task Execution.
Learn more to retry a Workflow Execution in the event of a failure.
Workflow Executions do not retry by default, and Retry Policies should be used with Workflow Executions only in certain situations.
- Go
- Java
- PHP
- Python
- TypeScript
Create an instance of a RetryPolicy
from the go.temporal.io/sdk/temporal
package and provide it as the value to the RetryPolicy
field of the instance of StartWorkflowOptions
.
- Type:
RetryPolicy
- Default: None
retrypolicy := &temporal.RetryPolicy{
InitialInterval: time.Second,
BackoffCoefficient: 2.0,
MaximumInterval: time.Second * 100,
}
workflowOptions := client.StartWorkflowOptions{
RetryPolicy: retrypolicy,
// ...
}
workflowRun, err := temporalClient.ExecuteWorkflow(context.Background(), workflowOptions, YourWorkflowDefinition)
if err != nil {
// ...
}
To set a Workflow Retry Options in the WorkflowStub
instance use WorkflowOptions.Builder.setWorkflowRetryOptions
.
- Type:
RetryOptions
- Default:
Null
which means no retries will be attempted.
//create Workflow stub for GreetWorkflowInterface
GreetWorkflowInterface workflow1 =
WorkerGreet.greetclient.newWorkflowStub(
GreetWorkflowInterface.class,
WorkflowOptions.newBuilder()
.setWorkflowId("GreetWF")
.setTaskQueue(WorkerGreet.TASK_QUEUE)
// Set Workflow Retry Options
.setRetryOptions(RetryOptions.newBuilder()
.build());
A Retry Policy can be configured with an instance of the RetryOptions
object.
To enable retries for a Workflow, you need to provide a Retry Policy object via ChildWorkflowOptions
for Child Workflows or via WorkflowOptions
for top-level Workflows.
$workflow = $this->workflowClient->newWorkflowStub(
CronWorkflowInterface::class,
WorkflowOptions::new()->withRetryOptions(
RetryOptions::new()->withInitialInterval(120)
)
);
Set the Retry Policy from either the start_workflow()
or execute_workflow()
asynchronous methods.
handle = await client.start_workflow(
"your-workflow-name",
"some arg",
id="your-workflow-id",
task_queue="your-task-queue",
start_signal="your-signal-name",
retry_policy=RetryPolicy(maximum_interval=timedelta(seconds=2)),
)
handle = await client.execute_workflow(
"your-workflow-name",
"some arg",
id="your-workflow-id",
task_queue="your-task-queue",
start_signal="your-signal-name",
retry_policy=RetryPolicy(maximum_interval=timedelta(seconds=2)),
)
Create an instance of the Retry Policy, known as retry
in TypeScript, from the WorkflowOptions
of the Client interface.
- TypeScript
- JavaScript
const handle = await client.workflow.start(example, {
taskQueue,
workflowId,
retry: {
maximumAttempts: 3,
},
});
var handle = await client.workflow.start(example, {
taskQueue: taskQueue,
workflowId: workflowId,
retry: {
maximumAttempts: 3
}
});
Activity timeouts
Each Activity timeout controls the maximum duration of a different aspect of an Activity Execution.
The following timeouts are available in the Activity Options.
- Schedule-To-Close Timeout
What is a Schedule-To-Close Timeout?
A Schedule-To-Close Timeout is the maximum amount of time allowed for the overall Activity Execution, from when the first Activity Task is scheduled to when the last Activity Task, in the chain of Activity Tasks that make up the Activity Execution, reaches a Closed status.
Learn more: is the maximum amount of time allowed for the overall Activity ExecutionWhat is an Activity Execution?
An Activity Execution is the full chain of Activity Task Executions.
Learn more. - Start-To-Close Timeout
What is a Start-To-Close Timeout?
A Start-To-Close Timeout is the maximum time allowed for a single Activity Task Execution.
Learn more: is the maximum time allowed for a single Activity Task ExecutionWhat is an Activity Task Execution?
An Activity Task Execution occurs when a Worker uses the context provided from the Activity Task and executes the Activity Definition.
Learn more. - Schedule-To-Start Timeout
What is a Schedule-To-Start Timeout?
A Schedule-To-Start Timeout is the maximum amount of time that is allowed from when an Activity Task is placed in a Task Queue to when a Worker picks it up from the Task Queue.
Learn more: is the maximum amount of time that is allowed from when an Activity TaskWhat is an Activity Task?
An Activity Task contains the context needed to make an Activity Task Execution.
Learn more is scheduled to when a WorkerWhat is a Worker?
In day-to-day conversations, the term Worker is used to denote both a Worker Program and a Worker Process. Temporal documentation aims to be explicit and differentiate between them.
Learn more starts that Activity Task.
An Activity Execution must have either the Start-To-Close or the Schedule-To-Close Timeout set.
- Go
- Java
- PHP
- Python
- TypeScript
To set an Activity Timeout in Go, create an instance of ActivityOptions
from the go.temporal.io/sdk/workflow
package, set the Activity Timeout field, and then use the WithActivityOptions()
API to apply the options to the instance of workflow.Context
.
Available timeouts are:
StartToCloseTimeout
ScheduleToClose
ScheduleToStartTimeout
activityoptions := workflow.ActivityOptions{
// Set Activity Timeout duration
ScheduleToCloseTimeout: 10 * time.Second,
// StartToCloseTimeout: 10 * time.Second,
// ScheduleToStartTimeout: 10 * time.Second,
}
ctx = workflow.WithActivityOptions(ctx, activityoptions)
var yourActivityResult YourActivityResult
err = workflow.ExecuteActivity(ctx, YourActivityDefinition, yourActivityParam).Get(ctx, &yourActivityResult)
if err != nil {
// ...
}
Set your Activity Timeout from the ActivityOptions.Builder
class.
Available timeouts are:
- ScheduleToCloseTimeout()
- ScheduleToStartTimeout()
- StartToCloseTimeout()
You can set Activity Options using an ActivityStub
within a Workflow implementation, or per-Activity using WorkflowImplementationOptions
within a Worker.
The following uses ActivityStub
.
GreetingActivities activities = Workflow.newActivityStub(GreetingActivities.class,
ActivityOptions.newBuilder()
.setScheduleToCloseTimeout(Duration.ofSeconds(5))
// .setStartToCloseTimeout(Duration.ofSeconds(2)
// .setScheduletoCloseTimeout(Duration.ofSeconds(20))
.build());
The following uses WorkflowImplementationOptions
.
WorkflowImplementationOptions options =
WorkflowImplementationOptions.newBuilder()
.setActivityOptions(
ImmutableMap.of(
"GetCustomerGreeting",
// Set Activity Execution timeout
ActivityOptions.newBuilder()
.setScheduleToCloseTimeout(Duration.ofSeconds(5))
// .setStartToCloseTimeout(Duration.ofSeconds(2))
// .setScheduleToStartTimeout(Duration.ofSeconds(5))
.build()))
.build();
If you define options per-Activity Type options with WorkflowImplementationOptions.setActivityOptions()
, setting them again specifically with ActivityStub
in a Workflow will override this setting.
Because Activities are reentrant, only a single stub can be used for multiple Activity invocations.
Available timeouts are:
- withScheduleToCloseTimeout()
- withStartToCloseTimeout()
- withScheduleToStartTimeout()
$this->greetingActivity = Workflow::newActivityStub(
GreetingActivityInterface::class,
// Set Activity Timeout duration
ActivityOptions::new()
->withScheduleToCloseTimeout(CarbonInterval::seconds(2))
// ->withStartToCloseTimeout(CarbonInterval::seconds(2))
// ->withScheduleToStartTimeout(CarbonInterval::seconds(10))
);
Activity options are set as keyword arguments after the Activity arguments.
Available timeouts are:
- schedule_to_close_timeout
- schedule_to_start_timeout
- start_to_close_timeout
@workflow.defn
class YourWorkflow:
@workflow.run
async def run(self, name: str) -> str:
return await workflow.execute_activity(
your_activity,
name,
schedule_to_close_timeout=timedelta(seconds=5),
# schedule_to_start_timeout=timedelta(seconds=5),
# start_to_close_timeout=timedelta(seconds=5),
)
When you call proxyActivities
in a Workflow Function, you can set a range of ActivityOptions
.
Available timeouts are:
// Sample of typical options you can set
const { greet } = proxyActivities<typeof activities>({
scheduleToCloseTimeout: '5m',
// startToCloseTimeout: "30s", // recommended
// scheduleToStartTimeout: "60s",
retry: {
// default retry policy if not specified
initialInterval: '1s',
backoffCoefficient: 2,
maximumAttempts: Infinity,
maximumInterval: 100 * initialInterval,
nonRetryableErrorTypes: [],
},
});
Activity retries
A Retry Policy works in cooperation with the timeouts to provide fine controls to optimize the execution experience.
Activity Executions are automatically associated with a default Retry PolicyWhat is a Retry Policy?
A Retry Policy is a collection of attributes that instructs the Temporal Server how to retry a failure of a Workflow Execution or an Activity Task Execution.
Learn more if a custom one is not provided.
- Go
- Java
- PHP
- Python
- TypeScript
To set a RetryPolicyWhat is a Retry Policy?
A Retry Policy is a collection of attributes that instructs the Temporal Server how to retry a failure of a Workflow Execution or an Activity Task Execution.
Learn more, create an instance of ActivityOptions
from the go.temporal.io/sdk/workflow
package, set the RetryPolicy
field, and then use the WithActivityOptions()
API to apply the options to the instance of workflow.Context
.
- Type:
RetryPolicy
- Default:
retrypolicy := &temporal.RetryPolicy{
InitialInterval: time.Second,
BackoffCoefficient: 2.0,
MaximumInterval: time.Second * 100, // 100 * InitialInterval
MaximumAttempts: 0, // Unlimited
NonRetryableErrorTypes: []string, // empty
}
Providing a Retry Policy here is a customization, and overwrites individual Field defaults.
retrypolicy := &temporal.RetryPolicy{
InitialInterval: time.Second,
BackoffCoefficient: 2.0,
MaximumInterval: time.Second * 100,
}
activityoptions := workflow.ActivityOptions{
RetryPolicy: retrypolicy,
}
ctx = workflow.WithActivityOptions(ctx, activityoptions)
var yourActivityResult YourActivityResult
err = workflow.ExecuteActivity(ctx, YourActivityDefinition, yourActivityParam).Get(ctx, &yourActivityResult)
if err != nil {
// ...
}
To set a Retry Policy, known as the Retry OptionsWhat is a Retry Policy?
A Retry Policy is a collection of attributes that instructs the Temporal Server how to retry a failure of a Workflow Execution or an Activity Task Execution.
Learn more in Java, use ActivityOptions.newBuilder.setRetryOptions()
.
Type:
RetryOptions
Default: Server-defined Activity Retry policy.
With
ActivityStub
private final ActivityOptions options =
ActivityOptions.newBuilder()
// note that either StartToCloseTimeout or ScheduleToCloseTimeout are
// required when setting Activity options.
.setStartToCloseTimeout(Duration.ofSeconds(5))
.setRetryOptions(
RetryOptions.newBuilder()
.setInitialInterval(Duration.ofSeconds(1))
.setMaximumInterval(Duration.ofSeconds(10))
.build())
.build();With
WorkflowImplementationOptions
WorkflowImplementationOptions options =
WorkflowImplementationOptions.newBuilder()
.setActivityOptions(
ImmutableMap.of(
"EmailCustomerGreeting",
ActivityOptions.newBuilder()
// note that either StartToCloseTimeout or ScheduleToCloseTimeout are
// required when setting Activity options.
.setStartToCloseTimeout(Duration.ofSeconds(5))
.setRetryOptions(
RetryOptions.newBuilder()
.setDoNotRetry(NullPointerException.class.getName())
.build())
.build()))
.build();
To set an Activity Retry, set {@link RetryOptions}
on {@link ActivityOptions}
.
The follow example creates a new Activity with the given options.
$this->greetingActivity = Workflow::newActivityStub(
GreetingActivityInterface::class,
ActivityOptions::new()
->withScheduleToCloseTimeout(CarbonInterval::seconds(10))
->withRetryOptions(
RetryOptions::new()
->withInitialInterval(CarbonInterval::seconds(1))
->withMaximumAttempts(5)
->withNonRetryableExceptions([\InvalidArgumentException::class])
)
);
}
For an executable code sample, see ActivityRetry sample in the PHP samples repository.
To create an Activity Retry Policy in Python, set the RetryPolicy class within the start_activity()
or execute_activity()
function.
The following example sets the maximum interval to 2 seconds.
workflow.execute_activity(
your_activity,
name,
start_to_close_timeout=timedelta(seconds=10),
retry_policy=RetryPolicy(maximum_interval=timedelta(seconds=2)),
)
To set Activity Retry Policies in TypeScript, pass ActivityOptions.retry
to proxyActivities
.
// Sample of typical options you can set
const { yourActivity } = proxyActivities<typeof activities>({
// ...
retry: {
// default retry policy if not specified
initialInterval: '1s',
backoffCoefficient: 2,
maximumAttempts: Infinity,
maximumInterval: 100 * initialInterval,
nonRetryableErrorTypes: [],
},
});
Activity retry simulator
Use this tool to visualize total Activity Execution times and experiment with different Activity timeouts and Retry Policies.
The simulator is based on a common Activity use-case, which is to call a third party HTTP API and return the results. See the example code snippets below.
Use the Activity Retries settings to configure how long the API request takes to succeed or fail. There is an option to generate scenarios. The Task Time in Queue simulates the time the Activity Task might be waiting in the Task Queue.
Use the Activity Timeouts and Retry Policy settings to see how they impact the success or failure of an Activity Execution.
Sample Activity
import axios from 'axios';
async function testActivity(url: string): Promise<void> {
await axios.get(url);
}
export default testActivity;
Activity Retries (in ms)
Activity Timeouts (in ms)
Retry Policy (in ms)
Success after 1 ms
{
"startToCloseTimeout": 10000,
"retryPolicy": {
"backoffCoefficient": 2,
"initialInterval": 1000
}
}
Activity Heartbeats
An Activity HeartbeatWhat is an Activity Heartbeat?
An Activity Heartbeat is a ping from the Worker that is executing the Activity to the Temporal Cluster. Each ping informs the Temporal Cluster that the Activity Execution is making progress and the Worker has not crashed.
Learn more is a ping from the Worker ProcessWhat is a Worker Process?
A Worker Process is responsible for polling a Task Queue, dequeueing a Task, executing your code in response to a Task, and responding to the Temporal Server with the results.
Learn more that is executing the Activity to the Temporal ClusterWhat is a Temporal Cluster?
A Temporal Cluster is the Temporal Server paired with persistence.
Learn more.
Each Heartbeat informs the Temporal Cluster that the Activity ExecutionWhat is an Activity Execution?
An Activity Execution is the full chain of Activity Task Executions.
Learn more is making progress and the Worker has not crashed.
If the Cluster does not receive a Heartbeat within a Heartbeat TimeoutWhat is a Heartbeat Timeout?
A Heartbeat Timeout is the maximum time between Activity Heartbeats.
Learn more time period, the Activity will be considered failed and another Activity Task ExecutionWhat is an Activity Task Execution?
An Activity Task Execution occurs when a Worker uses the context provided from the Activity Task and executes the Activity Definition.
Learn more may be scheduled according to the Retry Policy.
Heartbeats may not always be sent to the Cluster—they may be throttledWhat is an Activity Heartbeat?
An Activity Heartbeat is a ping from the Worker that is executing the Activity to the Temporal Cluster. Each ping informs the Temporal Cluster that the Activity Execution is making progress and the Worker has not crashed.
Learn more by the Worker.
Activity Cancellations are delivered to Activities from the Cluster when they Heartbeat. Activities that don't Heartbeat can't receive a Cancellation. Heartbeat throttling may lead to Cancellation getting delivered later than expected.
Heartbeats can contain a details
field describing the Activity's current progress.
If an Activity gets retried, the Activity can access the details
from the last Heartbeat that was sent to the Cluster.
- Go
- Java
- PHP
- Python
- TypeScript
To HeartbeatWhat is an Activity Heartbeat?
An Activity Heartbeat is a ping from the Worker that is executing the Activity to the Temporal Cluster. Each ping informs the Temporal Cluster that the Activity Execution is making progress and the Worker has not crashed.
Learn more in an Activity in Go, use the RecordHeartbeat
API.
import (
// ...
"go.temporal.io/sdk/workflow"
// ...
)
func YourActivityDefinition(ctx, YourActivityDefinitionParam) (YourActivityDefinitionResult, error) {
// ...
activity.RecordHeartbeat(ctx, details)
// ...
}
When an Activity Task Execution times out due to a missed Heartbeat, the last value of the details
variable above is returned to the calling Workflow in the details
field of TimeoutError
with TimeoutType
set to Heartbeat
.
You can also Heartbeat an Activity from an external source:
// The client is a heavyweight object that should be created once per process.
temporalClient, err := client.Dial(client.Options{})
// Record heartbeat.
err := temporalClient.RecordActivityHeartbeat(ctx, taskToken, details)
The parameters of the RecordActivityHeartbeat
function are:
taskToken
: The value of the binaryTaskToken
field of theActivityInfo
struct retrieved inside the Activity.details
: The serializable payload containing progress information.
If an Activity Execution Heartbeats its progress before it failed, the retry attempt will have access to the progress information, so that the Activity Execution can resume from the failed state. Here's an example of how this can be implemented:
func SampleActivity(ctx context.Context, inputArg InputParams) error {
startIdx := inputArg.StartIndex
if activity.HasHeartbeatDetails(ctx) {
// Recover from finished progress.
var finishedIndex int
if err := activity.GetHeartbeatDetails(ctx, &finishedIndex); err == nil {
startIdx = finishedIndex + 1 // Start from next one.
}
}
// Normal Activity logic...
for i:=startIdx; i<inputArg.EndIdx; i++ {
// Code for processing item i goes here...
activity.RecordHeartbeat(ctx, i) // Report progress.
}
}
To Heartbeat an Activity Execution in Java, use the Activity.getExecutionContext().heartbeat()
Class method.
public class YourActivityDefinitionImpl implements YourActivityDefinition {
@Override
public String yourActivityMethod(YourActivityMethodParam param) {
// ...
Activity.getExecutionContext().heartbeat(details);
// ...
}
// ...
}
The method takes an optional argument, the details
variable above that represents latest progress of the Activity Execution.
This method can take a variety of types such as an exception object, custom object, or string.
If the Activity Execution times out, the last Heartbeat details
are included in the thrown ActivityTimeoutException
, which can be caught by the calling Workflow.
The Workflow can then use the details
information to pass to the next Activity invocation if needed.
In the case of Activity retries, the last Heartbeat's details
are available and can be extracted from the last failed attempt by using Activity.getExecutionContext().getHeartbeatDetails(Class<V> detailsClass)
Some Activities are long-running.
To react to a crash quickly, use the Heartbeat mechanism, Activity::heartbeat()
, which lets the Temporal Server know that the Activity is still alive.
This acts as a periodic checkpoint mechanism for the progress of an Activity.
You can piggyback details
on an Activity Heartbeat.
If an Activity times out, the last value of details
is included in the TimeoutFailure
delivered to a Workflow.
Then the Workflow can pass the details to the next Activity invocation.
Additionally, you can access the details from within an Activity via Activity::getHeartbeatDetails
.
When an Activity is retried after a failure getHeartbeatDetails
enables you to get the value from the last successful Heartbeat.
use Temporal\Activity;
class FileProcessingActivitiesImpl implements FileProcessingActivities
{
// ...
public function download(
string $bucketName,
string $remoteName,
string $localName
): void
{
$this->dowloader->downloadWithProgress(
$bucketName,
$remoteName,
$localName,
// on progress
function ($progress) {
Activity::heartbeat($progress);
}
);
Activity::heartbeat(100); // download complete
// ...
}
// ...
}
To Heartbeat an Activity Execution in Python, use the heartbeat()
API.
@activity.defn
async def your_activity_definition() -> str:
activity.heartbeat("heartbeat details!")
In addition to obtaining cancellation information, Heartbeats also support detail data that persists on the server for retrieval during Activity retry.
If an Activity calls heartbeat(123, 456)
and then fails and is retried, heartbeat_details
returns an iterable containing 123
and 456
on the next Run.
Long-running Activities should Heartbeat their progress back to the Workflow for earlier detection of stalled Activities (with Heartbeat TimeoutWhat is a Heartbeat Timeout?
A Heartbeat Timeout is the maximum time between Activity Heartbeats.
Learn more) and resuming stalled Activities from checkpoints (with Heartbeat details).
To set Activity Heartbeat, use Context.current().heartbeat()
in your Activity implementation, and set heartbeatTimeout
in your Workflow.
- TypeScript
- JavaScript
// activity implementation
export async function example(sleepIntervalMs = 1000): Promise<void> {
for (let progress = 1; progress <= 1000; ++progress) {
await Context.current().sleep(sleepIntervalMs);
// record activity heartbeat
Context.current().heartbeat();
}
}
// ...
// workflow code calling activity
const { example } = proxyActivities<typeof activities>({
startToCloseTimeout: '1 hour',
heartbeatTimeout: '10s',
});
"use strict";
var __awaiter = (this && this.__awaiter) || function (thisArg, _arguments, P, generator) {
function adopt(value) { return value instanceof P ? value : new P(function (resolve) { resolve(value); }); }
return new (P || (P = Promise))(function (resolve, reject) {
function fulfilled(value) { try { step(generator.next(value)); } catch (e) { reject(e); } }
function rejected(value) { try { step(generator["throw"](value)); } catch (e) { reject(e); } }
function step(result) { result.done ? resolve(result.value) : adopt(result.value).then(fulfilled, rejected); }
step((generator = generator.apply(thisArg, _arguments || [])).next());
});
};
var __generator = (this && this.__generator) || function (thisArg, body) {
var _ = { label: 0, sent: function() { if (t[0] & 1) throw t[1]; return t[1]; }, trys: [], ops: [] }, f, y, t, g;
return g = { next: verb(0), "throw": verb(1), "return": verb(2) }, typeof Symbol === "function" && (g[Symbol.iterator] = function() { return this; }), g;
function verb(n) { return function (v) { return step([n, v]); }; }
function step(op) {
if (f) throw new TypeError("Generator is already executing.");
while (g && (g = 0, op[0] && (_ = 0)), _) try {
if (f = 1, y && (t = op[0] & 2 ? y["return"] : op[0] ? y["throw"] || ((t = y["return"]) && t.call(y), 0) : y.next) && !(t = t.call(y, op[1])).done) return t;
if (y = 0, t) op = [op[0] & 2, t.value];
switch (op[0]) {
case 0: case 1: t = op; break;
case 4: _.label++; return { value: op[1], done: false };
case 5: _.label++; y = op[1]; op = [0]; continue;
case 7: op = _.ops.pop(); _.trys.pop(); continue;
default:
if (!(t = _.trys, t = t.length > 0 && t[t.length - 1]) && (op[0] === 6 || op[0] === 2)) { _ = 0; continue; }
if (op[0] === 3 && (!t || (op[1] > t[0] && op[1] < t[3]))) { _.label = op[1]; break; }
if (op[0] === 6 && _.label < t[1]) { _.label = t[1]; t = op; break; }
if (t && _.label < t[2]) { _.label = t[2]; _.ops.push(op); break; }
if (t[2]) _.ops.pop();
_.trys.pop(); continue;
}
op = body.call(thisArg, _);
} catch (e) { op = [6, e]; y = 0; } finally { f = t = 0; }
if (op[0] & 5) throw op[1]; return { value: op[0] ? op[1] : void 0, done: true };
}
};
exports.__esModule = true;
exports.example = void 0;
// @ts-nocheck
// activity implementation
function example(sleepIntervalMs) {
if (sleepIntervalMs === void 0) { sleepIntervalMs = 1000; }
return __awaiter(this, void 0, void 0, function () {
var progress;
return __generator(this, function (_a) {
switch (_a.label) {
case 0:
progress = 1;
_a.label = 1;
case 1:
if (!(progress <= 1000)) return [3 /*break*/, 4];
return [4 /*yield*/, Context.current().sleep(sleepIntervalMs)];
case 2:
_a.sent();
// record activity heartbeat
Context.current().heartbeat();
_a.label = 3;
case 3:
++progress;
return [3 /*break*/, 1];
case 4: return [2 /*return*/];
}
});
});
}
exports.example = example;
// ...
// workflow code calling activity
var example = proxyActivities({
startToCloseTimeout: '1 hour',
heartbeatTimeout: '10s'
}).example;
In the previous example, setting the Heartbeat informs the Temporal Server of the Activity's progress at regular intervals.
If the Activity stalls or the Activity Worker becomes unavailable, the absence of Heartbeats prompts the Temporal Server to retry the Activity immediately, without waiting for startToCloseTimeout
to complete.
You can also add heartbeatDetails
as a checkpoint to collect data about failures during the execution, and use it to resume the Activity from that point.
The following example extends the previous sample to include a heartbeatDetails
checkpoint.
- TypeScript
- JavaScript
export async function example(sleepIntervalMs = 1000): Promise<void> {
const startingPoint = Context.current().info.heartbeatDetails || 1; // allow for resuming from heartbeat
for (let progress = startingPoint; progress <= 100; ++progress) {
await Context.current().sleep(sleepIntervalMs);
Context.current().heartbeat(progress);
}
}
"use strict";
var __awaiter = (this && this.__awaiter) || function (thisArg, _arguments, P, generator) {
function adopt(value) { return value instanceof P ? value : new P(function (resolve) { resolve(value); }); }
return new (P || (P = Promise))(function (resolve, reject) {
function fulfilled(value) { try { step(generator.next(value)); } catch (e) { reject(e); } }
function rejected(value) { try { step(generator["throw"](value)); } catch (e) { reject(e); } }
function step(result) { result.done ? resolve(result.value) : adopt(result.value).then(fulfilled, rejected); }
step((generator = generator.apply(thisArg, _arguments || [])).next());
});
};
var __generator = (this && this.__generator) || function (thisArg, body) {
var _ = { label: 0, sent: function() { if (t[0] & 1) throw t[1]; return t[1]; }, trys: [], ops: [] }, f, y, t, g;
return g = { next: verb(0), "throw": verb(1), "return": verb(2) }, typeof Symbol === "function" && (g[Symbol.iterator] = function() { return this; }), g;
function verb(n) { return function (v) { return step([n, v]); }; }
function step(op) {
if (f) throw new TypeError("Generator is already executing.");
while (g && (g = 0, op[0] && (_ = 0)), _) try {
if (f = 1, y && (t = op[0] & 2 ? y["return"] : op[0] ? y["throw"] || ((t = y["return"]) && t.call(y), 0) : y.next) && !(t = t.call(y, op[1