106
}
107
109
>
tag.Int("items", totalCount),
110
>
tag.Duration("timeout", timeout))
111
>
112
>
startTime := time.Now()
113
>
defer func() { metrics.FinalizerLatency.With(f.metricsHandler).Record(time.Since(startTime)) }()
114
116
>
defer cancel()
117
>
118
>
pool := goro.NewAdaptivePool(cclock.NewRealTimeSource(), 5, 15, 10*time.Millisecond, 10)
119
>
defer pool.Stop()
120
>
121
>
completionChannel := make(chan struct{})
122
>
go func() {
123
>
for _, callback := range f.callbacks {
124
>
// NOTE: Once `pool.Stop` is called, any remaining calls to `pool.Do` will do nothing.
125
>
pool.Do(func() {
126
>
defer func() { completionChannel <- struct{}{} }()
127
>
_ = callback(ctx)
128
})
129
}