40
nsID namespace.ID,
41
defaultVersion int64,
43
>
44
>
generator := NewEventGenerator(time.Now().UnixNano())
45
>
generator.SetVersion(defaultVersion)
46
>
// Functions
47
>
notPendingWorkflowTask := func(input ...any) bool {
48
>
count := 0
49
>
history := input[0].([]Vertex)
50
>
for _, e := range history {
51
>
switch e.GetName() {
52
>
case enumspb.EVENT_TYPE_WORKFLOW_TASK_SCHEDULED.String():
53
>
count++
54
case enumspb.EVENT_TYPE_WORKFLOW_TASK_COMPLETED.String(),
55
enumspb.EVENT_TYPE_WORKFLOW_TASK_FAILED.String(),
57
>
count--
58
}
59
}
61
}
63
>
history := input[0].([]Vertex)
64
>
for _, e := range history {
65
>
if e.GetName() == enumspb.EVENT_TYPE_ACTIVITY_TASK_COMPLETED.String() {
66
>
return true
67
>
}
68
}
70
}
72
>
count := 0
73
>
history := input[0].([]Vertex)
74
>
for _, e := range history {
75
>
switch e.GetName() {
76
>
case enumspb.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED.String():
77
>
count++
78
case enumspb.EVENT_TYPE_ACTIVITY_TASK_CANCELED.String(),
79
enumspb.EVENT_TYPE_ACTIVITY_TASK_FAILED.String(),
80
enumspb.EVENT_TYPE_ACTIVITY_TASK_TIMED_OUT.String(),
82
>
count--
83
}
84
}
86
}
88
>
if len(currentBatch) == 0 {
89
>
return true
90
>
}
91
93
>
for _, event := range history {
94
>
switch event.GetName() {
95
>
case enumspb.EVENT_TYPE_WORKFLOW_TASK_SCHEDULED.String():
96
>
hasPendingWorkflowTask = true
97
case enumspb.EVENT_TYPE_WORKFLOW_TASK_COMPLETED.String(),
98
enumspb.EVENT_TYPE_WORKFLOW_TASK_FAILED.String(),
100
>
hasPendingWorkflowTask = false
101
}
102
}
104
>
return false
105
>
}
106
>
if currentBatch[len(currentBatch)-1].GetName() == enumspb.EVENT_TYPE_WORKFLOW_TASK_SCHEDULED.String() {
107
return false
108
}
109
>
if currentBatch[0].GetName() == enumspb.EVENT_TYPE_WORKFLOW_TASK_COMPLETED.String() {
history_event_util.go
110
>
return len(currentBatch) == 1
111
>
}
112
>
return true
113
}
114
115
// Setup workflow task model
117
>
workflowTaskSchedule := NewHistoryEventVertex(enumspb.EVENT_TYPE_WORKFLOW_TASK_SCHEDULED.String())
118
>
workflowTaskSchedule.SetDataFunc(func(input ...any) any {
119
>
lastGeneratedEvent := input[1].(*historypb.HistoryEvent)
120
>
eventID := lastGeneratedEvent.GetEventId() + 1
121
>
version := input[2].(int64)
122
>
historyEvent := getDefaultHistoryEvent(eventID, version)
123
>
historyEvent.EventType = enumspb.EVENT_TYPE_WORKFLOW_TASK_SCHEDULED
124
>
historyEvent.Attributes = &historypb.HistoryEvent_WorkflowTaskScheduledEventAttributes{WorkflowTaskScheduledEventAttributes: &historypb.WorkflowTaskScheduledEventAttributes{
125
>
TaskQueue: &taskqueuepb.TaskQueue{
126
>
Name: taskQueue,
127
>
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
128
>
},
129
>
StartToCloseTimeout: durationpb.New(timeout),
130
>
Attempt: workflowTaskAttempts,
131
>
}}
132
>
return historyEvent
133
>
})
134
>
workflowTaskStart := NewHistoryEventVertex(enumspb.EVENT_TYPE_WORKFLOW_TASK_STARTED.String())
135
>
workflowTaskStart.SetIsStrictOnNextVertex(true)
136
>
workflowTaskStart.SetDataFunc(func(input ...any) any {
137
>
lastEvent := input[0].(*historypb.HistoryEvent)
138
>
lastGeneratedEvent := input[1].(*historypb.HistoryEvent)
139
>
eventID := lastGeneratedEvent.GetEventId() + 1
140
>
version := input[2].(int64)
141
>
historyEvent := getDefaultHistoryEvent(eventID, version)
142
>
historyEvent.EventType = enumspb.EVENT_TYPE_WORKFLOW_TASK_STARTED
143
>
historyEvent.Attributes = &historypb.HistoryEvent_WorkflowTaskStartedEventAttributes{WorkflowTaskStartedEventAttributes: &historypb.WorkflowTaskStartedEventAttributes{
144
>
ScheduledEventId: lastEvent.EventId,
145
>
Identity: identity,
146
>
RequestId: uuid.NewString(),
147
>
}}
148
>
return historyEvent
149
>
})
150
>
workflowTaskFail := NewHistoryEventVertex(enumspb.EVENT_TYPE_WORKFLOW_TASK_FAILED.String())
151
>
workflowTaskFail.SetDataFunc(func(input ...any) any {
152
>
lastEvent := input[0].(*historypb.HistoryEvent)
153
>
lastGeneratedEvent := input[1].(*historypb.HistoryEvent)
154
>
eventID := lastGeneratedEvent.GetEventId() + 1
155
>
version := input[2].(int64)
156
>
historyEvent := getDefaultHistoryEvent(eventID, version)
157
>
historyEvent.EventType = enumspb.EVENT_TYPE_WORKFLOW_TASK_FAILED
158
>
historyEvent.Attributes = &historypb.HistoryEvent_WorkflowTaskFailedEventAttributes{WorkflowTaskFailedEventAttributes: &historypb.WorkflowTaskFailedEventAttributes{
159
>
ScheduledEventId: lastEvent.GetWorkflowTaskStartedEventAttributes().ScheduledEventId,
160
>
StartedEventId: lastEvent.EventId,
161
>
Cause: enumspb.WORKFLOW_TASK_FAILED_CAUSE_UNHANDLED_COMMAND,
162
>
Identity: identity,
163
>
ForkEventVersion: version,
164
>
}}
165
>
return historyEvent
166
>
})
167
>
workflowTaskTimedOut := NewHistoryEventVertex(enumspb.EVENT_TYPE_WORKFLOW_TASK_TIMED_OUT.String())
168
>
workflowTaskTimedOut.SetDataFunc(func(input ...any) any {
169
>
lastEvent := input[0].(*historypb.HistoryEvent)
170
>
lastGeneratedEvent := input[1].(*historypb.HistoryEvent)
171
>
eventID := lastGeneratedEvent.GetEventId() + 1
172
>
version := input[2].(int64)
173
>
historyEvent := getDefaultHistoryEvent(eventID, version)
174
>
historyEvent.EventType = enumspb.EVENT_TYPE_WORKFLOW_TASK_TIMED_OUT
175
>
historyEvent.Attributes = &historypb.HistoryEvent_WorkflowTaskTimedOutEventAttributes{WorkflowTaskTimedOutEventAttributes: &historypb.WorkflowTaskTimedOutEventAttributes{
176
>
ScheduledEventId: lastEvent.GetWorkflowTaskStartedEventAttributes().ScheduledEventId,
177
>
StartedEventId: lastEvent.EventId,
178
>
TimeoutType: enumspb.TIMEOUT_TYPE_SCHEDULE_TO_START,
179
>
}}
180
>
return historyEvent
181
>
})
182
>
workflowTaskComplete := NewHistoryEventVertex(enumspb.EVENT_TYPE_WORKFLOW_TASK_COMPLETED.String())
183
>
workflowTaskComplete.SetDataFunc(func(input ...any) any {
184
>
lastEvent := input[0].(*historypb.HistoryEvent)
185
>
lastGeneratedEvent := input[1].(*historypb.HistoryEvent)
186
>
eventID := lastGeneratedEvent.GetEventId() + 1
187
>
version := input[2].(int64)
188
>
historyEvent := getDefaultHistoryEvent(eventID, version)
189
>
historyEvent.EventType = enumspb.EVENT_TYPE_WORKFLOW_TASK_COMPLETED
190
>
historyEvent.Attributes = &historypb.HistoryEvent_WorkflowTaskCompletedEventAttributes{WorkflowTaskCompletedEventAttributes: &historypb.WorkflowTaskCompletedEventAttributes{
191
>
ScheduledEventId: lastEvent.GetWorkflowTaskStartedEventAttributes().ScheduledEventId,
192
>
StartedEventId: lastEvent.EventId,
193
>
Identity: identity,
194
>
BinaryChecksum: checksum,
195
>
}}
196
>
return historyEvent
197
>
})
198
>
workflowTaskComplete.SetIsStrictOnNextVertex(true)
199
>
workflowTaskComplete.SetMaxNextVertex(2)
200
>
workflowTaskScheduleToStart := NewHistoryEventEdge(workflowTaskSchedule, workflowTaskStart)
201
>
workflowTaskStartToComplete := NewHistoryEventEdge(workflowTaskStart, workflowTaskComplete)
202
>
workflowTaskStartToFail := NewHistoryEventEdge(workflowTaskStart, workflowTaskFail)
203
>
workflowTaskStartToTimedOut := NewHistoryEventEdge(workflowTaskStart, workflowTaskTimedOut)
204
>
workflowTaskFailToSchedule := NewHistoryEventEdge(workflowTaskFail, workflowTaskSchedule)
205
>
workflowTaskFailToSchedule.SetCondition(notPendingWorkflowTask)
206
>
workflowTaskTimedOutToSchedule := NewHistoryEventEdge(workflowTaskTimedOut, workflowTaskSchedule)
207
>
workflowTaskTimedOutToSchedule.SetCondition(notPendingWorkflowTask)
208
>
historyEventModel.AddEdge(workflowTaskScheduleToStart, workflowTaskStartToComplete, workflowTaskStartToFail, workflowTaskStartToTimedOut,
209
>
workflowTaskFailToSchedule, workflowTaskTimedOutToSchedule)
210
>
211
>
// Setup workflow model
212
>
workflowModel := NewHistoryEventModel()
213
>
214
>
workflowStart := NewHistoryEventVertex(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED.String())
215
>
workflowStart.SetDataFunc(func(input ...any) any {
216
>
historyEvent := getDefaultHistoryEvent(1, defaultVersion)
217
>
historyEvent.EventType = enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED
218
>
historyEvent.Attributes = &historypb.HistoryEvent_WorkflowExecutionStartedEventAttributes{WorkflowExecutionStartedEventAttributes: &historypb.WorkflowExecutionStartedEventAttributes{
219
>
WorkflowType: &commonpb.WorkflowType{
220
>
Name: workflowType,
221
>
},
222
>
TaskQueue: &taskqueuepb.TaskQueue{
223
>
Name: taskQueue,
224
>
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
225
>
},
226
>
WorkflowExecutionTimeout: durationpb.New(timeout),
227
>
WorkflowRunTimeout: durationpb.New(timeout),
228
>
WorkflowTaskTimeout: durationpb.New(timeout),
229
>
Identity: identity,
230
>
FirstExecutionRunId: uuid.NewString(),
231
>
Attempt: 1,
232
>
}}
233
>
return historyEvent
234
>
})
235
>
workflowSignal := NewHistoryEventVertex(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED.String())
236
>
workflowSignal.SetDataFunc(func(input ...any) any {
238
>
eventID := lastGeneratedEvent.GetEventId() + 1
239
>
version := input[2].(int64)
240
>
historyEvent := getDefaultHistoryEvent(eventID, version)
241
>
historyEvent.EventType = enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED
242
>
historyEvent.Attributes = &historypb.HistoryEvent_WorkflowExecutionSignaledEventAttributes{WorkflowExecutionSignaledEventAttributes: &historypb.WorkflowExecutionSignaledEventAttributes{
243
>
SignalName: signal,
244
>
Identity: identity,
245
>
}}
246
>
return historyEvent
247
>
})
248
>
workflowComplete := NewHistoryEventVertex(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED.String())
history_event_util.go
249
>
workflowComplete.SetDataFunc(func(input ...any) any {
251
>
eventID := lastEvent.GetEventId() + 1
252
>
version := input[2].(int64)
253
>
historyEvent := getDefaultHistoryEvent(eventID, version)
254
>
historyEvent.EventType = enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED
255
>
historyEvent.Attributes = &historypb.HistoryEvent_WorkflowExecutionCompletedEventAttributes{WorkflowExecutionCompletedEventAttributes: &historypb.WorkflowExecutionCompletedEventAttributes{
256
>
WorkflowTaskCompletedEventId: lastEvent.EventId,
257
>
}}
258
>
return historyEvent
259
>
})
260
>
continueAsNew := NewHistoryEventVertex(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_CONTINUED_AS_NEW.String())
history_event_util.go
261
>
continueAsNew.SetDataFunc(func(input ...any) any {
262
lastEvent := input[0].(*historypb.HistoryEvent)
263
eventID := lastEvent.GetEventId() + 1