applyProcessFunc takes a ProcessFunc and handle progress, status messages, etc
(ctx context.Context, p merger.Patch, op merger.Operation, operationId string, cursor *int64, total int64, retry bool)
| 279 | |
| 280 | // applyProcessFunc takes a ProcessFunc and handle progress, status messages, etc |
| 281 | func (pr *Processor) applyProcessFunc(ctx context.Context, p merger.Patch, op merger.Operation, operationId string, cursor *int64, total int64, retry bool) error { |
| 282 | |
| 283 | callback, progressString, completeString, errorString, fields := pr.dataForOperation(p, op) |
| 284 | |
| 285 | pgs := make(chan int64) |
| 286 | var lastProgress float32 |
| 287 | defer close(pgs) |
| 288 | go func() { |
| 289 | for pg := range pgs { |
| 290 | *cursor += pg |
| 291 | progress := float32(*cursor) / float32(total) |
| 292 | if pg < 0 || progress-lastProgress > 0.01 { // Send percent per percent, or if it's negative (error reverted pg value) |
| 293 | log.Logger(pr.GlobalContext).Debug("Sending PG", zap.Float32("pg", progress)) |
| 294 | op.Status(model.NewProcessingStatus(pr.logAsString(progressString, nil, fields...)).SetProgress(progress)) |
| 295 | lastProgress = progress |
| 296 | } |
| 297 | } |
| 298 | }() |
| 299 | var err error |
| 300 | if retry { |
| 301 | err = model.RetryWithCtx(ctx, func(retry int) error { |
| 302 | e := callback(ctx, op, operationId, pgs) |
| 303 | if e != nil { |
| 304 | pr.Logger().Error(errorString, fields...) |
| 305 | op.Status(model.NewProcessingStatus(fmt.Sprintf("%s (%s) - retrying...", errorString, e.Error())).SetError(e)) |
| 306 | } |
| 307 | return e |
| 308 | }, 5*time.Second, 20*time.Second) |
| 309 | } else { |
| 310 | err = callback(ctx, op, operationId, pgs) |
| 311 | } |
| 312 | if err != nil { |
| 313 | fields = append(fields, zap.Error(err)) |
| 314 | if !pr.Silent { |
| 315 | pr.Logger().Error(errorString, fields...) |
| 316 | } |
| 317 | } else { |
| 318 | op.SetProcessed() |
| 319 | if !pr.Silent { |
| 320 | pr.Logger().Info(completeString, fields...) |
| 321 | } |
| 322 | } |
| 323 | |
| 324 | loggerString := completeString |
| 325 | if err != nil { |
| 326 | loggerString = errorString |
| 327 | } |
| 328 | var end float32 |
| 329 | if total > 0 { |
| 330 | end = float32(*cursor) / float32(total) |
| 331 | } |
| 332 | op.Status(model.NewProcessingStatus(pr.logAsString(loggerString, err, fields...)).SetError(err).SetProgress(end)) |
| 333 | |
| 334 | return err |
| 335 | } |
| 336 | |
| 337 | // logAsStrings transforms zap Fields to string |
| 338 | func (pr *Processor) logAsString(msg string, err error, fields ...zapcore.Field) string { |
no test coverage detected