Function HookAndRemove

internal/task_group/transfer.go:24–73

(ctx context.Context, dstPath string, payloads ...any)

type DstPathToHook string

func HookAndRemove(ctx context.Context, dstPath string, payloads ...any) {
	dstStorage, dstActualPath, err := op.GetStorageAndActualPath(dstPath)
	if err != nil {
		log.Error(errors.WithMessage(err, "failed get dst storage"))
		return
	}
	dstNeedHandleHook := setting.GetBool(conf.HandleHookAfterWriting)
	dstHandleHookLimit := setting.GetFloat(conf.HandleHookRateLimit, .0)
	var listLimiter *rate.Limiter
	if dstNeedHandleHook && dstHandleHookLimit > .0 {
		listLimiter = rate.NewLimiter(rate.Limit(dstHandleHookLimit), 1)
	}
	hookedPaths := make(map[string]struct{})
	handleHook := func(actualPath string) {
		if _, ok := hookedPaths[actualPath]; ok {
			return
		}
		if listLimiter != nil {
			_ = listLimiter.Wait(ctx)
		}
		files, e := op.List(ctx, dstStorage, actualPath, model.ListArgs{SkipHook: true})
		if e != nil {
			log.Errorf("failed handle objs update hook: %v", e)
		} else {
			op.HandleObjsUpdateHook(ctx, utils.GetFullPath(dstStorage.GetStorage().MountPath, actualPath), files)
			hookedPaths[actualPath] = struct{}{}
		}
	}
	if dstNeedHandleHook {
		handleHook(dstActualPath)
	}
	for _, payload := range payloads {
		switch p := payload.(type) {
		case DstPathToHook:
			if dstNeedHandleHook {
				handleHook(string(p))
			}
		case SrcPathToRemove:
			srcStorage, srcActualPath, err := op.GetStorageAndActualPath(string(p))
			if err != nil {
				log.Error(errors.WithMessage(err, "failed get src storage"))
				continue
			}
			err = verifyAndRemove(ctx, srcStorage, dstStorage, srcActualPath, dstActualPath)
			if err != nil {
				log.Error(err)
			}
		}
	}
}

func verifyAndRemove(ctx context.Context, srcStorage, dstStorage driver.Driver, srcPath, dstPath string) error {
	srcObj, err := op.GetUnwrap(ctx, srcStorage, srcPath)