client

Documentation

Index

• Constants
• func ClientProcessFn(ackTimeout time.Duration, counter *atomic.Int64, noRecovery bool, ...) func(ctx context.Context, log *slog.Logger, msg jetstream.Msg) (bool, error)
• func ReParentSpan(ctx context.Context, state *model.WorkflowState) context.Context
• type JobGetter
• type MessageFnExecutor
• type MessageFnLocator
• type MessageSendCompleter
• type ProcessInstanceErrorCanceller
• type ServiceFnExecutor
• type ServiceFnLocator
• type ServiceTaskCompleter
• type ServiceTaskProcessParams
• type WorkflowErrorHandler

Constants

const InternalActivityId keys.ContextKey = "__INTERNAL_ACID"

InternalActivityId is a constant of type ContextKey used for storing the internal activity ID in a context.

const InternalExecutionId keys.ContextKey = "__INTERNAL_EXID"

InternalExecutionId is a constant of type ContextKey used for storing the internal execution ID in a context.

const InternalProcessInstanceId keys.ContextKey = "__INTERNAL_PIID"

InternalProcessInstanceId is a constant of type ContextKey used for storing the internal process instance ID in a context.

const InternalTaskId keys.ContextKey = "__INTERNAL_TID"

InternalTaskId is a constant of type ContextKey used for storing the internal task ID in a context.

Functions

func ClientProcessFn

func ClientProcessFn(ackTimeout time.Duration, counter *atomic.Int64, noRecovery bool, jobGetter JobGetter, params ServiceTaskProcessParams) func(ctx context.Context, log *slog.Logger, msg jetstream.Msg) (bool, error)

ClientProcessFn is a function that processes client messages in a Jetstream subscription. It increments a counter, checks version compatibility, re-parents a span, and completes service tasks or message tasks.

Parameters: - ackTimeout: The duration for waiting to acknowledge the message. - counter: A pointer to an atomic.Int64 variable representing a counter. - jobGetter: An implementation of the JobGetter interface that retrieves SHAR Jobs. - params: An instance of ServiceTaskProcessParams that provides various behaviours to the service task processor.

Returns: - A function that processes client messages and returns a boolean and an error. The boolean indicates whether the message processing is complete, and the error provides any encountered error.

func ReParentSpan

func ReParentSpan(ctx context.Context, state *model.WorkflowState) context.Context

ReParentSpan re-parents a span in the given context with the span ID obtained from the WorkflowState ID. If the span context in the context is valid, it replaces the span ID with the 64-bit representation obtained from the WorkflowState ID. Otherwise, it returns the original context.

Parameters: - ctx: The context to re-parent the span in. - state: The WorkflowState containing the ID to extract the new span ID from.

Returns: - The context with the re-parented span ID or the original context if the span context is invalid.

Types

type JobGetter

type JobGetter interface {
	GetJob(ctx context.Context, id string) (*model.WorkflowState, error)
}

JobGetter represents the ability to retrieve SHAR Jobs.

type MessageFnExecutor

type MessageFnExecutor func(ctx context.Context, trackingID string, job *model.WorkflowState, def *task.FnDef, inVars model.Vars) error

MessageFnExecutor is the prototype function for executing a message task function.

type MessageFnLocator

type MessageFnLocator func(job *model.WorkflowState) (*task.FnDef, error)

MessageFnLocator is the prototype function for locating a message task function.

type MessageSendCompleter

type MessageSendCompleter func(ctx context.Context, trackingID string, newVars model.Vars) error

MessageSendCompleter the prototype function for completing a message task.

type ProcessInstanceErrorCanceller

type ProcessInstanceErrorCanceller func(ctx context.Context, processInstanceID string, wfe *model.Error) error

ProcessInstanceErrorCanceller is the prototype function for cancelling a process instance in the event of an error.

type ServiceFnExecutor

type ServiceFnExecutor func(ctx context.Context, trackingID string, job *model.WorkflowState, def *task.FnDef, inVars model.Vars) (model.Vars, error)

ServiceFnExecutor is the prototype function for executing a service task function.

type ServiceFnLocator

type ServiceFnLocator func(job *model.WorkflowState) (*task.FnDef, error)

ServiceFnLocator is the prototype function for locating a service task function.

type ServiceTaskCompleter

type ServiceTaskCompleter func(ctx context.Context, trackingID string, newVars model.Vars, compensating bool) error

ServiceTaskCompleter is the prototype function for completing a service task.

type ServiceTaskProcessParams

type ServiceTaskProcessParams struct {
	SvcFnExecutor    ServiceFnExecutor
	MsgFnExecutor    MessageFnExecutor
	SvcFnLocator     ServiceFnLocator
	MsgFnLocator     MessageFnLocator
	SvcTaskCompleter ServiceTaskCompleter
	MsgSendCompleter MessageSendCompleter
	WfErrorHandler   WorkflowErrorHandler
	PiErrorCanceller ProcessInstanceErrorCanceller
}

ServiceTaskProcessParams provides a behaviour to the service task processor.

type WorkflowErrorHandler

type WorkflowErrorHandler func(ctx context.Context, ns string, trackingID string, errorCode string, binVars []byte) (*model.HandleWorkflowErrorResponse, error)

WorkflowErrorHandler is the prototype function for handling a workflow error.