refactor: optimize stream, link, and resource management (#486)

* refactor: optimize stream, link, and resource management

* Link.MFile改为io.ReadSeeker类型

* fix (crypt): read on closed response body

* chore

* chore

* chore

Author: j2rong4cn

drivers/alias/driver.go

@@ -113,6 +113,7 @@ func (d *Alias) Link(ctx context.Context, file model.Obj, args model.LinkArgs) (
 	for _, dst := range dsts {
 		link, err := d.link(ctx, dst, sub, args)
 		if err == nil {
+			link.Expiration = nil // 去除非必要缓存，d.link里op.Lin有缓存
 			if !args.Redirect && len(link.URL) > 0 {
 				// 正常情况下 多并发 仅支持返回URL的驱动
 				// alias套娃alias 可以让crypt、mega等驱动(不返回URL的) 支持并发

drivers/crypt/driver.go

@@ -254,58 +254,56 @@ func (d *Crypt) Link(ctx context.Context, file model.Obj, args model.LinkArgs) (
 	if remoteLink.RangeReadCloser == nil && remoteLink.MFile == nil && len(remoteLink.URL) == 0 {
 		return nil, fmt.Errorf("the remote storage driver need to be enhanced to support encrytion")
 	}
+	resultRangeReadCloser := &model.RangeReadCloser{}
+	resultRangeReadCloser.TryAdd(remoteLink.MFile)
+	if remoteLink.RangeReadCloser != nil {
+		resultRangeReadCloser.AddClosers(remoteLink.RangeReadCloser.GetClosers())
+	}
 	remoteFileSize := remoteFile.GetSize()
-	remoteClosers := utils.EmptyClosers()
 	rangeReaderFunc := func(ctx context.Context, underlyingOffset, underlyingLength int64) (io.ReadCloser, error) {
 		length := underlyingLength
 		if underlyingLength >= 0 && underlyingOffset+underlyingLength >= remoteFileSize {
 			length = -1
 		}
-		rrc := remoteLink.RangeReadCloser
-		if len(remoteLink.URL) > 0 {
-			var converted, err = stream.GetRangeReadCloserFromLink(remoteFileSize, remoteLink)
+		if remoteLink.MFile != nil {
+			_, err := remoteLink.MFile.Seek(underlyingOffset, io.SeekStart)
 			if err != nil {
 				return nil, err
 			}
-			rrc = converted
+			//keep reuse same MFile and close at last.
+			return io.NopCloser(remoteLink.MFile), nil
 		}
-		if rrc != nil {
-			remoteReader, err := rrc.RangeRead(ctx, http_range.Range{Start: underlyingOffset, Length: length})
-			remoteClosers.AddClosers(rrc.GetClosers())
+		rrc := remoteLink.RangeReadCloser
+		if rrc == nil && len(remoteLink.URL) > 0 {
+			var err error
+			rrc, err = stream.GetRangeReadCloserFromLink(remoteFileSize, remoteLink)
 			if err != nil {
 				return nil, err
 			}
-			return remoteReader, nil
+			resultRangeReadCloser.AddClosers(rrc.GetClosers())
+			remoteLink.RangeReadCloser = rrc
 		}
-		if remoteLink.MFile != nil {
-			_, err := remoteLink.MFile.Seek(underlyingOffset, io.SeekStart)
+		if rrc != nil {
+			remoteReader, err := rrc.RangeRead(ctx, http_range.Range{Start: underlyingOffset, Length: length})
 			if err != nil {
 				return nil, err
 			}
-			//keep reuse same MFile and close at last.
-			remoteClosers.Add(remoteLink.MFile)
-			return io.NopCloser(remoteLink.MFile), nil
+			return remoteReader, nil
 		}
-
 		return nil, errs.NotSupport
 
 	}
-	resultRangeReader := func(ctx context.Context, httpRange http_range.Range) (io.ReadCloser, error) {
+	resultRangeReadCloser.RangeReader = func(ctx context.Context, httpRange http_range.Range) (io.ReadCloser, error) {
 		readSeeker, err := d.cipher.DecryptDataSeek(ctx, rangeReaderFunc, httpRange.Start, httpRange.Length)
 		if err != nil {
 			return nil, err
 		}
 		return readSeeker, nil
 	}
 
-	resultRangeReadCloser := &model.RangeReadCloser{RangeReader: resultRangeReader, Closers: remoteClosers}
-	resultLink := &model.Link{
+	return &model.Link{
 		RangeReadCloser: resultRangeReadCloser,
-		Expiration:      remoteLink.Expiration,
-	}
-
-	return resultLink, nil
-
+	}, nil
 }
 
 func (d *Crypt) MakeDir(ctx context.Context, parentDir model.Obj, dirName string) error {

drivers/doubao/util.go

@@ -524,7 +524,6 @@ func (d *Doubao) UploadByMultipart(ctx context.Context, config *UploadConfig, fi
 	if err != nil {
 		return nil, fmt.Errorf("failed to cache file: %w", err)
 	}
-	defer tempFile.Close()
 	up(10.0) // 更新进度
 	// 设置并行上传
 	threadG, uploadCtx := errgroup.NewGroupWithContext(ctx, d.uploadThread,

drivers/ftp/driver.go

@@ -7,6 +7,7 @@ import (
 	"github.com/OpenListTeam/OpenList/v4/internal/driver"
 	"github.com/OpenListTeam/OpenList/v4/internal/errs"
 	"github.com/OpenListTeam/OpenList/v4/internal/model"
+	"github.com/OpenListTeam/OpenList/v4/internal/stream"
 	"github.com/jlaffaye/ftp"
 )
 
@@ -66,7 +67,11 @@ func (d *FTP) Link(ctx context.Context, file model.Obj, args model.LinkArgs) (*m
 
 	r := NewFileReader(d.conn, encode(file.GetPath(), d.Encoding), file.GetSize())
 	link := &model.Link{
-		MFile: r,
+		MFile: &stream.RateLimitFile{
+			File:    r,
+			Limiter: stream.ServerDownloadLimit,
+			Ctx:     ctx,
+		},
 	}
 	return link, nil
 }

drivers/halalcloud/driver.go

@@ -256,9 +256,6 @@ func (d *HalalCloud) getLink(ctx context.Context, file model.Obj, args model.Lin
 		if httpRange.Length >= 0 && httpRange.Start+httpRange.Length >= size {
 			length = -1
 		}
-		if err != nil {
-			return nil, fmt.Errorf("open download file failed: %w", err)
-		}
 		oo := &openObject{
 			ctx:     ctx,
 			d:       fileAddrs,

drivers/halalcloud/types.go

@@ -96,7 +96,3 @@ type SteamFile struct {
 func (s *SteamFile) Read(p []byte) (n int, err error) {
 	return s.file.Read(p)
 }
-
-func (s *SteamFile) Close() error {
-	return s.file.Close()
-}

drivers/local/driver.go

@@ -242,7 +242,7 @@ func (d *Local) Link(ctx context.Context, file model.Obj, args model.LinkArgs) (
 			}
 			link.MFile = open
 		} else {
-			link.MFile = model.NewNopMFile(bytes.NewReader(buf.Bytes()))
+			link.MFile = bytes.NewReader(buf.Bytes())
 			//link.Header.Set("Content-Length", strconv.Itoa(buf.Len()))
 		}
 	} else {

drivers/mediatrack/driver.go

@@ -184,9 +184,6 @@ func (d *MediaTrack) Put(ctx context.Context, dstDir model.Obj, file model.FileS
 	if err != nil {
 		return err
 	}
-	defer func() {
-		_ = tempFile.Close()
-	}()
 	uploader := s3manager.NewUploader(s)
 	if file.GetSize() > s3manager.MaxUploadParts*s3manager.DefaultUploadPartSize {
 		uploader.PartSize = file.GetSize() / (s3manager.MaxUploadParts - 1)

drivers/netease_music/types.go

@@ -2,7 +2,6 @@ package netease_music
 
 import (
 	"context"
-	"io"
 	"net/http"
 	"strconv"
 	"strings"
@@ -11,7 +10,6 @@ import (
 	"github.com/OpenListTeam/OpenList/v4/internal/driver"
 	"github.com/OpenListTeam/OpenList/v4/internal/model"
 	"github.com/OpenListTeam/OpenList/v4/internal/sign"
-	"github.com/OpenListTeam/OpenList/v4/pkg/http_range"
 	"github.com/OpenListTeam/OpenList/v4/pkg/utils"
 	"github.com/OpenListTeam/OpenList/v4/pkg/utils/random"
 	"github.com/OpenListTeam/OpenList/v4/server/common"
@@ -55,17 +53,8 @@ func (lrc *LyricObj) getProxyLink(ctx context.Context) *model.Link {
 }
 
 func (lrc *LyricObj) getLyricLink() *model.Link {
-	reader := strings.NewReader(lrc.lyric)
 	return &model.Link{
-		RangeReadCloser: &model.RangeReadCloser{
-			RangeReader: func(ctx context.Context, httpRange http_range.Range) (io.ReadCloser, error) {
-				if httpRange.Length < 0 {
-					return io.NopCloser(reader), nil
-				}
-				sr := io.NewSectionReader(reader, httpRange.Start, httpRange.Length)
-				return io.NopCloser(sr), nil
-			},
-		},
+		MFile: strings.NewReader(lrc.lyric),
 	}
 }
 

drivers/sftp/driver.go

@@ -8,6 +8,7 @@ import (
 	"github.com/OpenListTeam/OpenList/v4/internal/driver"
 	"github.com/OpenListTeam/OpenList/v4/internal/errs"
 	"github.com/OpenListTeam/OpenList/v4/internal/model"
+	"github.com/OpenListTeam/OpenList/v4/internal/stream"
 	"github.com/OpenListTeam/OpenList/v4/pkg/utils"
 	"github.com/pkg/sftp"
 	log "github.com/sirupsen/logrus"
@@ -62,10 +63,13 @@ func (d *SFTP) Link(ctx context.Context, file model.Obj, args model.LinkArgs) (*
 	if err != nil {
 		return nil, err
 	}
-	link := &model.Link{
-		MFile: remoteFile,
-	}
-	return link, nil
+	return &model.Link{
+		MFile: &stream.RateLimitFile{
+			File:    remoteFile,
+			Limiter: stream.ServerDownloadLimit,
+			Ctx:     ctx,
+		},
+	}, nil
 }
 
 func (d *SFTP) MakeDir(ctx context.Context, parentDir model.Obj, dirName string) error {