300
}
301
303
>
304
>
// 1.b. Filter out tasks from messages beyond the last-desired message.
305
>
historyTasks := make([]*commonspb.HistoryTask, 0, len(response.DlqTasks))
306
>
maxBatchMessageID := int64(persistence.FirstQueueMessageID)
307
>
308
>
for _, task := range response.DlqTasks {
309
>
if task.Metadata.MessageId <= params.MaxMessageID {
310
>
historyTasks = append(historyTasks, task.Payload)
311
>
maxBatchMessageID = max(maxBatchMessageID, task.Metadata.MessageId)
312
>
}
313
}
314
315
// 2. Re-enqueue tasks.
316
>
err = workflow.ExecuteActivity(ctx, reEnqueueTasksActivityName, params, historyTasks).Get(ctx, nil)
workflow.go
317
>
if err != nil {
318
return err
319
}