22
)
23
24
>
func CatchupWorkflow(ctx workflow.Context, params CatchUpParams) (CatchUpOutput, error) {
catchup_workflow.go
25
>
if err := validateCatchupParams(¶ms); err != nil {
26
return CatchUpOutput{}, err
27
}
28
30
>
InitialInterval: time.Second,
31
>
MaximumInterval: time.Second,
32
>
BackoffCoefficient: 1,
33
>
}
34
>
ao := workflow.ActivityOptions{
35
>
StartToCloseTimeout: time.Hour,
36
>
HeartbeatTimeout: time.Second * 10,
37
>
RetryPolicy: retryPolicy,
38
>
}
39
>
ctx1 := workflow.WithActivityOptions(ctx, ao)
40
>
41
>
var a *activities
42
>
err := workflow.ExecuteActivity(ctx1, a.WaitCatchup, params).Get(ctx, nil)
43
>
if err != nil {
44
return CatchUpOutput{}, err
45
}
46
48
}
49
51
>
if len(params.Namespace) == 0 {
52
return temporal.NewNonRetryableApplicationError("InvalidArgument: Namespace is required", "InvalidArgument", nil)
53
}
55
return temporal.NewNonRetryableApplicationError("InvalidArgument: CatchupCluster is required", "InvalidArgument", nil)
56
}
57
59
}