package downloader import ( "fmt" "io" lru "github.com/hashicorp/golang-lru/v2" "gitlink.org.cn/cloudream/common/pkgs/iterator" "gitlink.org.cn/cloudream/jcs-pub/client/internal/db" "gitlink.org.cn/cloudream/jcs-pub/client/internal/downloader/strategy" clitypes "gitlink.org.cn/cloudream/jcs-pub/client/types" "gitlink.org.cn/cloudream/jcs-pub/common/pkgs/connectivity" "gitlink.org.cn/cloudream/jcs-pub/common/pkgs/storage/pool" ) const ( DefaultMaxStripCacheCount = 128 ) type DownloadIterator = iterator.Iterator[*Downloading] type DownloadReqeust struct { ObjectID clitypes.ObjectID Offset int64 Length int64 } type downloadReqeust2 struct { Detail *clitypes.ObjectDetail Raw DownloadReqeust } type Downloading struct { Object *clitypes.Object File io.ReadCloser // 文件流,如果文件不存在,那么为nil Request DownloadReqeust } type Downloader struct { strips *StripCache cfg Config conn *connectivity.Collector stgPool *pool.Pool selector *strategy.Selector db *db.DB } func NewDownloader(cfg Config, conn *connectivity.Collector, stgPool *pool.Pool, sel *strategy.Selector, db *db.DB) *Downloader { if cfg.MaxStripCacheCount == 0 { cfg.MaxStripCacheCount = DefaultMaxStripCacheCount } ch, _ := lru.New[ECStripKey, ObjectECStrip](cfg.MaxStripCacheCount) return &Downloader{ strips: ch, cfg: cfg, conn: conn, stgPool: stgPool, selector: sel, db: db, } } func (d *Downloader) DownloadObjects(reqs []DownloadReqeust) DownloadIterator { objIDs := make([]clitypes.ObjectID, len(reqs)) for i, req := range reqs { objIDs[i] = req.ObjectID } if len(objIDs) == 0 { return iterator.Empty[*Downloading]() } objDetails, err := db.DoTx11(d.db, d.db.Object().BatchGetDetails, objIDs) if err != nil { return iterator.FuseError[*Downloading](fmt.Errorf("request to db: %w", err)) } detailsMap := make(map[clitypes.ObjectID]*clitypes.ObjectDetail) for _, detail := range objDetails { d := detail detailsMap[detail.Object.ObjectID] = &d } req2s := make([]downloadReqeust2, len(reqs)) for i, req := range reqs { req2s[i] = downloadReqeust2{ Detail: detailsMap[req.ObjectID], Raw: req, } } return NewDownloadObjectIterator(d, req2s) } func (d *Downloader) DownloadObjectByDetail(detail clitypes.ObjectDetail, off int64, length int64) (*Downloading, error) { req2s := []downloadReqeust2{{ Detail: &detail, Raw: DownloadReqeust{ ObjectID: detail.Object.ObjectID, Offset: off, Length: length, }, }} iter := NewDownloadObjectIterator(d, req2s) return iter.MoveNext() } func (d *Downloader) DownloadPackage(pkgID clitypes.PackageID, prefix string) (clitypes.Package, DownloadIterator, error) { pkg, details, err := db.DoTx02(d.db, func(tx db.SQLContext) (clitypes.Package, []clitypes.ObjectDetail, error) { pkg, err := d.db.Package().GetByID(tx, pkgID) if err != nil { return clitypes.Package{}, nil, err } var details []clitypes.ObjectDetail if prefix != "" { objs, err := d.db.Object().GetWithPathPrefix(tx, pkgID, prefix) if err != nil { return clitypes.Package{}, nil, err } objIDs := make([]clitypes.ObjectID, len(objs)) for i, obj := range objs { objIDs[i] = obj.ObjectID } allBlocks, err := d.db.ObjectBlock().BatchGetByObjectID(tx, objIDs) if err != nil { return clitypes.Package{}, nil, err } allPinnedObjs, err := d.db.PinnedObject().BatchGetByObjectID(tx, objIDs) if err != nil { return clitypes.Package{}, nil, err } details = make([]clitypes.ObjectDetail, 0, len(objs)) for _, obj := range objs { detail := clitypes.ObjectDetail{ Object: obj, } details = append(details, detail) } clitypes.DetailsFillObjectBlocks(details, allBlocks) clitypes.DetailsFillPinnedAt(details, allPinnedObjs) } else { details, err = d.db.Object().GetPackageObjectDetails(tx, pkgID) if err != nil { return clitypes.Package{}, nil, err } } return pkg, details, nil }) if err != nil { return clitypes.Package{}, nil, err } req2s := make([]downloadReqeust2, len(details)) for i, objDetail := range details { dt := objDetail req2s[i] = downloadReqeust2{ Detail: &dt, Raw: DownloadReqeust{ ObjectID: objDetail.Object.ObjectID, Offset: 0, Length: objDetail.Object.Size, }, } } return pkg, NewDownloadObjectIterator(d, req2s), nil } type ObjectECStrip struct { Data []byte ObjectFileHash clitypes.FileHash // 添加这条缓存时,Object的FileHash } type ECStripKey struct { ObjectID clitypes.ObjectID StripIndex int64 } type StripCache = lru.Cache[ECStripKey, ObjectECStrip]