52
}
53
54
>
func newBackfillerWithState(ctx chasm.MutableContext, state *schedulerpb.BackfillerState) *Backfiller {
backfiller.go
55
>
backfiller := &Backfiller{
56
>
BackfillerState: state,
57
>
EventLog: chasm.NewComponentField(ctx, NewEventLog(ctx)),
58
>
}
59
>
backfiller.scheduleTask(ctx, chasm.TaskScheduledTimeImmediate)
60
>
return backfiller
61
>
}
62
63
// scheduleTask schedules a BackfillerTask at the given time.
64
>
func (b *Backfiller) scheduleTask(ctx chasm.MutableContext, scheduledTime time.Time) {
backfiller.go
65
>
b.getOrCreateEventLog(ctx).LogEvent(ctx,
66
>
fmt.Sprintf("scheduled backfillerTask for %s", scheduledTime.Format(time.RFC3339)))
67
>
ctx.AddTask(b, chasm.TaskAttributes{
68
>
ScheduledTime: scheduledTime,
69
>
}, &schedulerpb.BackfillerTask{})
70
>
}
71
72
func (b *Backfiller) LifecycleState(ctx chasm.Context) chasm.LifecycleState {