package cmdline import ( "fmt" "os" "testing" "time" "github.com/spf13/cobra" "gitlink.org.cn/cloudream/common/pkgs/logger" "gitlink.org.cn/cloudream/jcs-pub/client/internal/accessstat" "gitlink.org.cn/cloudream/jcs-pub/client/internal/accesstoken" "gitlink.org.cn/cloudream/jcs-pub/client/internal/config" "gitlink.org.cn/cloudream/jcs-pub/client/internal/db" "gitlink.org.cn/cloudream/jcs-pub/client/internal/downloader" "gitlink.org.cn/cloudream/jcs-pub/client/internal/downloader/strategy" "gitlink.org.cn/cloudream/jcs-pub/client/internal/http" "gitlink.org.cn/cloudream/jcs-pub/client/internal/metacache" "gitlink.org.cn/cloudream/jcs-pub/client/internal/mount" "gitlink.org.cn/cloudream/jcs-pub/client/internal/mount/vfstest" "gitlink.org.cn/cloudream/jcs-pub/client/internal/services" "gitlink.org.cn/cloudream/jcs-pub/client/internal/spacesyncer" "gitlink.org.cn/cloudream/jcs-pub/client/internal/uploader" stgglb "gitlink.org.cn/cloudream/jcs-pub/common/globals" "gitlink.org.cn/cloudream/jcs-pub/common/models/datamap" "gitlink.org.cn/cloudream/jcs-pub/common/pkgs/connectivity" "gitlink.org.cn/cloudream/jcs-pub/common/pkgs/publock" "gitlink.org.cn/cloudream/jcs-pub/common/pkgs/storage/pool" "gitlink.org.cn/cloudream/jcs-pub/common/pkgs/sysevent" ) // 初始化函数,将ServeHTTP命令注册到命令列表中。 func init() { var configPath string var opt serveHTTPOptions cmd := cobra.Command{ Use: "vfstest", Short: "test vfs", Run: func(cmd *cobra.Command, args []string) { vfsTest(configPath, opt) }, } cmd.Flags().StringVarP(&configPath, "config", "c", "", "config file path") cmd.Flags().BoolVarP(&opt.DisableHTTP, "no-http", "", false, "disable http server") cmd.Flags().StringVarP(&opt.HTTPListenAddr, "listen", "", "", "http listen address, will override config file") cmd.Flags().BoolVarP(&opt.DisableMount, "no-mount", "", false, "disable mount") cmd.Flags().StringVarP(&opt.MountPoint, "mount", "", "", "mount point, will override config file") RootCmd.AddCommand(&cmd) } func vfsTest(configPath string, opts serveHTTPOptions) { err := config.Init(configPath) if err != nil { fmt.Printf("init config failed, err: %s", err.Error()) os.Exit(1) } err = logger.Init(&config.Cfg().Logger) if err != nil { fmt.Printf("init logger failed, err: %s", err.Error()) os.Exit(1) } stgglb.InitLocal(config.Cfg().Local) stgglb.StandaloneMode = opts.Standalone || config.Cfg().AccessToken == nil var accToken *accesstoken.Keeper if !opts.Standalone { tempCli, err := config.Cfg().CoordinatorRPC.BuildTempClient() if err != nil { logger.Warnf("build coordinator rpc temp client: %v", err) os.Exit(1) } accToken, err = accesstoken.New(*config.Cfg().AccessToken, tempCli) tempCli.Release() if err != nil { logger.Warnf("new access token keeper: %v", err) os.Exit(1) } hubRPCCfg, err := config.Cfg().HubRPC.Build(accToken) if err != nil { logger.Warnf("build hub rpc pool config: %v", err) os.Exit(1) } corRPCCfg, err := config.Cfg().CoordinatorRPC.Build(accToken) if err != nil { logger.Warnf("build coordinator rpc pool config: %v", err) os.Exit(1) } stgglb.UserID = accToken.GetToken().UserID stgglb.InitPools(hubRPCCfg, corRPCCfg) } else { stgglb.UserID = 0 accToken = accesstoken.NewDisabled() } accTokenChan := accToken.Start() defer accToken.Stop() // 数据库 db, err := db.NewDB(&config.Cfg().DB) if err != nil { logger.Fatalf("new db failed, err: %s", err.Error()) } // 初始化系统事件发布器 evtPub, err := sysevent.NewPublisher(config.Cfg().SysEvent, &datamap.SourceClient{ UserID: stgglb.UserID, }) if err != nil { logger.Errorf("new sysevent publisher: %v", err) os.Exit(1) } evtPubChan := evtPub.Start() defer evtPub.Stop() // 连接性信息收集 var conCol *connectivity.Collector if stgglb.StandaloneMode { conCol = connectivity.NewDisabled() } else { conCol = connectivity.NewEnabled(config.Cfg().Connectivity) } conColChan := conCol.Start() defer conCol.Stop() conCol.CollectInPlace() // 元数据缓存 metaCacheHost := metacache.NewHost(db) go metaCacheHost.Serve() stgMeta := metaCacheHost.AddStorageMeta() hubMeta := metaCacheHost.AddHubMeta() conMeta := metaCacheHost.AddConnectivity() // 公共锁 publock := publock.NewService() // 访问统计 acStat := accessstat.NewAccessStat(accessstat.Config{ // TODO 考虑放到配置里 ReportInterval: time.Second * 10, }, db) acStatChan := acStat.Start() defer acStat.Stop() // 存储管理器 stgPool := pool.NewPool() // 下载策略 strgSel := strategy.NewSelector(config.Cfg().DownloadStrategy, stgMeta, hubMeta, conMeta) // 下载器 dlder := downloader.NewDownloader(config.Cfg().Downloader, conCol, stgPool, strgSel, db) // 上传器 uploader := uploader.NewUploader(publock, conCol, stgPool, stgMeta, db) // 用户空间同步功能 spaceSync := spacesyncer.New(db, stgPool, stgMeta) spaceSyncChan := spaceSync.Start() defer spaceSync.Stop() // 挂载 mntCfg := config.Cfg().Mount if !opts.DisableMount && mntCfg != nil && mntCfg.Enabled { if opts.MountPoint != "" { mntCfg.MountPoint = opts.MountPoint } } else { mntCfg = nil } mnt := mount.NewMount(mntCfg, db, uploader, dlder) mntChan := mnt.Start() defer mnt.Stop() svc := services.NewService(publock, dlder, acStat, uploader, strgSel, stgMeta, db, evtPub, mnt, stgPool, spaceSync, nil) // HTTP接口 httpCfgJSON := config.Cfg().HTTP if !opts.DisableHTTP && httpCfgJSON != nil && httpCfgJSON.Enabled { if opts.HTTPListenAddr != "" { httpCfgJSON.Listen = opts.HTTPListenAddr } } else { httpCfgJSON = &http.ConfigJSON{ Enabled: false, } } httpCfg, err := httpCfgJSON.Build() if err != nil { logger.Errorf("build http config: %v", err) os.Exit(1) } httpSvr := http.NewServer(httpCfg, svc) httpChan := httpSvr.Start() defer httpSvr.Stop() go func() { <-time.After(5 * time.Second) testing.Main(nil, []testing.InternalTest{ {"VFS", func(t *testing.T) { vfstest.RunTests(t, mnt) }}, }, nil, nil) }() /// 开始监听各个模块的事件 accTokenEvt := accTokenChan.Receive() evtPubEvt := evtPubChan.Receive() conColEvt := conColChan.Receive() acStatEvt := acStatChan.Receive() spaceSyncEvt := spaceSyncChan.Receive() httpEvt := httpChan.Receive() mntEvt := mntChan.Receive() loop: for { select { case e := <-accTokenEvt.Chan(): if e.Err != nil { logger.Errorf("receive access token event: %v", err) break loop } switch e := e.Value.(type) { case accesstoken.ExitEvent: if e.Err != nil { logger.Errorf("access token keeper exit with error: %v", err) } else { logger.Info("access token keeper exited") } break loop } accTokenEvt = accTokenChan.Receive() case e := <-evtPubEvt.Chan(): if e.Err != nil { logger.Errorf("receive publisher event: %v", err) break loop } switch val := e.Value.(type) { case sysevent.PublishError: logger.Errorf("publishing event: %v", val) case sysevent.PublisherExited: if val.Err != nil { logger.Errorf("publisher exited with error: %v", val.Err) } else { logger.Info("publisher exited") } break loop case sysevent.OtherError: logger.Errorf("sysevent: %v", val) } evtPubEvt = evtPubChan.Receive() case e := <-conColEvt.Chan(): if e.Err != nil { logger.Errorf("receive connectivity event: %v", err) break loop } switch e := e.Value.(type) { case connectivity.ExitEvent: if e.Err != nil { logger.Errorf("connectivity collector exited with error: %v", e.Err) } else { logger.Info("connectivity collector exited") } break loop case connectivity.CollectedEvent: } case e := <-acStatEvt.Chan(): if e.Err != nil { logger.Errorf("receive access stat event: %v", err) break loop } switch e := e.Value.(type) { case accessstat.ExitEvent: logger.Infof("access stat exited, err: %v", e.Err) break loop } acStatEvt = acStatChan.Receive() case e := <-spaceSyncEvt.Chan(): if e.Err != nil { logger.Errorf("receive space sync event: %v", err) } spaceSyncEvt = spaceSyncChan.Receive() case e := <-httpEvt.Chan(): if e.Err != nil { logger.Errorf("receive http event: %v", err) break loop } switch e := e.Value.(type) { case http.ExitEvent: logger.Infof("http server exited, err: %v", e.Err) break loop } httpEvt = httpChan.Receive() case e := <-mntEvt.Chan(): if e.Err != nil { logger.Errorf("receive mount event: %v", e.Err) break loop } switch e := e.Value.(type) { case mount.ExitEvent: logger.Infof("mount exited, err: %v", e.Err) break loop } mntEvt = mntChan.Receive() } } }