perf(stream): improve file stream range reading and caching mechanism

drivers/115/driver.go

@@ -186,9 +186,7 @@ func (d *Pan115) Put(ctx context.Context, dstDir model.Obj, stream model.FileStr
 	preHash = strings.ToUpper(preHash)
 	fullHash := stream.GetHash().GetHash(utils.SHA1)
 	if len(fullHash) != utils.SHA1.Width {
-		cacheFileProgress := model.UpdateProgressWithRange(up, 0, 50)
-		up = model.UpdateProgressWithRange(up, 50, 100)
-		_, fullHash, err = streamPkg.CacheFullInTempFileAndHash(stream, cacheFileProgress, utils.SHA1)
+		_, fullHash, err = streamPkg.CacheFullAndHash(stream, &up, utils.SHA1)
 		if err != nil {
 			return nil, err
 		}

drivers/115/util.go

@@ -321,7 +321,7 @@ func (d *Pan115) UploadByMultipart(ctx context.Context, params *driver115.Upload
 		err       error
 	)
 
-	tmpF, err := s.CacheFullInTempFile()
+	tmpF, err := s.CacheFullAndWriter(&up, nil)
 	if err != nil {
 		return nil, err
 	}

drivers/115_open/driver.go

@@ -239,9 +239,7 @@ func (d *Open115) Put(ctx context.Context, dstDir model.Obj, file model.FileStre
 	}
 	sha1 := file.GetHash().GetHash(utils.SHA1)
 	if len(sha1) != utils.SHA1.Width {
-		cacheFileProgress := model.UpdateProgressWithRange(up, 0, 50)
-		up = model.UpdateProgressWithRange(up, 50, 100)
-		_, sha1, err = stream.CacheFullInTempFileAndHash(file, cacheFileProgress, utils.SHA1)
+		_, sha1, err = stream.CacheFullAndHash(file, &up, utils.SHA1)
 		if err != nil {
 			return err
 		}

drivers/115_open/upload.go

@@ -86,13 +86,14 @@ func (d *Open115) multpartUpload(ctx context.Context, stream model.FileStreamer,
 
 	fileSize := stream.GetSize()
 	chunkSize := calPartSize(fileSize)
-	partNum := (stream.GetSize() + chunkSize - 1) / chunkSize
-	parts := make([]oss.UploadPart, partNum)
-	offset := int64(0)
-	ss, err := streamPkg.NewStreamSectionReader(stream, int(chunkSize))
+	ss, err := streamPkg.NewStreamSectionReader(stream, int(chunkSize), &up)
 	if err != nil {
 		return err
 	}
+
+	partNum := (stream.GetSize() + chunkSize - 1) / chunkSize
+	parts := make([]oss.UploadPart, partNum)
+	offset := int64(0)
 	for i := int64(1); i <= partNum; i++ {
 		if utils.IsCanceled(ctx) {
 			return ctx.Err()
@@ -119,7 +120,7 @@ func (d *Open115) multpartUpload(ctx context.Context, stream model.FileStreamer,
 			retry.Attempts(3),
 			retry.DelayType(retry.BackOffDelay),
 			retry.Delay(time.Second))
-		ss.RecycleSectionReader(rd)
+		ss.FreeSectionReader(rd)
 		if err != nil {
 			return err
 		}

drivers/123/driver.go

@@ -182,9 +182,7 @@ func (d *Pan123) Put(ctx context.Context, dstDir model.Obj, file model.FileStrea
 	etag := file.GetHash().GetHash(utils.MD5)
 	var err error
 	if len(etag) < utils.MD5.Width {
-		cacheFileProgress := model.UpdateProgressWithRange(up, 0, 50)
-		up = model.UpdateProgressWithRange(up, 50, 100)
-		_, etag, err = stream.CacheFullInTempFileAndHash(file, cacheFileProgress, utils.MD5)
+		_, etag, err = stream.CacheFullAndHash(file, &up, utils.MD5)
 		if err != nil {
 			return err
 		}

drivers/123/upload.go

@@ -81,6 +81,12 @@ func (d *Pan123) newUpload(ctx context.Context, upReq *UploadResp, file model.Fi
 	if size > chunkSize {
 		chunkCount = int((size + chunkSize - 1) / chunkSize)
 	}
+
+	ss, err := stream.NewStreamSectionReader(file, int(chunkSize), &up)
+	if err != nil {
+		return err
+	}
+
 	lastChunkSize := size % chunkSize
 	if lastChunkSize == 0 {
 		lastChunkSize = chunkSize
@@ -92,10 +98,6 @@ func (d *Pan123) newUpload(ctx context.Context, upReq *UploadResp, file model.Fi
 		batchSize = 10
 		getS3UploadUrl = d.getS3PreSignedUrls
 	}
-	ss, err := stream.NewStreamSectionReader(file, int(chunkSize))
-	if err != nil {
-		return err
-	}
 
 	thread := min(int(chunkCount), d.UploadThread)
 	threadG, uploadCtx := errgroup.NewOrderedGroupWithContext(ctx, thread,
@@ -180,7 +182,7 @@ func (d *Pan123) newUpload(ctx context.Context, upReq *UploadResp, file model.Fi
 					return nil
 				},
 				After: func(err error) {
-					ss.RecycleSectionReader(reader)
+					ss.FreeSectionReader(reader)
 				},
 			})
 		}

drivers/123_open/driver.go

@@ -132,9 +132,7 @@ func (d *Open123) Put(ctx context.Context, dstDir model.Obj, file model.FileStre
 	// etag 文件md5
 	etag := file.GetHash().GetHash(utils.MD5)
 	if len(etag) < utils.MD5.Width {
-		cacheFileProgress := model.UpdateProgressWithRange(up, 0, 50)
-		up = model.UpdateProgressWithRange(up, 50, 100)
-		_, etag, err = stream.CacheFullInTempFileAndHash(file, cacheFileProgress, utils.MD5)
+		_, etag, err = stream.CacheFullAndHash(file, &up, utils.MD5)
 		if err != nil {
 			return nil, err
 		}

drivers/123_open/upload.go

@@ -46,17 +46,19 @@ func (d *Open123) Upload(ctx context.Context, file model.FileStreamer, createRes
 	uploadDomain := createResp.Data.Servers[0]
 	size := file.GetSize()
 	chunkSize := createResp.Data.SliceSize
+
+	ss, err := stream.NewStreamSectionReader(file, int(chunkSize), &up)
+	if err != nil {
+		return err
+	}
+
 	uploadNums := (size + chunkSize - 1) / chunkSize
 	thread := min(int(uploadNums), d.UploadThread)
 	threadG, uploadCtx := errgroup.NewOrderedGroupWithContext(ctx, thread,
 		retry.Attempts(3),
 		retry.Delay(time.Second),
 		retry.DelayType(retry.BackOffDelay))
 
-	ss, err := stream.NewStreamSectionReader(file, int(chunkSize))
-	if err != nil {
-		return err
-	}
 	for partIndex := range uploadNums {
 		if utils.IsCanceled(uploadCtx) {
 			break
@@ -157,7 +159,7 @@ func (d *Open123) Upload(ctx context.Context, file model.FileStreamer, createRes
 				return nil
 			},
 			After: func(err error) {
-				ss.RecycleSectionReader(reader)
+				ss.FreeSectionReader(reader)
 			},
 		})
 	}

drivers/139/driver.go

@@ -522,9 +522,7 @@ func (d *Yun139) Put(ctx context.Context, dstDir model.Obj, stream model.FileStr
 		var err error
 		fullHash := stream.GetHash().GetHash(utils.SHA256)
 		if len(fullHash) != utils.SHA256.Width {
-			cacheFileProgress := model.UpdateProgressWithRange(up, 0, 50)
-			up = model.UpdateProgressWithRange(up, 50, 100)
-			_, fullHash, err = streamPkg.CacheFullInTempFileAndHash(stream, cacheFileProgress, utils.SHA256)
+			_, fullHash, err = streamPkg.CacheFullAndHash(stream, &up, utils.SHA256)
 			if err != nil {
 				return err
 			}

drivers/189_tv/utils.go

@@ -5,17 +5,19 @@ import (
 	"encoding/base64"
 	"encoding/xml"
 	"fmt"
-	"github.com/skip2/go-qrcode"
 	"io"
 	"net/http"
 	"strconv"
 	"strings"
 	"time"
 
+	"github.com/skip2/go-qrcode"
+
 	"github.com/OpenListTeam/OpenList/v4/drivers/base"
 	"github.com/OpenListTeam/OpenList/v4/internal/driver"
 	"github.com/OpenListTeam/OpenList/v4/internal/model"
 	"github.com/OpenListTeam/OpenList/v4/internal/op"
+	"github.com/OpenListTeam/OpenList/v4/internal/stream"
 	"github.com/OpenListTeam/OpenList/v4/pkg/utils"
 
 	"github.com/go-resty/resty/v2"
@@ -311,11 +313,14 @@ func (y *Cloud189TV) RapidUpload(ctx context.Context, dstDir model.Obj, stream m
 
 // 旧版本上传，家庭云不支持覆盖
 func (y *Cloud189TV) OldUpload(ctx context.Context, dstDir model.Obj, file model.FileStreamer, up driver.UpdateProgress, isFamily bool, overwrite bool) (model.Obj, error) {
-	tempFile, err := file.CacheFullInTempFile()
-	if err != nil {
-		return nil, err
+	fileMd5 := file.GetHash().GetHash(utils.MD5)
+	var tempFile = file.GetFile()
+	var err error
+	if len(fileMd5) != utils.MD5.Width {
+		tempFile, fileMd5, err = stream.CacheFullAndHash(file, &up, utils.MD5)
+	} else if tempFile == nil {
+		tempFile, err = file.CacheFullAndWriter(&up, nil)
 	}
-	fileMd5, err := utils.HashFile(utils.MD5, tempFile)
 	if err != nil {
 		return nil, err
 	}
@@ -345,7 +350,7 @@ func (y *Cloud189TV) OldUpload(ctx context.Context, dstDir model.Obj, file model
 			header["Edrive-UploadFileId"] = fmt.Sprint(status.UploadFileId)
 		}
 
-		_, err := y.put(ctx, status.FileUploadUrl, header, true, io.NopCloser(tempFile), isFamily)
+		_, err := y.put(ctx, status.FileUploadUrl, header, true, tempFile, isFamily)
 		if err, ok := err.(*RespErr); ok && err.Code != "InputStreamReadError" {
 			return nil, err
 		}

drivers/189pc/utils.go

@@ -500,7 +500,8 @@ func (y *Cloud189PC) StreamUpload(ctx context.Context, dstDir model.Obj, file mo
 	if err != nil {
 		return nil, err
 	}
-	ss, err := stream.NewStreamSectionReader(file, int(sliceSize))
+
+	ss, err := stream.NewStreamSectionReader(file, int(sliceSize), &up)
 	if err != nil {
 		return nil, err
 	}
@@ -581,7 +582,7 @@ func (y *Cloud189PC) StreamUpload(ctx context.Context, dstDir model.Obj, file mo
 				return nil
 			},
 			After: func(err error) {
-				ss.RecycleSectionReader(reader)
+				ss.FreeSectionReader(reader)
 			},
 		},
 		)
@@ -857,9 +858,7 @@ func (y *Cloud189PC) GetMultiUploadUrls(ctx context.Context, isFamily bool, uplo
 
 // 旧版本上传，家庭云不支持覆盖
 func (y *Cloud189PC) OldUpload(ctx context.Context, dstDir model.Obj, file model.FileStreamer, up driver.UpdateProgress, isFamily bool, overwrite bool) (model.Obj, error) {
-	cacheFileProgress := model.UpdateProgressWithRange(up, 0, 50)
-	up = model.UpdateProgressWithRange(up, 50, 100)
-	tempFile, fileMd5, err := stream.CacheFullInTempFileAndHash(file, cacheFileProgress, utils.MD5)
+	tempFile, fileMd5, err := stream.CacheFullAndHash(file, &up, utils.MD5)
 	if err != nil {
 		return nil, err
 	}

drivers/alias/driver.go

@@ -5,13 +5,15 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"net/url"
 	stdpath "path"
 	"strings"
 
 	"github.com/OpenListTeam/OpenList/v4/internal/driver"
 	"github.com/OpenListTeam/OpenList/v4/internal/errs"
 	"github.com/OpenListTeam/OpenList/v4/internal/fs"
 	"github.com/OpenListTeam/OpenList/v4/internal/model"
+	"github.com/OpenListTeam/OpenList/v4/internal/op"
 	"github.com/OpenListTeam/OpenList/v4/internal/sign"
 	"github.com/OpenListTeam/OpenList/v4/internal/stream"
 	"github.com/OpenListTeam/OpenList/v4/pkg/utils"
@@ -160,33 +162,26 @@ func (d *Alias) Link(ctx context.Context, file model.Obj, args model.LinkArgs) (
 					sign.Sign(reqPath)),
 			}, nil
 		}
+
+		resultLink := *link
+		resultLink.SyncClosers = utils.NewSyncClosers(link)
 		if args.Redirect {
-			return link, nil
+			return &resultLink, nil
 		}
 
-		resultLink := &model.Link{
-			URL:           link.URL,
-			Header:        link.Header,
-			RangeReader:   link.RangeReader,
-			MFile:         link.MFile,
-			Concurrency:   link.Concurrency,
-			PartSize:      link.PartSize,
-			ContentLength: link.ContentLength,
-			SyncClosers:   utils.NewSyncClosers(link),
-		}
 		if resultLink.ContentLength == 0 {
 			resultLink.ContentLength = fi.GetSize()
 		}
 		if resultLink.MFile != nil {
-			return resultLink, nil
+			return &resultLink, nil
 		}
 		if d.DownloadConcurrency > 0 {
 			resultLink.Concurrency = d.DownloadConcurrency
 		}
 		if d.DownloadPartSize > 0 {
 			resultLink.PartSize = d.DownloadPartSize * utils.KB
 		}
-		return resultLink, nil
+		return &resultLink, nil
 	}
 	return nil, errs.ObjectNotFound
 }
@@ -313,24 +308,29 @@ func (d *Alias) Put(ctx context.Context, dstDir model.Obj, s model.FileStreamer,
 	reqPath, err := d.getReqPath(ctx, dstDir, true)
 	if err == nil {
 		if len(reqPath) == 1 {
-			return fs.PutDirectly(ctx, *reqPath[0], &stream.FileStream{
-				Obj:          s,
-				Mimetype:     s.GetMimetype(),
-				WebPutAsTask: s.NeedStore(),
-				Reader:       s,
-			})
+			storage, reqActualPath, err := op.GetStorageAndActualPath(*reqPath[0])
+			if err != nil {
+				return err
+			}
+			return op.Put(ctx, storage, reqActualPath, &stream.FileStream{
+				Obj:      s,
+				Mimetype: s.GetMimetype(),
+				Reader:   s,
+			}, up)
 		} else {
-			file, err := s.CacheFullInTempFile()
+			file, err := s.CacheFullAndWriter(nil, nil)
 			if err != nil {
 				return err
 			}
-			for _, path := range reqPath {
+			count := float64(len(reqPath) + 1)
+			up(100 / count)
+			for i, path := range reqPath {
 				err = errors.Join(err, fs.PutDirectly(ctx, *path, &stream.FileStream{
-					Obj:          s,
-					Mimetype:     s.GetMimetype(),
-					WebPutAsTask: s.NeedStore(),
-					Reader:       file,
+					Obj:      s,
+					Mimetype: s.GetMimetype(),
+					Reader:   file,
 				}))
+				up(float64(i+2) / float64(count) * 100)
 				_, e := file.Seek(0, io.SeekStart)
 				if e != nil {
 					return errors.Join(err, e)
@@ -402,10 +402,24 @@ func (d *Alias) Extract(ctx context.Context, obj model.Obj, args model.ArchiveIn
 		return nil, errs.ObjectNotFound
 	}
 	for _, dst := range dsts {
-		link, err := d.extract(ctx, dst, sub, args)
-		if err == nil {
-			return link, nil
+		reqPath := stdpath.Join(dst, sub)
+		link, err := d.extract(ctx, reqPath, args)
+		if err != nil {
+			continue
+		}
+		if link == nil {
+			return &model.Link{
+				URL: fmt.Sprintf("%s/ap%s?inner=%s&pass=%s&sign=%s",
+					common.GetApiUrl(ctx),
+					utils.EncodePath(reqPath, true),
+					utils.EncodePath(args.InnerPath, true),
+					url.QueryEscape(args.Password),
+					sign.SignArchive(reqPath)),
+			}, nil
 		}
+		resultLink := *link
+		resultLink.SyncClosers = utils.NewSyncClosers(link)
+		return &resultLink, nil
 	}
 	return nil, errs.NotImplement
 }