go.temporal.io/server/chasm/registrable_task.go
222 LOC · 96 covered · 126 uncovered · 20 ranges · 957 concepts · 15 introducers · 402 tests
File neighbourhood
The centred file is linked to every concept that introduces one of its ranges, every test that runs code from the file, and the gray connector concepts standing between those tests and the file's own introducer concepts. Undirected links join concepts to every file where they introduce source and concepts to the tests they introduce; arrows show specialization between the displayed concepts and bridge only concepts omitted from this view. Concept colors match the source ranges below; connector concepts have no source color and are shown in gray.
Focused file, its introducer and connector concepts, their introduced files, and tests that run code from the file
In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the related-file, concept, and source links on this page.
Graph controls are ready.
Interactive rendering requires JavaScript and WebGL. Use the related-file, concept, and source links on this page while the interactive map is unavailable.
1
package chasm
2
3
import (
4
"context"
5
"fmt"
6
"reflect"
7
)
8
9
// SingletonTaskMode controls how the framework handles a new task of a singleton type
10
// when a task of the same type already exists on the component instance.
11
type SingletonTaskMode int
12
13
const (
14
// SingletonTaskModeReplace removes the existing task and schedules the new one in its place.
15
SingletonTaskModeReplace SingletonTaskMode = iota + 1
16
// SingletonTaskModeIgnore keeps the existing task and discards the new one.
17
SingletonTaskModeIgnore
18
)
19
20
type (
21
RegistrableTask struct {
22
taskType string
23
goType reflect.Type
24
componentGoType reflect.Type // It is not clear how this one is used.
25
validateFn validateFn
26
pureTaskExecuteFn pureTaskExecuteFn
27
sideEffectTaskExecuteFn sideEffectTaskExecuteFn
28
sideEffectTaskDiscardFn sideEffectTaskDiscardFn
29
isPureTask bool
30
outboundTaskGroup string // For grouping on the outbound queue. See [WithTaskGroup] for details.
31
singletonMode SingletonTaskMode // If non-zero, at most one task of this type may exist per component instance.
32
33
// Those two fields are initialized when the component is registered to a library.
34
library namer
35
taskTypeID uint32
36
}
37
38
RegistrableTaskOption func(*RegistrableTask)
39
40
validateFn func(Context, any, TaskInvocation, any, *Registry) (bool, error)
41
pureTaskExecuteFn func(MutableContext, any, TaskAttributes, any, *Registry) error
42
sideEffectTaskExecuteFn func(context.Context, ComponentRef, TaskAttributes, any) error
43
sideEffectTaskDiscardFn func(context.Context, ComponentRef, TaskAttributes, any) error
44
)
45
46
// NewRegistrableSideEffectTask creates a new registrable side-effect task. NOTE: C is not Component but any.
47
// The handler's Discard method is called on standby clusters when a task has been pending past the discard delay.
48
func NewRegistrableSideEffectTask[C any, T any](
49
taskType string,
50
handler SideEffectTaskHandler[C, T],
51
opts ...RegistrableTaskOption,
53
>
return newRegistrableTask(
54
>
taskType,
55
>
reflect.TypeFor[T](),
56
>
reflect.TypeFor[C](),
57
>
func(
58
>
ctx Context,
59
>
component any,
60
>
taskInvocation TaskInvocation,
61
>
taskData any,
62
>
registry *Registry,
63
>
) (bool, error) {
65
>
ctx,
66
>
component.(C),
67
>
taskInvocation,
68
>
taskData.(T),
69
>
)
70
>
},
71
nil, // pureTaskExecuteFn is not used for side effect tasks
72
func(
73
ctx context.Context,
74
componentRef ComponentRef,
75
taskAttrs TaskAttributes,
76
taskData any,
78
>
return handler.Execute(ctx, componentRef, taskAttrs, taskData.(T))
79
>
},
80
false,
82
>
return handler.Discard(ctx, ref, attrs, task.(T))
83
>
},
84
opts...,
85
)
86
}
87
88
func NewRegistrablePureTask[C any, T any](
89
taskType string,
90
handler PureTaskHandler[C, T],
91
opts ...RegistrableTaskOption,
93
>
return newRegistrableTask(
94
>
taskType,
95
>
reflect.TypeFor[T](),
96
>
reflect.TypeFor[C](),
97
>
func(
98
>
ctx Context,
99
>
component any,
100
>
taskInvocation TaskInvocation,
101
>
taskData any,
102
>
registry *Registry,
103
>
) (bool, error) {
105
>
ctx,
106
>
component.(C),
107
>
taskInvocation,
108
>
taskData.(T),
109
>
)
110
>
},
111
func(
112
ctx MutableContext,
113
component any,
114
taskAttrs TaskAttributes,
115
taskData any,
116
registry *Registry,
118
>
return handler.Execute(
119
>
ctx,
120
>
component.(C),
121
>
taskAttrs,
122
>
taskData.(T),
123
>
)
124
>
},
125
nil, // sideEffectTaskExecuteFn is not used for pure tasks
126
true,
127
nil, // sideEffectTaskDiscardFn is not used for pure tasks
128
opts...,
129
)
130
}
131
132
func newRegistrableTask(
133
taskType string,
134
goType, componentGoType reflect.Type,
135
validateFn validateFn,
136
pureTaskExecuteFn pureTaskExecuteFn,
137
sideEffectTaskExecuteFn sideEffectTaskExecuteFn,
138
isPureTask bool,
139
sideEffectTaskDiscardFn sideEffectTaskDiscardFn,
140
opts ...RegistrableTaskOption,
142
>
rt := &RegistrableTask{
143
>
taskType: taskType,
144
>
goType: goType,
145
>
componentGoType: componentGoType,
146
>
validateFn: validateFn,
147
>
pureTaskExecuteFn: pureTaskExecuteFn,
148
>
sideEffectTaskExecuteFn: sideEffectTaskExecuteFn,
149
>
sideEffectTaskDiscardFn: sideEffectTaskDiscardFn,
150
>
isPureTask: isPureTask,
151
>
}
152
>
153
>
for _, opt := range opts {
155
>
}
156
158
}
159
160
func (rt *RegistrableTask) registerToLibrary(
161
library namer,
163
>
if rt.library != nil {
164
>
return "", 0, fmt.Errorf("task %s is already registered in library %s", rt.taskType, rt.library.Name())
registrable_task.go ×1
165
>
}
166
168
>
169
>
fqn := rt.fqType()
170
>
rt.taskTypeID = GenerateTypeID(fqn)
171
>
// If outboundTaskGroup wasn't set on creation default it here,
172
>
// since this is the first place we will have the fqn.
173
>
if rt.outboundTaskGroup == "" {
175
>
}
177
}
178
179
// TaskGroup returns the side-effect task group for the task.
181
>
return rt.outboundTaskGroup
182
>
}
183
184
// GoType returns the reflect.Type of the task's Go struct.
185
func (rt *RegistrableTask) GoType() reflect.Type {
186
return rt.goType
187
}
188
189
// fqType returns the fully qualified name of the task, which is a combination of
190
// the library name and the task type. This is used to uniquely identify
191
// the task in the registry.
193
>
if rt.library == nil {
194
// this should never happen because the task is only accessible from the library.
195
panic("task is not registered to a library")
196
}
198
}
199
200
// WithTaskGroup sets the task group for the task. The task group is used when
201
// the side effect's destination is specified for grouping semantics on the outbound queue,
202
// affects multi-cursor and the circuit breaker.
203
// If task group isn't provided, the task group will default to the fully qualified name at library registration.
205
>
return func(rt *RegistrableTask) {
206
>
rt.outboundTaskGroup = taskgroup
207
>
}
208
}
209
210
// WithSingletonTask configures the task type as a singleton: at most one task of this type
211
// may exist per component instance at any time. The mode controls what happens when a new
212
// task is added while one already exists:
213
// - [SingletonTaskModeReplace]: the existing task is removed and the new one takes its place.
214
// - [SingletonTaskModeIgnore]: the existing task is kept and the new one is discarded.
215
//
216
// Singleton semantics are enforced after task validation, so an invalid new task is dropped
217
// before any replacement or ignore logic applies.
219
>
return func(rt *RegistrableTask) {
220
>
rt.singletonMode = mode
221
>
}
222
}