Method RunWithNextTaskCallback

internal/fs/copy_move.go:179–264

(f func(nextTask *FileTransferTask) error)

}

func (t *FileTransferTask) RunWithNextTaskCallback(f func(nextTask *FileTransferTask) error) error {
	t.Status = "getting src object"
	srcObj, err := op.Get(t.Ctx(), t.SrcStorage, t.SrcActualPath)
	if err != nil {
		return errors.WithMessagef(err, "failed get src [%s] file", t.SrcActualPath)
	}

	if srcObj.IsDir() {
		t.Status = "src object is dir, listing objs"
		objs, err := op.List(t.Ctx(), t.SrcStorage, t.SrcActualPath, model.ListArgs{})
		if err != nil {
			return errors.WithMessagef(err, "failed list src [%s] objs", t.SrcActualPath)
		}
		dstActualPath := stdpath.Join(t.DstActualPath, srcObj.GetName())
		task_group.TransferCoordinator.AppendPayload(t.groupID, task_group.DstPathToHook(dstActualPath))

		existedObjs := make(map[string]bool)
		if t.TaskType == merge {
			dstObjs, err := op.List(t.Ctx(), t.DstStorage, dstActualPath, model.ListArgs{})
			if err != nil && !errors.Is(err, errs.ObjectNotFound) {
				// 目标文件夹不存在的情况不是错误，会在之后新建文件夹
				// 这种情况显然不需要统计existedObjs，dstObjs保持为nil，下面这个for将不会执行
				return errors.WithMessagef(err, "failed list dst [%s] objs", dstActualPath)
			}
			for _, obj := range dstObjs {
				if err := t.Ctx().Err(); err != nil {
					return err
				}
				if !obj.IsDir() {
					existedObjs[obj.GetName()] = true
				}
			}
		}

		for _, obj := range objs {
			if err := t.Ctx().Err(); err != nil {
				return err
			}

			if t.TaskType == merge && !obj.IsDir() && existedObjs[obj.GetName()] {
				// skip existed file
				continue
			}

			err = f(&FileTransferTask{
				TaskType: t.TaskType,
				TaskData: TaskData{
					TaskExtension: task.TaskExtension{
						Creator: t.Creator,
						ApiUrl:  t.ApiUrl,
					},
					SrcStorage:    t.SrcStorage,
					DstStorage:    t.DstStorage,
					SrcActualPath: stdpath.Join(t.SrcActualPath, obj.GetName()),
					DstActualPath: dstActualPath,
					SrcStorageMp:  t.SrcStorageMp,
					DstStorageMp:  t.DstStorageMp,
				},