README
¶
Temporal drivers
Use this module to run Arkoren schedules and workflows on Temporal. Install it
with go get arkoren.dev/driver/temporal.
Register drivers before starting the shared host, and stop the host during application cleanup. NewFrom borrows its native client: the caller remains responsible for closing it. Worker callbacks must cooperate with cancellation; a cleanup deadline bounds the caller's wait, not arbitrary application code.
Contributor tests
From the repository root, run make check MODULE=driver/temporal and
make integration MODULE=driver/temporal. Integration starts a disposable local
development server and removes only its Compose project's containers and
volumes. Port 17233 must be free. The development server is not a production
deployment configuration.
Documentation
¶
Overview ¶
Package temporal holds the code shared by the Temporal drivers: the Host, which owns one Temporal client and one worker for a task queue.
Construction callbacks edit one fresh Builder, once in order; a supplied nil callback panics. Expanded nil or empty slices leave defaults unchanged. Published SDK records are value copies, not recursive native snapshots.
host := temporal.New(func(host *temporal.Builder) {
host.TaskQueue("orders")
host.NoWorker(true)
})
_ = host.Client() // Nil until Start.
A module root never holds a driver. Each driver, such as arkoren.dev/driver/temporal/temporalschedule, registers its workflows and activities with a Host through Host.Register before Host.Start runs, and reads the client through Host.Client.
New builds a Host that dials Temporal in Host.Start and closes the client in Host.Stop. NewFrom wraps a caller-owned client, which Host.Stop leaves open. Its applicable worker settings include NoWorker and WorkerStopTimeout, in addition to TaskQueue and WorkerOptions. Logger and connection settings are ignored by NewFrom: workers inherit the caller-owned client's logger. A nonzero native worker stop timeout still wins over the host timeout. A dialing host adapts slog.Default when neither host nor native client supplies a logger.
With no registrations, or with Builder.NoWorker, Host.Start creates no worker and the Host is only a client; a process that just starts runs uses this. A nil NewFrom client fails in Start. A canceled startup does not publish a started host. SDK worker startup cannot itself be interrupted: Start checks cancellation again after it returns and releases acquired resources on failure. Registration callbacks and SDK hooks must return for startup to finish.
Host.Stop initiates cleanup even with an expired context. It waits for the worker only until that context ends; owned client release follows completion of native worker cleanup, possibly later. Call Stop after Start returns. Builder.WorkerStopTimeout (ten seconds by default) is the native grace wait, potentially applied separately to internal workers, not a total shutdown budget. Expiry cancels activity contexts, not arbitrary application code. SDK shutdown hooks must return; callbacks that ignore cancellation can outlive cleanup. Use Stop's context to bound the caller's wait after startup. NoWorker defaults to false; a later false restores permission, but never forces a worker without registrations. WorkerDisabled reports the stored flag.
Index ¶
- Variables
- type Builder
- func (host *Builder) ClientOptions(options client.Options) *Builder
- func (host *Builder) HostPort(hostPort string) *Builder
- func (host *Builder) Logger(logger *slog.Logger) *Builder
- func (host *Builder) Namespace(namespace string) *Builder
- func (host *Builder) NoWorker(disabled bool) *Builder
- func (host *Builder) TaskQueue(taskQueue string) *Builder
- func (host *Builder) WorkerOptions(options worker.Options) *Builder
- func (host *Builder) WorkerStopTimeout(timeout time.Duration) *Builder
- type BuilderFunc
- type Host
- func (host *Host) Client() client.Client
- func (host *Host) Ping(ctx context.Context) error
- func (host *Host) Register(fn func(worker.Registry)) error
- func (host *Host) Start(ctx context.Context) error
- func (host *Host) Stop(ctx context.Context) error
- func (host *Host) TaskQueue() string
- func (host *Host) WorkerDisabled() bool
Constants ¶
This section is empty.
Variables ¶
var ErrStarted = errors.New("temporal host is started")
ErrStarted reports a registration attempted after Host.Start. Successful startup closes registration; failed startup leaves it available.
Example:
if errors.Is(err, ErrStarted) {
return fmt.Errorf("register drivers before starting the host: %w", err)
}
Functions ¶
This section is empty.
Types ¶
type Builder ¶
type Builder struct {
// contains filtered or unexported fields
}
Builder edits the construction settings of New. Callbacks share one fresh draft and run once in order; a nil callback panics. Retaining a draft does not permit changes to the published component. Dependencies and native SDK internals retain their existing ownership.
Example:
component := New(func(host *Builder) {
host.Logger(slog.Default())
})
_ = component
func (*Builder) ClientOptions ¶
ClientOptions is the base set of client options, for TLS, credentials, and interceptors. HostPort and Namespace always override the matching fields, including host defaults when no explicit setting is supplied. NewFrom ignores this record; configure the native client before wrapping it. It returns the same draft for subsequent settings.
Example:
component := New(func(host *Builder) {
host.ClientOptions(client.Options{})
})
_ = component
func (*Builder) HostPort ¶
HostPort is the address of the Temporal frontend. The default is client.DefaultHostPort. NewFrom ignores it. It returns the same draft for subsequent settings.
Example:
component := New(func(host *Builder) {
host.HostPort("localhost:7233")
})
_ = component
func (*Builder) Logger ¶
Logger supplies the logger for a client dialed by Host.Start. Nil adapts slog.Default at startup; an explicit native ClientOptions.Logger wins. NewFrom ignores this setting because SDK workers inherit the caller-owned client's logger, which this host cannot replace. It returns the same draft for subsequent settings.
Example:
component := New(func(host *Builder) {
host.Logger(slog.Default())
})
_ = component
func (*Builder) Namespace ¶
Namespace is the Temporal namespace. The default is "default". NewFrom ignores it. It returns the same draft for subsequent settings.
Example:
component := New(func(host *Builder) {
host.Namespace("example")
})
_ = component
func (*Builder) NoWorker ¶
NoWorker makes Host.Start dial without creating a worker, even when drivers registered workflows or activities. A Host with no registrations never creates a worker. It defaults to false. A later false restores permission without forcing an empty worker. Host.WorkerDisabled reports the configured flag. It returns the same draft for subsequent settings.
Example:
component := New(func(host *Builder) {
host.NoWorker(true)
})
_ = component
func (*Builder) TaskQueue ¶
TaskQueue is the task queue the worker polls. The default is "arkoren". It returns the same draft for subsequent settings.
Example:
component := New(func(host *Builder) {
host.TaskQueue("example")
})
_ = component
func (*Builder) WorkerOptions ¶
WorkerOptions is passed to worker.New. It returns the same draft for subsequent settings.
Example:
component := New(func(host *Builder) {
host.WorkerOptions(worker.Options{})
})
_ = component
func (*Builder) WorkerStopTimeout ¶
WorkerStopTimeout supplies Temporal's native graceful-stop wait. Zero uses ten seconds; a negative value disables that wait. A nonzero native WorkerOptions.WorkerStopTimeout wins. The SDK can apply the wait separately to internal workers and then cancel activity contexts; this is not a total shutdown deadline. Callbacks must cooperate, and SDK shutdown hooks must return. It returns the same draft for subsequent settings.
Example:
component := New(func(host *Builder) {
host.WorkerStopTimeout(time.Second)
})
_ = component
type BuilderFunc ¶
type BuilderFunc = func(*Builder)
BuilderFunc configures a fresh host construction draft. Callbacks run directly once, synchronously in order; nil entries panic and publication detaches the draft settings, not native clients or SDK internals.
Example:
policies := []BuilderFunc{
func(host *Builder) {
host.TaskQueue("orders")
},
}
host := New(policies...)
_ = host.TaskQueue()
type Host ¶
type Host struct {
// contains filtered or unexported fields
}
Host owns one Temporal client and one worker. Drivers register workflows and activities with Host.Register before Host.Start. NewFrom borrows its client instead; serialize Start and Stop calls.
Example:
host := New(func(host *Builder) {
host.TaskQueue("orders")
})
_ = host.TaskQueue()
func New ¶
func New(configure ...BuilderFunc) *Host
New returns a Host that dials Temporal in Host.Start. It performs no I/O. Construction callbacks run once synchronously in order on a fresh Builder; nil entries panic. SDK option records are copied by value, not recursively.
Example:
host := New(func(host *Builder) {
host.TaskQueue("orders")
})
_ = host.Client() // Nil until Start.
func NewFrom ¶
func NewFrom(conn client.Client, configure ...BuilderFunc) *Host
NewFrom returns a Host that wraps a caller-owned client. Host.Stop does not close it. TaskQueue, WorkerOptions, WorkerStopTimeout, and NoWorker apply; Logger and connection settings are ignored. SDK workers inherit the supplied client's logger, which must be configured before creating that client. Callbacks have the same once-only execution and publication rules as New. A nil client is rejected by Start without dialing a replacement.
Example:
wrap := func(connection client.Client) *Host {
return NewFrom(connection, func(host *Builder) {
host.TaskQueue("orders")
host.NoWorker(true)
})
}
_ = wrap // The caller retains responsibility for closing the connection.
func (*Host) Client ¶
Client returns the Temporal client. It is nil before Host.Start unless the Host was built with NewFrom. Native access does not transfer ownership or replace Start/Stop; callers must not close a client owned by the host.
Example:
native := host.Client()
if native == nil {
return errors.New("start the Temporal host before native access")
}
_ = native.ScheduleClient()
func (*Host) Ping ¶
Ping checks native server health through the host's current client. It returns an error without a client; a supplied NewFrom client is available immediately. This observes server reachability, not dispatcher or worker callback readiness.
Example:
if err := host.Ping(ctx); err != nil {
return fmt.Errorf("Temporal is not ready: %w", err)
}
func (*Host) Register ¶
Register queues fn to run against the worker before it starts. It returns ErrStarted after successful Start and rejects nil functions. Callbacks run synchronously during startup; panics propagate after owned client cleanup. Registration does not run callbacks or perform lifecycle I/O.
Example:
err := host.Register(func(registry worker.Registry) {
registry.RegisterActivity(sendReceipt)
})
if err != nil {
return err
}
func (*Host) Start ¶
Start dials Temporal, unless the client was supplied, then creates the worker, applies the registrations, and starts the worker. With no registrations, or with Builder.NoWorker, it creates no worker and the Host is only a client. A canceled context is rejected before acquiring resources. A nil NewFrom client is an error, not a request to dial a connection. SDK worker startup has no context parameter: cancellation during that call is observed after it returns, then acquired resources are released before Start returns the cancellation error. Registration and SDK hooks must return; Start cannot interrupt arbitrary callbacks. Failed attempts may be retried.
Example:
if err := host.Start(ctx); err != nil {
return err
}
func (*Host) Stop ¶
Stop initiates worker cleanup once and waits for it until ctx is done. An expired context still initiates cleanup. The owned native client is closed after worker cleanup returns, possibly after Stop returns; borrowed clients stay open. Call Stop after Start has returned: lifecycle calls are serialized.
WorkerStopTimeout is native graceful-stop policy, not a total Stop deadline. SDK shutdown hooks must return and activities must cooperate with context cancellation; neither Stop nor the SDK can forcibly stop arbitrary callbacks. Repeated calls wait for the same worker cleanup, each with its own context.
Example:
if err := host.Stop(ctx); err != nil {
return err // Worker cleanup continues after a context error.
}
func (*Host) TaskQueue ¶
TaskQueue returns the task queue the worker polls. This immutable value also routes client submissions to matching workers.
Example:
host := New(func(host *Builder) {
host.TaskQueue("orders")
})
queue := host.TaskQueue()
_ = queue // "orders", usable before startup.
func (*Host) WorkerDisabled ¶
WorkerDisabled reports whether the Host was built with Builder.NoWorker. It reports stored permission, not whether a worker is currently running.
Example:
host := New(func(host *Builder) {
host.NoWorker(true)
})
_ = host.WorkerDisabled() // True even before Start.
Source Files
Directories
¶
| Path | Synopsis |
|---|---|
|
internal
|
|
|
itest
Package itest holds the shared setup of the Temporal integration tests that run against the temporal service of driver/temporal/compose.yaml.
|
Package itest holds the shared setup of the Temporal integration tests that run against the temporal service of driver/temporal/compose.yaml. |
|
Package temporalschedule provides a Temporal-backed schedule driver.
|
Package temporalschedule provides a Temporal-backed schedule driver. |
|
Package temporalworkflow implements workflow.Driver on Temporal workflows and activities.
|
Package temporalworkflow implements workflow.Driver on Temporal workflows and activities. |
|
internal/execution
Package execution owns deterministic Temporal workflow execution.
|
Package execution owns deterministic Temporal workflow execution. |
|
internal/protocol
Package protocol owns Temporal raw JSON and native failure envelopes.
|
Package protocol owns Temporal raw JSON and native failure envelopes. |