package spacesyncer import ( "context" "fmt" "io" "time" "gitlink.org.cn/cloudream/common/pkgs/logger" "gitlink.org.cn/cloudream/common/pkgs/trie" "gitlink.org.cn/cloudream/jcs-pub/common/pkgs/ioswitch/exec" "gitlink.org.cn/cloudream/jcs-pub/common/pkgs/ioswitch2" "gitlink.org.cn/cloudream/jcs-pub/common/pkgs/ioswitch2/parser" stgtypes "gitlink.org.cn/cloudream/jcs-pub/common/pkgs/storage/types" jcstypes "gitlink.org.cn/cloudream/jcs-pub/common/types" ) func executeFull(syncer *SpaceSyncer, task *task) { log := logger.WithField("Mod", logMod).WithField("TaskID", task.Task.TaskID) startTime := time.Now() log.Infof("begin full sync task") defer func() { log.Infof("full sync task finished, time: %v", time.Since(startTime)) }() srcSpace := syncer.spaceMeta.Get(task.Task.SrcUserSpaceID) if srcSpace == nil { log.Warnf("src space %v not found", task.Task.SrcUserSpaceID) return } dstSpaceIDs := make([]jcstypes.UserSpaceID, len(task.Task.Dests)) for i := range task.Task.Dests { dstSpaceIDs[i] = task.Task.Dests[i].DestUserSpaceID } dstSpaces := syncer.spaceMeta.GetMany(dstSpaceIDs) for i := range dstSpaces { if dstSpaces[i] == nil { log.Warnf("dst space %v not found", dstSpaceIDs[i]) return } } srcBase, err := syncer.stgPool.GetBaseStore(srcSpace) if err != nil { log.Warnf("get src base store: %v", err) return } filter := buildFilter(task) srcDirReader := srcBase.ReadDir(task.Task.SrcPath) defer srcDirReader.Close() srcDirTree := trie.NewTrie[*stgtypes.DirEntry]() fileCnt := 0 for { isEOF := false ft := ioswitch2.NewFromTo() cnt := 0 for { e, err := srcDirReader.Next() if err == io.EOF { isEOF = true break } if err != nil { log.Warnf("read src dir: %v", err) return } rela := e.Path.Clone() rela.PopFrontN(task.Task.SrcPath.Len()) ne := e ne.Path = rela.Clone() if !filter(ne) { continue } if e.IsDir { // 如果是一个目录,则创建对应的Dir节点,且在创建过程中清除掉路径上的Dir信息(仅保留最后一个Dir节点) createDirNode(srcDirTree, rela.Comps(), &e) continue } fmt.Printf("rela: %v\n", rela) // 如果是一个文件,那么它路径上的目录都可以在写入时一并创建,所以可以清理掉路径上的Dir节点 removeDirNode(srcDirTree, rela.Comps()) ft.AddFrom(ioswitch2.NewFromBaseStore(*srcSpace, e.Path)) for i, dst := range dstSpaces { dstPath := task.Task.Dests[i].DestPath.Clone() dstPath.Concat(rela) ft.AddTo(ioswitch2.NewToBaseStore(*dst, dstPath)) } cnt++ fileCnt++ // 每一批转发50个文件 if cnt > 50 { break } } if len(ft.Froms) > 0 { planBld := exec.NewPlanBuilder() err := parser.Parse(ft, planBld) if err != nil { log.Warnf("parse fromto to plan: %v", err) return } execCtx := exec.NewWithContext(task.Context) exec.SetValueByType(execCtx, syncer.stgPool) _, err = planBld.Execute(execCtx).Wait(context.Background()) if err != nil { log.Warnf("execute plan: %v", err) return } } if isEOF { break } } log.Infof("%v files synced", fileCnt) if !task.Task.Options.NoEmptyDirectories { dstBases := make([]stgtypes.BaseStore, len(dstSpaces)) for i := range dstSpaces { dstBases[i], err = syncer.stgPool.GetBaseStore(dstSpaces[i]) if err != nil { log.Warnf("get dst base store: %v", err) continue } } srcDirTree.Iterate(func(path []string, node *trie.Node[*stgtypes.DirEntry], isWordNode bool) trie.VisitCtrl { if node.Value == nil { return trie.VisitContinue } for i, base := range dstBases { if base != nil { dirPath := task.Task.Dests[i].DestPath.Clone() dirPath.Push(path...) err := base.Mkdir(dirPath) if err != nil { log.Warnf("mkdir %v at user space %v: %v", dirPath, dstSpaces[i].String(), err) } } } return trie.VisitContinue }) } }