Documentation
¶
Overview ¶
Package worker contains functions to manage lifecycle of a Temporal client side worker.
Index ¶
Constants ¶
const ( // BlockWorkflow is the default WorkflowPanicPolicy policy for handling workflow panics and detected non-determinism. // This option causes workflow to get stuck in the workflow task retry loop. // It is expected that after the problem is discovered and fixed the workflows are going to continue // without any additional manual intervention. BlockWorkflow = internal.BlockWorkflow // FailWorkflow WorkflowPanicPolicy immediately fails workflow execution if workflow code throws panic or // detects non-determinism. This feature is convenient during development. // WARNING: enabling this in production can cause all open workflows to fail on a single bug or bad deployment. FailWorkflow = internal.FailWorkflow )
Variables ¶
This section is empty.
Functions ¶
func EnableVerboseLogging ¶
func EnableVerboseLogging(enable bool)
EnableVerboseLogging enable or disable verbose logging of internal Temporal library components. Most customers don't need this feature, unless advised by the Temporal team member. Also there is no guarantee that this API is not going to change.
func SetBinaryChecksum ¶ added in v0.10.0
func SetBinaryChecksum(checksum string)
SetBinaryChecksum sets the identifier of the binary(aka BinaryChecksum). The identifier is mainly used in recording reset points when respondDecisionTaskCompleted. For each workflow, the very first decision completed by a binary will be associated as a auto-reset point for the binary. So that when a customer wants to mark the binary as bad, the workflow will be reset to that point -- which means workflow will forget all progress generated by the binary. On another hand, once the binary is marked as bad, the bad binary cannot poll decision and make any progress any more.
func SetStickyWorkflowCacheSize ¶ added in v0.7.1
func SetStickyWorkflowCacheSize(cacheSize int)
SetStickyWorkflowCacheSize sets the cache size for sticky workflow cache. Sticky workflow execution is the affinity between decision tasks of a specific workflow execution to a specific worker. The affinity is set if sticky execution is enabled via Worker.Options (It is enabled by default unless disabled explicitly). The benefit of sticky execution is that workflow does not have to reconstruct the state by replaying from beginning of history events. But the cost is it consumes more memory as it rely on caching workflow execution's running state on the worker. The cache is shared between workers running within same process. This must be called before any worker is started. If not called, the default size of 10K (might change in future) will be used.
Types ¶
type Options ¶
type Options = internal.WorkerOptions
Options is used to configure a worker instance.
type Worker ¶
type Worker interface {
// RegisterWorkflow - registers a workflow function with the worker.
// A workflow takes a workflow.Context and input and returns a (result, error) or just error.
// Examples:
// func sampleWorkflow(ctx workflow.Context, input []byte) (result []byte, err error)
// func sampleWorkflow(ctx workflow.Context, arg1 int, arg2 string) (result []byte, err error)
// func sampleWorkflow(ctx workflow.Context) (result []byte, err error)
// func sampleWorkflow(ctx workflow.Context, arg1 int) (result string, err error)
// Serialization of all primitive types, structures is supported ... except channels, functions, variadic, unsafe pointer.
// For global registration consider workflow.Register
// This method panics if workflowFunc doesn't comply with the expected format or tries to register the same workflow
RegisterWorkflow(w interface{})
// RegisterWorkflowWithOptions registers the workflow function with options.
// The user can use options to provide an external name for the workflow or leave it empty if no
// external name is required. This can be used as
// worker.RegisterWorkflowWithOptions(sampleWorkflow, RegisterWorkflowOptions{})
// worker.RegisterWorkflowWithOptions(sampleWorkflow, RegisterWorkflowOptions{Name: "foo"})
// This method panics if workflowFunc doesn't comply with the expected format or tries to register the same workflow
// type name twice. Use workflow.RegisterOptions.DisableAlreadyRegisteredCheck to allow multiple registrations.
RegisterWorkflowWithOptions(w interface{}, options workflow.RegisterOptions)
// RegisterActivity - register an activity function or a pointer to a structure with the worker.
// An activity function takes a context and input and returns a (result, error) or just error.
//
// And activity struct is a structure with all its exported methods treated as activities. The default
// name of each activity is the method name.
//
// Examples:
// func sampleActivity(ctx context.Context, input []byte) (result []byte, err error)
// func sampleActivity(ctx context.Context, arg1 int, arg2 string) (result *customerStruct, err error)
// func sampleActivity(ctx context.Context) (err error)
// func sampleActivity() (result string, err error)
// func sampleActivity(arg1 bool) (result int, err error)
// func sampleActivity(arg1 bool) (err error)
//
// type Activities struct {
// // fields
// }
// func (a *Activities) SampleActivity1(ctx context.Context, arg1 int, arg2 string) (result *customerStruct, err error) {
// ...
// }
//
// func (a *Activities) SampleActivity2(ctx context.Context, arg1 int, arg2 *customerStruct) (result string, err error) {
// ...
// }
//
// Serialization of all primitive types, structures is supported ... except channels, functions, variadic, unsafe pointer.
// This method panics if activityFunc doesn't comply with the expected format or an activity with the same
// type name is registered more than once.
// For global registration consider activity.Register
RegisterActivity(a interface{})
// RegisterActivityWithOptions registers the activity function or struct pointer with options.
// The user can use options to provide an external name for the activity or leave it empty if no
// external name is required. This can be used as
// worker.RegisterActivityWithOptions(barActivity, RegisterActivityOptions{})
// worker.RegisterActivityWithOptions(barActivity, RegisterActivityOptions{Name: "barExternal"})
// When registering the structure that implements activities the name is used as a prefix that is
// prepended to the activity method name.
// worker.RegisterActivityWithOptions(&Activities{ ... }, RegisterActivityOptions{Name: "MyActivities_"})
// To override each name of activities defined through a structure register the methods one by one:
// activities := &Activities{ ... }
// worker.RegisterActivityWithOptions(activities.SampleActivity1, RegisterActivityOptions{Name: "Sample1"})
// worker.RegisterActivityWithOptions(activities.SampleActivity2, RegisterActivityOptions{Name: "Sample2"})
// See RegisterActivity function for more info.
// The other use of options is to disable duplicated activity registration check
// which might be useful for integration tests.
// worker.RegisterActivityWithOptions(barActivity, RegisterActivityOptions{DisableAlreadyRegisteredCheck: true})
RegisterActivityWithOptions(a interface{}, options activity.RegisterOptions)
// Start the worker in a non-blocking fashion.
Start() error
// Run the worker in a blocking fashion. Stop the worker when process is killed with SIGINT or SIGTERM.
// Returns error only if worker fails to start.
Run() error
// Stop the worker.
Stop()
}
Worker hosts workflow and activity implementations. Use worker.New(...) to create an instance.
func New ¶
New creates an instance of worker for managing workflow and activity executions.
namespace - the name of the temporal namespace
taskList - is the task list name you use to identify your client worker, also
identifies group of workflow and activity implementations that are
hosted by a single worker process
options - configure any worker specific options like logger, metrics, identity
type WorkflowPanicPolicy ¶ added in v0.22.0
type WorkflowPanicPolicy = internal.WorkflowPanicPolicy
WorkflowPanicPolicy is used for configuring how worker deals with workflow code panicking which includes non backwards compatible changes to the workflow code without appropriate versioning (see workflow.GetVersion). The default behavior is to block workflow execution until the problem is fixed.
type WorkflowReplayer ¶ added in v0.10.8
type WorkflowReplayer interface {
// RegisterWorkflow registers workflow that is going to be replayed
RegisterWorkflow(w interface{})
// RegisterWorkflowWithOptions registers workflow that is going to be replayed with user provided name
RegisterWorkflowWithOptions(w interface{}, options workflow.RegisterOptions)
// ReplayWorkflowHistory executes a single decision task for the given json history file.
// Use for testing the backwards compatibility of code changes and troubleshooting workflows in a debugger.
// The logger is an optional parameter. Defaults to the noop logger.
ReplayWorkflowHistory(logger *zap.Logger, history *historypb.History) error
// ReplayWorkflowHistoryFromJSONFile executes a single decision task for the json history file downloaded from the cli.
// To download the history file: temporal workflow showid <workflow_id> -of <output_filename>
// See https://github.com/temporalio/temporal/blob/master/tools/cli/README.md for full documentation
// Use for testing the backwards compatibility of code changes and troubleshooting workflows in a debugger.
// The logger is an optional parameter. Defaults to the noop logger.
ReplayWorkflowHistoryFromJSONFile(logger *zap.Logger, jsonfileName string) error
// ReplayPartialWorkflowHistoryFromJSONFile executes a single decision task for the json history file upto provided
// lastEventID(inclusive), downloaded from the cli.
// To download the history file: temporal workflow showid <workflow_id> -of <output_filename>
// See https://github.com/temporalio/temporal/blob/master/tools/cli/README.md for full documentation
// Use for testing the backwards compatibility of code changes and troubleshooting workflows in a debugger.
// The logger is an optional parameter. Defaults to the noop logger.
ReplayPartialWorkflowHistoryFromJSONFile(logger *zap.Logger, jsonfileName string, lastEventID int64) error
// ReplayWorkflowExecution loads a workflow execution history from the Temporal service and executes a single decision task for it.
// Use for testing the backwards compatibility of code changes and troubleshooting workflows in a debugger.
// The logger is the only optional parameter. Defaults to the noop logger.
ReplayWorkflowExecution(ctx context.Context, service workflowservice.WorkflowServiceClient, logger *zap.Logger, namespace string, execution workflow.Execution) error
}
WorkflowReplayer supports replaying a workflow from its event history. Use for troubleshooting and backwards compatibility unit tests. For example if a workflow failed in production then its history can be downloaded through UI or CLI and replayed in a debugger as many times as necessary. Use this class to create unit tests that check if workflow changes are backwards compatible. It is important to maintain backwards compatibility through use of workflow.GetVersion to ensure that new deployments are not going to break open workflows.
func NewWorkflowReplayer ¶ added in v0.10.8
func NewWorkflowReplayer() WorkflowReplayer
NewWorkflowReplayer creates a WorkflowReplayer instance.
Source Files