Skip to main content

Namespace: worker

The temporal worker connects to the service and runs workflows and activities.

Usage​

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

async function run() {
  const connection = await NativeConnection.connect({
    address: 'localhost:7233',
  });
  try {
    const worker = await Worker.create({
      connection,
      namespace: 'default',
      taskQueue: 'sleep-for-days',
      workflowsPath: require.resolve('./workflows'),
      activities,
    });
    await worker.run();
  } finally {
    // Close the connection once the worker has stopped
    await connection.close();
  }
}

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

Classes​

Interfaces​

References​

DataConverter​

Re-exports DataConverter

Headers​

Re-exports Headers

IllegalStateError​

Re-exports IllegalStateError

LogLevel​

Re-exports LogLevel

LogMetadata​

Re-exports LogMetadata

Logger​

Re-exports Logger

LoggerSinks​

Re-exports LoggerSinks

MetricsExporter​

Renames and re-exports MetricsExporterConfig

Next​

Re-exports Next

defaultPayloadConverter​

Re-exports defaultPayloadConverter

Type Aliases​

ActivityInterceptorsFactory​

Ƭ ActivityInterceptorsFactory: (ctx: Context) => ActivityInterceptors

Instantiate Activity Interceptors for a given Temporal Activity execution.

This factory function is called for each Activity execution by this Worker, receiving the corresponding ActivityContext.

Note that there's no strict requirement to have different instances of Activity Interceptors for each activity execution. It is acceptable, and often sufficient, for this function to return the same instance of Activity Interceptors for all activity executions, ignoring the ctx parameter. Interceptor functions may then rely on the Context.current to access the Activity Context where needed.

Type declaration​

▸ (ctx): ActivityInterceptors

Parameters​
NameType
ctxContext
Returns​

ActivityInterceptors

CompiledWorkerInterceptors​

Ƭ CompiledWorkerInterceptors: Required<Pick<WorkerInterceptors, "client" | "activity" | "nexus" | "workflowModules">>

GetLogAttributesInput​

Ƭ GetLogAttributesInput: Record<string, unknown>

Input for ActivityOutboundCallsInterceptor.getLogAttributes

GetMetricTagsInput​

Ƭ GetMetricTagsInput: MetricTags

Input for ActivityOutboundCallsInterceptor.getMetricTags

GrpcCompressionConfig​

Ƭ GrpcCompressionConfig: GzipGrpcCompressionConfig | NoneGrpcCompressionConfig

Transport-level gRPC compression configuration.

History​

Ƭ History: IHistory

InjectedSink​

Ƭ InjectedSink<T>: { [P in keyof T]: InjectedSinkFunction<T[P]> }

Converts a Sink from a mapping of name to function to a mapping of name to InjectedSinkFunction

Type parameters​

NameType
Textends Sink

InjectedSinks​

Ƭ InjectedSinks<T>: { [P in keyof T]: InjectedSink<T[P]> }

Converts a Sinks interface from a mapping of name to Sink to a mapping of name to InjectedSink.

Used for type checking Sink injection against supplied type param T.

Type parameters​

NameType
Textends Sinks

LogExporterConfig​

Ƭ LogExporterConfig: { filter?: string | CoreLogFilterOptions } & Partial<ConsoleLogger | ForwardLogger>

Configuration for logs emitted by the native side of the SDK.

See

TelemetryOptions.logging

MetricsExporterConfig​

Ƭ MetricsExporterConfig: { attachServiceName?: boolean ; globalTags?: Record<string, string> ; metricPrefix?: string ; temporality?: "cumulative" | "delta" } & PrometheusMetricsExporter | OtelCollectorExporter | BufferedMetricsExporter

Configuration for exporting metrics emitted by Core.

NexusInboundCallsInterceptor​

Ƭ NexusInboundCallsInterceptor: Object

Type declaration​

NameType
cancelOperation?(input: NexusCancelOperationInput, next: Next<NexusInboundCallsInterceptor, "cancelOperation">) => Promise<void>
startOperation?(input: NexusStartOperationInput, next: Next<NexusInboundCallsInterceptor, "startOperation">) => Promise<NexusStartOperationOutput>

NexusInterceptors​

Ƭ NexusInterceptors: Object

