Documentation
¶
Index ¶
- func NewClient(opts ...Option) (client.Client, error)
- type Config
- type DashboardStats
- type Option
- func WithClientOptions(fn func(*client.Options)) Option
- func WithConfig(c Config) Option
- func WithCredentials(creds client.Credentials) Option
- func WithHostPort(addr string) Option
- func WithNamespace(ns string) Option
- func WithOTelConfig(otelCfg *otel.Config) Option
- func WithTLS(tlsCfg *tls.Config) Option
- type ScheduleManager
- func (sm *ScheduleManager) Close(ctx context.Context)
- func (sm *ScheduleManager) CreateSchedule(ctx context.Context, scheduleID string, spec client.ScheduleSpec, ...) (client.ScheduleHandle, error)
- func (sm *ScheduleManager) CreateScheduleWithOptions(ctx context.Context, options client.ScheduleOptions) (client.ScheduleHandle, error)
- func (sm *ScheduleManager) CreateWorkflowSchedule(ctx context.Context, scheduleName string, options WorkflowScheduleOptions) (client.ScheduleHandle, error)
- func (sm *ScheduleManager) DeleteSchedule(ctx context.Context, scheduleID string) error
- func (sm *ScheduleManager) DeleteSchedules(ctx context.Context) error
- func (sm *ScheduleManager) GetClient() client.Client
- func (sm *ScheduleManager) GetSchedule(ctx context.Context, scheduleID string) (client.ScheduleHandle, error)
- func (sm *ScheduleManager) GetScheduleHandlers() map[string]client.ScheduleHandle
- func (sm *ScheduleManager) ListSchedules(ctx context.Context, limit int) ([]*client.ScheduleListEntry, error)
- func (sm *ScheduleManager) UpdateSchedule(ctx context.Context, scheduleID string, spec client.ScheduleSpec, ...) error
- type WorkerManager
- func (wm *WorkerManager) Close(ctx context.Context)
- func (wm *WorkerManager) GetClient() client.Client
- func (wm *WorkerManager) GetWorkers() []worker.Worker
- func (wm *WorkerManager) Register(taskQueue string, options worker.Options) worker.Worker
- func (wm *WorkerManager) Start(ctx context.Context, w worker.Worker) error
- func (wm *WorkerManager) StartAll(ctx context.Context) error
- type WorkflowDetails
- type WorkflowManager
- func (wm *WorkflowManager) CancelWorkflow(ctx context.Context, workflowID, runID string) error
- func (wm *WorkflowManager) CountWorkflows(ctx context.Context, query string) (int64, error)
- func (wm *WorkflowManager) DescribeWorkflow(ctx context.Context, workflowID, runID string) (*WorkflowDetails, error)
- func (wm *WorkflowManager) GetClient() client.Client
- func (wm *WorkflowManager) GetDashboardStats(ctx context.Context) (*DashboardStats, error)
- func (wm *WorkflowManager) GetRecentWorkflows(ctx context.Context, limit int) ([]*WorkflowDetails, error)
- func (wm *WorkflowManager) GetWorkflowHistory(ctx context.Context, workflowID, runID string) (*workflowservice.GetWorkflowExecutionHistoryResponse, error)
- func (wm *WorkflowManager) GetWorkflowResult(ctx context.Context, workflowID, runID string, valuePtr any) error
- func (wm *WorkflowManager) GetWorkflowStatus(ctx context.Context, workflowID, runID string) (enums.WorkflowExecutionStatus, error)
- func (wm *WorkflowManager) ListCompletedWorkflows(ctx context.Context, pageSize int) ([]*WorkflowDetails, error)
- func (wm *WorkflowManager) ListFailedWorkflows(ctx context.Context, pageSize int) ([]*WorkflowDetails, error)
- func (wm *WorkflowManager) ListRunningWorkflows(ctx context.Context, pageSize int) ([]*WorkflowDetails, error)
- func (wm *WorkflowManager) ListWorkflows(ctx context.Context, pageSize int, query string) ([]*WorkflowDetails, error)
- func (wm *WorkflowManager) ListWorkflowsByStatus(ctx context.Context, status enums.WorkflowExecutionStatus, pageSize int) ([]*WorkflowDetails, error)
- func (wm *WorkflowManager) QueryWorkflow(ctx context.Context, workflowID, runID, queryType string, args ...any) (any, error)
- func (wm *WorkflowManager) SearchWorkflowsByID(ctx context.Context, workflowIDPrefix string, pageSize int) ([]*WorkflowDetails, error)
- func (wm *WorkflowManager) SearchWorkflowsByType(ctx context.Context, workflowType string, pageSize int) ([]*WorkflowDetails, error)
- func (wm *WorkflowManager) SignalWorkflow(ctx context.Context, workflowID, runID, signalName string, arg any) error
- func (wm *WorkflowManager) TerminateWorkflow(ctx context.Context, workflowID, runID, reason string) error
- type WorkflowScheduleOptions
- type ZerologAdapter
- func (z *ZerologAdapter) Debug(msg string, keyvals ...any)
- func (z *ZerologAdapter) Error(msg string, keyvals ...any)
- func (z *ZerologAdapter) Info(msg string, keyvals ...any)
- func (z *ZerologAdapter) Warn(msg string, keyvals ...any)
- func (z *ZerologAdapter) With(keyvals ...any) temporallog.Logger
- func (z *ZerologAdapter) WithCallerSkip(skip int) temporallog.Logger
Examples ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func NewClient ¶
NewClient creates a Temporal client. It starts from DefaultConfig and applies the given options in order.
Example ¶
ExampleNewClient demonstrates creating a Temporal client with the options API. Without options, NewClient dials localhost:7233 in the "default" namespace. The caller owns the returned client and must close it.
This example has no Output comment because the result depends on a running Temporal server; it is compile-checked but not executed by go test.
package main
import (
"fmt"
"github.com/jasoet/pkg/v3/temporal"
)
func main() {
c, err := temporal.NewClient(
temporal.WithHostPort("localhost:7233"),
temporal.WithNamespace("default"),
)
if err != nil {
// No Temporal server is listening; handle the dial error.
fmt.Println("dial failed:", err != nil)
return
}
defer c.Close()
fmt.Println("connected:", c != nil)
}
Output:
Types ¶
type Config ¶
type Config struct {
HostPort string `yaml:"hostPort" mapstructure:"hostPort"`
Namespace string `yaml:"namespace" mapstructure:"namespace"`
OTelConfig *otel.Config `yaml:"-" mapstructure:"-"`
// TLS configures a TLS connection to the Temporal frontend. Required for
// Temporal Cloud and any TLS-enabled self-hosted server. Not serializable.
TLS *tls.Config `yaml:"-" mapstructure:"-"`
// Credentials authenticates the client — e.g. an API key for Temporal
// Cloud (client.NewAPIKeyStaticCredentials) or mTLS. Not serializable.
Credentials client.Credentials `yaml:"-" mapstructure:"-"`
// contains filtered or unexported fields
}
func DefaultConfig ¶
func DefaultConfig() *Config
DefaultConfig returns a Config with sensible defaults. It is a pure factory function and performs no I/O or logging.
type DashboardStats ¶
type DashboardStats struct {
TotalRunning int64
TotalCompleted int64
TotalFailed int64
TotalCanceled int64
TotalTerminated int64
AverageDuration time.Duration
}
DashboardStats provides aggregated workflow statistics
type Option ¶
type Option func(*Config)
Option mutates a Config. Options are applied to DefaultConfig by NewClient.
func WithClientOptions ¶
WithClientOptions registers a hook that receives the fully-assembled client.Options immediately before dialing, letting callers set any SDK option this package does not expose directly. Hooks run in registration order, last, so they can override everything set beforehand.
func WithConfig ¶
WithConfig replaces the entire configuration with c.
func WithCredentials ¶
func WithCredentials(creds client.Credentials) Option
WithCredentials sets the client credentials used to authenticate — for example an API key for Temporal Cloud via client.NewAPIKeyStaticCredentials, or mTLS credentials.
func WithHostPort ¶
WithHostPort sets the Temporal frontend address (host:port).
func WithNamespace ¶
WithNamespace sets the Temporal namespace.
func WithOTelConfig ¶
WithOTelConfig attaches OTel tracing/metrics to the client.
type ScheduleManager ¶
type ScheduleManager struct {
// contains filtered or unexported fields
}
func NewScheduleManager ¶
func NewScheduleManager(temporalClient client.Client) (*ScheduleManager, error)
NewScheduleManager creates a ScheduleManager using the provided client. The caller retains ownership of the client and is responsible for closing it; Close does not close the client.
Example ¶
ExampleNewScheduleManager demonstrates creating a ScheduleManager from a caller-owned client. ScheduleManager.Close(ctx) does not close the client; the caller closes it.
This example has no Output comment because the result depends on a running Temporal server; it is compile-checked but not executed by go test.
package main
import (
"context"
"fmt"
"time"
"github.com/jasoet/pkg/v3/temporal"
)
func main() {
ctx := context.Background()
c, err := temporal.NewClient()
if err != nil {
fmt.Println("dial failed:", err != nil)
return
}
defer c.Close()
sm, err := temporal.NewScheduleManager(c)
if err != nil {
fmt.Println("manager failed:", err != nil)
return
}
defer sm.Close(ctx)
// Create an interval schedule (requires a running Temporal server).
_, err = sm.CreateWorkflowSchedule(ctx, "hourly-report", temporal.WorkflowScheduleOptions{
WorkflowID: "report-workflow",
Workflow: "ReportWorkflow", // or a workflow function reference
TaskQueue: "reports",
Interval: time.Hour,
Args: []any{"daily-report"},
})
if err != nil {
fmt.Println("schedule failed:", err != nil)
return
}
fmt.Println("schedule created")
}
Output:
func (*ScheduleManager) Close ¶
func (sm *ScheduleManager) Close(ctx context.Context)
Close closes the ScheduleManager. It does not close the Temporal client; the caller owns the client and must close it. The ctx parameter is used for logging only.
func (*ScheduleManager) CreateSchedule ¶
func (sm *ScheduleManager) CreateSchedule(ctx context.Context, scheduleID string, spec client.ScheduleSpec, action *client.ScheduleWorkflowAction) (client.ScheduleHandle, error)
func (*ScheduleManager) CreateScheduleWithOptions ¶
func (sm *ScheduleManager) CreateScheduleWithOptions(ctx context.Context, options client.ScheduleOptions) (client.ScheduleHandle, error)
func (*ScheduleManager) CreateWorkflowSchedule ¶
func (sm *ScheduleManager) CreateWorkflowSchedule(ctx context.Context, scheduleName string, options WorkflowScheduleOptions) (client.ScheduleHandle, error)
func (*ScheduleManager) DeleteSchedule ¶
func (sm *ScheduleManager) DeleteSchedule(ctx context.Context, scheduleID string) error
DeleteSchedule deletes a specific schedule by ID
func (*ScheduleManager) DeleteSchedules ¶
func (sm *ScheduleManager) DeleteSchedules(ctx context.Context) error
DeleteSchedules deletes every schedule this manager is tracking. It is convergent under partial failure: each schedule is deleted independently, a NotFound response is treated as success (the schedule is already gone), and every successfully-removed schedule is dropped from tracking so a retry converges instead of re-deleting or re-hitting NotFound. Schedules whose deletion genuinely failed are left in tracking and their errors are joined and returned. RPCs are made without holding the manager lock.
func (*ScheduleManager) GetClient ¶
func (sm *ScheduleManager) GetClient() client.Client
func (*ScheduleManager) GetSchedule ¶
func (sm *ScheduleManager) GetSchedule(ctx context.Context, scheduleID string) (client.ScheduleHandle, error)
GetSchedule retrieves a schedule handle by ID
func (*ScheduleManager) GetScheduleHandlers ¶
func (sm *ScheduleManager) GetScheduleHandlers() map[string]client.ScheduleHandle
func (*ScheduleManager) ListSchedules ¶
func (sm *ScheduleManager) ListSchedules(ctx context.Context, limit int) ([]*client.ScheduleListEntry, error)
ListSchedules lists all schedules with a limit
func (*ScheduleManager) UpdateSchedule ¶
func (sm *ScheduleManager) UpdateSchedule(ctx context.Context, scheduleID string, spec client.ScheduleSpec, action *client.ScheduleWorkflowAction) error
UpdateSchedule updates an existing schedule
type WorkerManager ¶
type WorkerManager struct {
// contains filtered or unexported fields
}
func NewWorkerManager ¶
func NewWorkerManager(client client.Client) (*WorkerManager, error)
NewWorkerManager creates a WorkerManager using the provided client. The caller retains ownership of the client and is responsible for closing it; Close does not close the client.
func (*WorkerManager) Close ¶
func (wm *WorkerManager) Close(ctx context.Context)
Close stops all registered workers. It does not close the Temporal client; the caller owns the client and must close it. The ctx parameter is used for logging only. Close is idempotent and safe to call concurrently: the underlying workers are stopped at most once, avoiding the double-close panic that the SDK's worker.Stop() raises on a second invocation.
func (*WorkerManager) GetClient ¶
func (wm *WorkerManager) GetClient() client.Client
GetClient returns the Temporal client provided at construction. The client is owned by the caller; Close does not close it.
func (*WorkerManager) GetWorkers ¶
func (wm *WorkerManager) GetWorkers() []worker.Worker
type WorkflowDetails ¶
type WorkflowDetails struct {
WorkflowID string
RunID string
WorkflowType string
Status enums.WorkflowExecutionStatus
StartTime time.Time
CloseTime time.Time
ExecutionTime time.Duration
HistoryLength int64
}
WorkflowDetails contains detailed information about a workflow execution
type WorkflowManager ¶
type WorkflowManager struct {
// contains filtered or unexported fields
}
WorkflowManager provides workflow query and management operations.
All operations resolve their namespace consistently: an empty namespace (the default from NewWorkflowManager) means "use the namespace the client was configured with", matching the client-method calls (Describe/Cancel/ Signal/Query/GetResult) which always target the client's namespace. Supply an explicit namespace via NewWorkflowManagerWithNamespace only to scope the visibility queries (List/Count) to a different namespace.
func NewWorkflowManager ¶
func NewWorkflowManager(client client.Client) (*WorkflowManager, error)
NewWorkflowManager creates a new WorkflowManager that inherits the namespace the client was configured with, so every operation agrees on the namespace. Use NewWorkflowManagerWithNamespace to scope visibility queries to a different namespace. The caller retains ownership of the client and is responsible for closing it.
func NewWorkflowManagerWithNamespace ¶
func NewWorkflowManagerWithNamespace(client client.Client, namespace string) (*WorkflowManager, error)
NewWorkflowManagerWithNamespace creates a new WorkflowManager that scopes its List/Count visibility queries to the given namespace. Pass "" to inherit the client's configured namespace (equivalent to NewWorkflowManager). Supplying a namespace that differs from the client's affects only List/Count; the other operations always target the client's namespace. The caller retains ownership of the client and is responsible for closing it.
func (*WorkflowManager) CancelWorkflow ¶
func (wm *WorkflowManager) CancelWorkflow(ctx context.Context, workflowID, runID string) error
CancelWorkflow cancels a running workflow execution
func (*WorkflowManager) CountWorkflows ¶
CountWorkflows counts workflows matching a query
func (*WorkflowManager) DescribeWorkflow ¶
func (wm *WorkflowManager) DescribeWorkflow(ctx context.Context, workflowID, runID string) (*WorkflowDetails, error)
DescribeWorkflow retrieves detailed information about a specific workflow execution
func (*WorkflowManager) GetClient ¶
func (wm *WorkflowManager) GetClient() client.Client
GetClient returns the Temporal client provided at construction. The client is owned by the caller and must be closed by the caller.
func (*WorkflowManager) GetDashboardStats ¶
func (wm *WorkflowManager) GetDashboardStats(ctx context.Context) (*DashboardStats, error)
GetDashboardStats retrieves aggregated statistics for all workflows
func (*WorkflowManager) GetRecentWorkflows ¶
func (wm *WorkflowManager) GetRecentWorkflows(ctx context.Context, limit int) ([]*WorkflowDetails, error)
GetRecentWorkflows retrieves the most recent workflow executions
func (*WorkflowManager) GetWorkflowHistory ¶
func (wm *WorkflowManager) GetWorkflowHistory(ctx context.Context, workflowID, runID string) (*workflowservice.GetWorkflowExecutionHistoryResponse, error)
GetWorkflowHistory retrieves the full event history of a workflow execution. It queries the client's configured namespace (the manager's explicit namespace override, if any, applies only to List/Count visibility queries). All pages are collected into the returned response's History.Events.
func (*WorkflowManager) GetWorkflowResult ¶
func (wm *WorkflowManager) GetWorkflowResult(ctx context.Context, workflowID, runID string, valuePtr any) error
GetWorkflowResult retrieves the result of a workflow run into valuePtr.
This BLOCKS until the run completes: run.Get waits for a still-running workflow to finish rather than returning immediately. For dashboard-style callers that must not hang on a running workflow, gate this behind a GetWorkflowStatus check for a terminal status, or pass a ctx with a deadline.
func (*WorkflowManager) GetWorkflowStatus ¶
func (wm *WorkflowManager) GetWorkflowStatus(ctx context.Context, workflowID, runID string) (enums.WorkflowExecutionStatus, error)
GetWorkflowStatus returns the current status of a workflow execution
func (*WorkflowManager) ListCompletedWorkflows ¶
func (wm *WorkflowManager) ListCompletedWorkflows(ctx context.Context, pageSize int) ([]*WorkflowDetails, error)
ListCompletedWorkflows returns completed workflows
func (*WorkflowManager) ListFailedWorkflows ¶
func (wm *WorkflowManager) ListFailedWorkflows(ctx context.Context, pageSize int) ([]*WorkflowDetails, error)
ListFailedWorkflows returns failed workflows
func (*WorkflowManager) ListRunningWorkflows ¶
func (wm *WorkflowManager) ListRunningWorkflows(ctx context.Context, pageSize int) ([]*WorkflowDetails, error)
ListRunningWorkflows returns all currently running workflows
func (*WorkflowManager) ListWorkflows ¶
func (wm *WorkflowManager) ListWorkflows(ctx context.Context, pageSize int, query string) ([]*WorkflowDetails, error)
ListWorkflows lists workflows with pagination and optional query filter
func (*WorkflowManager) ListWorkflowsByStatus ¶
func (wm *WorkflowManager) ListWorkflowsByStatus(ctx context.Context, status enums.WorkflowExecutionStatus, pageSize int) ([]*WorkflowDetails, error)
ListWorkflowsByStatus lists workflows filtered by execution status
func (*WorkflowManager) QueryWorkflow ¶
func (wm *WorkflowManager) QueryWorkflow(ctx context.Context, workflowID, runID, queryType string, args ...any) (any, error)
QueryWorkflow queries a running workflow for custom data. The returned value is the SDK's raw converter.EncodedValue (not the decoded payload); type-assert it to converter.EncodedValue and call Get(&dst) to decode into a typed value.
func (*WorkflowManager) SearchWorkflowsByID ¶
func (wm *WorkflowManager) SearchWorkflowsByID(ctx context.Context, workflowIDPrefix string, pageSize int) ([]*WorkflowDetails, error)
SearchWorkflowsByID searches for workflows matching a workflow ID pattern
func (*WorkflowManager) SearchWorkflowsByType ¶
func (wm *WorkflowManager) SearchWorkflowsByType(ctx context.Context, workflowType string, pageSize int) ([]*WorkflowDetails, error)
SearchWorkflowsByType searches workflows by workflow type name
func (*WorkflowManager) SignalWorkflow ¶
func (wm *WorkflowManager) SignalWorkflow(ctx context.Context, workflowID, runID, signalName string, arg any) error
SignalWorkflow sends a signal to a running workflow
func (*WorkflowManager) TerminateWorkflow ¶
func (wm *WorkflowManager) TerminateWorkflow(ctx context.Context, workflowID, runID, reason string) error
TerminateWorkflow terminates a workflow execution with a reason
type WorkflowScheduleOptions ¶
type ZerologAdapter ¶
type ZerologAdapter struct {
// contains filtered or unexported fields
}
func NewZerologAdapter ¶
func NewZerologAdapter(logger zerolog.Logger) *ZerologAdapter
NewZerologAdapter creates a new ZerologAdapter
func (*ZerologAdapter) Debug ¶
func (z *ZerologAdapter) Debug(msg string, keyvals ...any)
func (*ZerologAdapter) Error ¶
func (z *ZerologAdapter) Error(msg string, keyvals ...any)
func (*ZerologAdapter) Info ¶
func (z *ZerologAdapter) Info(msg string, keyvals ...any)
func (*ZerologAdapter) Warn ¶
func (z *ZerologAdapter) Warn(msg string, keyvals ...any)
func (*ZerologAdapter) With ¶
func (z *ZerologAdapter) With(keyvals ...any) temporallog.Logger
func (*ZerologAdapter) WithCallerSkip ¶
func (z *ZerologAdapter) WithCallerSkip(skip int) temporallog.Logger
Directories
¶
| Path | Synopsis |
|---|---|
|
Package job provides a type-focused abstraction for one registered Temporal workflow.
|
Package job provides a type-focused abstraction for one registered Temporal workflow. |
|
Package testcontainer provides utilities for running Temporal server in Docker containers for integration testing.
|
Package testcontainer provides utilities for running Temporal server in Docker containers for integration testing. |