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

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

* 。

* add bytes_test.go

* fix(stream): handle EOF and buffer reading more gracefully

* 注释

* refactor: update CacheFullAndWriter to accept pointer for UpdateProgress

* update tests

* Update drivers/google_drive/util.go

Co-authored-by: Copilot <[email protected]>
Signed-off-by: j2rong4cn <[email protected]>

* 更优雅的克隆Link

* 修复stream已缓存但无法重复读取

* 将Bytes类型重命名为Reader

* 修复栈溢出

* update tests

---------

Signed-off-by: j2rong4cn <[email protected]>
Co-authored-by: Copilot <[email protected]>

Author: j2rong4cn

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
 		}