Interceptors for Nexus Operation execution.

Type declaration​

NameType
inbound?NexusInboundCallsInterceptor
outbound?NexusOutboundCallsInterceptor

NexusInterceptorsFactory​

Ƭ NexusInterceptorsFactory: (ctx: nexus.OperationContext) => NexusInterceptors

Instantiate Nexus Interceptors for a given Nexus Operation execution.

This factory function is called for each Nexus Operation executed by this Worker, receiving the corresponding nexus.OperationContext | OperationContext.

Note that there's no strict requirement to have different instances of Nexus Interceptors for each Nexus Operation execution. It is acceptable and often sufficient for this function to return the same instance of Nexus Interceptors for all Nexus Operation executions, ignoring the ctx parameter. Interceptor functions may then rely on APIs provided by the @temporalio/nexus package to access the Nexus Operation Context where needed.

Type declaration​

▸ (ctx): NexusInterceptors

Parameters​
NameType
ctxnexus.OperationContext
Returns​

NexusInterceptors

NexusOutboundCallsInterceptor​

Ƭ NexusOutboundCallsInterceptor: Object

Type declaration​

NameType
getLogAttributes?(input: GetLogAttributesInput, next: Next<NexusOutboundCallsInterceptor, "getLogAttributes">) => Record<string, unknown>
getMetricTags?(input: GetMetricTagsInput, next: Next<NexusOutboundCallsInterceptor, "getMetricTags">) => MetricTags

PatchActivationCallback​

Ƭ PatchActivationCallback: (input: PatchActivationInput) => boolean

Type declaration​

▸ (input): boolean

Parameters​
NameType
inputPatchActivationInput
Returns​

boolean

ReplayHistoriesIterable​

Ƭ ReplayHistoriesIterable: AsyncIterable<HistoryAndWorkflowId> | Iterable<HistoryAndWorkflowId>

An iterable on workflow histories and their IDs, used for batch replaying.

ResourceBasedSlotsForType​

Ƭ ResourceBasedSlotsForType: ResourceBasedSlotOptions & ResourceBasedTunerConfig & { type: "resource-based" }

Resource based slot supplier options for a specific kind of slot.

SlotInfo​

Ƭ SlotInfo: WorkflowSlotInfo | ActivitySlotInfo | LocalActivitySlotInfo | NexusSlotInfo

SlotSupplier​

Ƭ SlotSupplier<SI>: ResourceBasedSlotsForType | FixedSizeSlotSupplier | CustomSlotSupplier<SI>

Controls how slots are handed out for a specific task type.

Type parameters​

NameType
SIextends SlotInfo

State​

Ƭ State: "INITIALIZED" | "RUNNING" | "STOPPED" | "STOPPING" | "DRAINING" | "DRAINED" | "FAILED"

The worker's possible states

  • INITIALIZED - The initial state of the Worker after calling Worker.create and successful connection to the server
  • RUNNING - Worker.run was called, polling task queues
  • STOPPING - Worker.shutdown was called or received shutdown signal, worker will forcefully shutdown in shutdownGraceTime
  • DRAINING - Core has indicated that shutdown is complete and all Workflow tasks have been drained, waiting for activities and cached workflows eviction
  • DRAINED - All activities and workflows have completed, ready to shutdown
  • STOPPED - Shutdown complete, Worker.run resolves
  • FAILED - Worker encountered an unrecoverable error, Worker.run should reject with the error

WorkerDeploymentOptions​

Ƭ WorkerDeploymentOptions: { defaultVersioningBehavior?: VersioningBehavior ; useWorkerVersioning: boolean ; version: WorkerDeploymentVersion } & { defaultVersioningBehavior: VersioningBehavior ; useWorkerVersioning: true } | { defaultVersioningBehavior?: never ; useWorkerVersioning: false }

Allows specifying the deployment version of the worker and whether to use deployment-based worker versioning.

WorkerTuner​

Ƭ WorkerTuner: ResourceBasedTuner | ResourceBasedTunerWithController | TunerHolder

A worker tuner allows the customization of the performance characteristics of workers by controlling how "slots" are handed out for different task types. In order to poll for and then run tasks, a slot must first be reserved by the SlotSupplier returned by the tuner.

WorkflowBundleOption​

Ƭ WorkflowBundleOption: WorkflowBundle | WorkflowBundleWithSourceMap | WorkflowBundlePath | WorkflowBundlePathWithSourceMap

Variables​

LogTimestamp​

• Const LogTimestamp: unique symbol

ShutdownError​

• Const ShutdownError: typeof bridgeErrors.ShutdownError

TransportError​

• Const TransportError: typeof bridgeErrors.TransportError

UnexpectedError​

• Const UnexpectedError: typeof bridgeErrors.UnexpectedError

WorkflowInboundLogInterceptor​

• Const WorkflowInboundLogInterceptor: typeof WorkflowLogInterceptor

Deprecated

WorkflowInboundLogInterceptor is deprecated. Workflow lifecycle events are now automatically logged by the SDK. To customize workflow log attributes, simply register a custom WorkflowInterceptors that intercepts the outbound.getLogAttributes() method.

defaultWorkflowInterceptorModules​

• Const defaultWorkflowInterceptorModules: never[]

Deprecated

Including defaultWorkflowInterceptorModules in BundlerOptions.workflowInterceptorModules is no longer required.

errors​

• Const errors: Object

Deprecated

Import error classes directly

Type declaration​

NameType
GracefulShutdownPeriodExpiredErrortypeof GracefulShutdownPeriodExpiredError
IllegalStateErrortypeof IllegalStateError
ShutdownErrortypeof bridgeErrors.ShutdownError
TransportErrortypeof bridgeErrors.TransportError
UnexpectedErrortypeof bridgeErrors.UnexpectedError

Functions​

activityLogAttributes​

▸ activityLogAttributes(info): Record<string, unknown>

Returns a map of attributes to be set on log messages for a given Activity

Parameters​

NameType
infoInfo

Returns​

Record<string, unknown>

appendDefaultInterceptors​

▸ appendDefaultInterceptors(interceptors, logger?): WorkerInterceptors

Appends the default Worker logging interceptors to given interceptor arrays.

Parameters​

NameTypeDescription
interceptorsWorkerInterceptors-
logger?Loggera Logger - defaults to the Runtime singleton logger.

Returns​

WorkerInterceptors

Deprecated

Calling appendDefaultInterceptors() is no longer required. To configure a custom logger, set the Runtime.logger property instead.

bundleWorkflowCode​

▸ bundleWorkflowCode(options): Promise<WorkflowBundleWithSourceMap>

Create a bundle to pass to WorkerOptions.workflowBundle. Helpful for reducing Worker startup time in production.

When using with Worker.runReplayHistory, make sure to pass the same interceptors and payload converter used when the history was generated.

Parameters​

NameType
optionsBundleOptions

Returns​

Promise<WorkflowBundleWithSourceMap>

defaultSinks​

▸ defaultSinks(logger?): InjectedSinks<LoggerSinks>

Build the sink used internally by the SDK to forwards log messages from the Workflow sandbox to an actual logger.

Parameters​

NameTypeDescription
logger?Loggera Logger - defaults to the Runtime singleton logger.

Returns​

InjectedSinks<LoggerSinks>

Deprecated

Calling defaultSink() is no longer required. To configure a custom logger, set the Runtime.logger property instead.

makeTelemetryFilterString​

▸ makeTelemetryFilterString(options): string

A helper to build a filter string for use in RuntimeOptions.telemetryOptions.tracingFilter.

Note that one may instead simply pass a CoreLogFilterOptions object directly to RuntimeOptions.telemetryOptions.logging.filter. This function may however still be useful in some particular use cases and will therefore be kept around.

Parameters​

NameType
optionsCoreLogFilterOptions

Returns​

string

startDebugReplayer​

▸ startDebugReplayer(options): void

Start a replayer for debugging purposes.

Use this method to integrate the replayer with external debuggers like the Temporal VS Code debbuger extension.

Parameters​

NameType
optionsReplayWorkerOptions

Returns​

void

workflowLogAttributes​

▸ workflowLogAttributes(info): Record<string, unknown>

Returns a map of attributes to be set by default on log messages for a given Workflow. Note that this function may be called from outside of the Workflow context (eg. by the worker itself).

Parameters​

NameType
infoWorkflowInfo

Returns​

Record<string, unknown>