perf(link): optimize concurrent response (#641)

* fix(crypt): bug caused by link cache

* perf(crypt,mega,halalcloud,quark,uc): optimize concurrent response link

* chore: 删除无用代码

* ftp

* 修复bug；资源释放

* 添加SyncClosers

* local,sftp,smb

* 重构，优化，增强

* Update internal/stream/util.go

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

* chore

* chore

* 优化，修复bug

* .

---------

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

Author: j2rong4cn

drivers/115/meta.go

@@ -18,7 +18,6 @@ var config = driver.Config{
 	Name:        "115 Cloud",
 	DefaultRoot: "0",
 	// OnlyProxy:   true,
-	// OnlyLocal:         true,
 	// NoOverwriteUpload: true,
 }
 

drivers/115_open/meta.go

@@ -11,23 +11,14 @@ type Addition struct {
 	// define other
 	OrderBy        string  `json:"order_by" type:"select" options:"file_name,file_size,user_utime,file_type"`
 	OrderDirection string  `json:"order_direction" type:"select" options:"asc,desc"`
-	LimitRate      float64  `json:"limit_rate" type:"float" default:"1" help:"limit all api request rate ([limit]r/1s)"`
+	LimitRate      float64 `json:"limit_rate" type:"float" default:"1" help:"limit all api request rate ([limit]r/1s)"`
 	AccessToken    string  `json:"access_token" required:"true"`
 	RefreshToken   string  `json:"refresh_token" required:"true"`
 }
 
 var config = driver.Config{
-	Name:              "115 Open",
-	LocalSort:         false,
-	OnlyLocal:         false,
-	OnlyProxy:         false,
-	NoCache:           false,
-	NoUpload:          false,
-	NeedMs:            false,
-	DefaultRoot:       "0",
-	CheckStatus:       false,
-	Alert:             "",
-	NoOverwriteUpload: false,
+	Name:        "115 Open",
+	DefaultRoot: "0",
 }
 
 func init() {

drivers/115_share/meta.go

@@ -19,12 +19,7 @@ type Addition struct {
 var config = driver.Config{
 	Name:        "115 Share",
 	DefaultRoot: "0",
-	// OnlyProxy:   true,
-	// OnlyLocal:         true,
-	CheckStatus:       false,
-	Alert:             "",
-	NoOverwriteUpload: true,
-	NoUpload:          true,
+	NoUpload:    true,
 }
 
 func init() {

drivers/123_share/meta.go

@@ -15,17 +15,10 @@ type Addition struct {
 }
 
 var config = driver.Config{
-	Name:              "123PanShare",
-	LocalSort:         true,
-	OnlyLocal:         false,
-	OnlyProxy:         false,
-	NoCache:           false,
-	NoUpload:          true,
-	NeedMs:            false,
-	DefaultRoot:       "0",
-	CheckStatus:       false,
-	Alert:             "",
-	NoOverwriteUpload: false,
+	Name:        "123PanShare",
+	LocalSort:   true,
+	NoUpload:    true,
+	DefaultRoot: "0",
 }
 
 func init() {

drivers/alias/driver.go

@@ -3,6 +3,7 @@ package alias
 import (
 	"context"
 	"errors"
+	"fmt"
 	"io"
 	stdpath "path"
 	"strings"
@@ -11,8 +12,10 @@ import (
 	"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/sign"
 	"github.com/OpenListTeam/OpenList/v4/internal/stream"
 	"github.com/OpenListTeam/OpenList/v4/pkg/utils"
+	"github.com/OpenListTeam/OpenList/v4/server/common"
 )
 
 type Alias struct {
@@ -111,21 +114,43 @@ func (d *Alias) Link(ctx context.Context, file model.Obj, args model.LinkArgs) (
 		return nil, errs.ObjectNotFound
 	}
 	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的) 支持并发
-				if d.DownloadConcurrency > 0 {
-					link.Concurrency = d.DownloadConcurrency
-				}
-				if d.DownloadPartSize > 0 {
-					link.PartSize = d.DownloadPartSize * utils.KB
+		reqPath := stdpath.Join(dst, sub)
+		link, file, err := d.link(ctx, reqPath, args)
+		if err != nil {
+			continue
+		}
+		var resultLink *model.Link
+		if link != nil {
+			resultLink = &model.Link{
+				URL:         link.URL,
+				Header:      link.Header,
+				RangeReader: link.RangeReader,
+				SyncClosers: utils.NewSyncClosers(link),
+			}
+			if link.MFile != nil {
+				resultLink.RangeReader = &model.FileRangeReader{
+					RangeReaderIF: stream.GetRangeReaderFromMFile(file.GetSize(), link.MFile),
 				}
 			}
-			return link, nil
+
+		} else {
+			resultLink = &model.Link{
+				URL: fmt.Sprintf("%s/p%s?sign=%s",
+					common.GetApiUrl(ctx),
+					utils.EncodePath(reqPath, true),
+					sign.Sign(reqPath)),
+			}
+
+		}
+		if !args.Redirect {
+			if d.DownloadConcurrency > 0 {
+				resultLink.Concurrency = d.DownloadConcurrency
+			}
+			if d.DownloadPartSize > 0 {
+				resultLink.PartSize = d.DownloadPartSize * utils.KB
+			}
 		}
+		return resultLink, nil
 	}
 	return nil, errs.ObjectNotFound
 }
@@ -251,9 +276,13 @@ 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], s)
+			return fs.PutDirectly(ctx, *reqPath[0], &stream.FileStream{
+				Obj:          s,
+				Mimetype:     s.GetMimetype(),
+				WebPutAsTask: s.NeedStore(),
+				Reader:       s,
+			})
 		} else {
-			defer s.Close()
 			file, err := s.CacheFullInTempFile()
 			if err != nil {
 				return err
@@ -338,14 +367,6 @@ func (d *Alias) Extract(ctx context.Context, obj model.Obj, args model.ArchiveIn
 	for _, dst := range dsts {
 		link, err := d.extract(ctx, dst, sub, args)
 		if err == nil {
-			if !args.Redirect && len(link.URL) > 0 {
-				if d.DownloadConcurrency > 0 {
-					link.Concurrency = d.DownloadConcurrency
-				}
-				if d.DownloadPartSize > 0 {
-					link.PartSize = d.DownloadPartSize * utils.KB
-				}
-			}
 			return link, nil
 		}
 	}

drivers/alias/util.go

@@ -96,37 +96,23 @@ func (d *Alias) list(ctx context.Context, dst, sub string, args *fs.ListArgs) ([
 	})
 }
 
-func (d *Alias) link(ctx context.Context, dst, sub string, args model.LinkArgs) (*model.Link, error) {
-	reqPath := stdpath.Join(dst, sub)
-	// 参考 crypt 驱动
+func (d *Alias) link(ctx context.Context, reqPath string, args model.LinkArgs) (*model.Link, model.Obj, error) {
 	storage, reqActualPath, err := op.GetStorageAndActualPath(reqPath)
 	if err != nil {
-		return nil, err
-	}
-	useRawLink := len(common.GetApiUrl(ctx)) == 0 // ftp、s3
-	if !useRawLink {
-		_, ok := storage.(*Alias)
-		useRawLink = !ok && !args.Redirect
+		return nil, nil, err
 	}
-	if useRawLink {
-		link, _, err := op.Link(ctx, storage, reqActualPath, args)
-		return link, err
+	// proxy || ftp,s3
+	if !args.Redirect || len(common.GetApiUrl(ctx)) == 0 {
+		return op.Link(ctx, storage, reqActualPath, args)
 	}
-	_, err = fs.Get(ctx, reqPath, &fs.GetArgs{NoLog: true})
+	obj, err := fs.Get(ctx, reqPath, &fs.GetArgs{NoLog: true})
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
-	if common.ShouldProxy(storage, stdpath.Base(sub)) {
-		link := &model.Link{
-			URL: fmt.Sprintf("%s/p%s?sign=%s",
-				common.GetApiUrl(ctx),
-				utils.EncodePath(reqPath, true),
-				sign.Sign(reqPath)),
-		}
-		return link, nil
+	if common.ShouldProxy(storage, stdpath.Base(reqPath)) {
+		return nil, obj, nil
 	}
-	link, _, err := op.Link(ctx, storage, reqActualPath, args)
-	return link, err
+	return op.Link(ctx, storage, reqActualPath, args)
 }
 
 func (d *Alias) getReqPath(ctx context.Context, obj model.Obj, isParent bool) ([]*string, error) {

drivers/aliyundrive/driver.go

@@ -165,7 +165,7 @@ func (d *AliDrive) Remove(ctx context.Context, obj model.Obj) error {
 }
 
 func (d *AliDrive) Put(ctx context.Context, dstDir model.Obj, streamer model.FileStreamer, up driver.UpdateProgress) error {
-	file := stream.FileStream{
+	file := &stream.FileStream{
 		Obj:      streamer,
 		Reader:   streamer,
 		Mimetype: streamer.GetMimetype(),
@@ -209,7 +209,7 @@ func (d *AliDrive) Put(ctx context.Context, dstDir model.Obj, streamer model.Fil
 				io.Closer
 			}{
 				Reader: io.MultiReader(buf, file),
-				Closer: &file,
+				Closer: file,
 			}
 		}
 	} else {

drivers/aliyundrive_open/meta.go

@@ -25,12 +25,6 @@ type Addition struct {
 
 var config = driver.Config{
 	Name:              "AliyundriveOpen",
-	LocalSort:         false,
-	OnlyLocal:         false,
-	OnlyProxy:         false,
-	NoCache:           false,
-	NoUpload:          false,
-	NeedMs:            false,
 	DefaultRoot:       "root",
 	NoOverwriteUpload: true,
 }

drivers/chaoxing/meta.go

@@ -32,7 +32,6 @@ func init() {
 			config: driver.Config{
 				Name:              "ChaoXingGroupDrive",
 				OnlyProxy:         true,
-				OnlyLocal:         false,
 				DefaultRoot:       "-1",
 				NoOverwriteUpload: true,
 			},

drivers/cloudreve_v4/meta.go

@@ -26,15 +26,8 @@ type Addition struct {
 
 var config = driver.Config{
 	Name:              "Cloudreve V4",
-	LocalSort:         false,
-	OnlyLocal:         false,
-	OnlyProxy:         false,
-	NoCache:           false,
-	NoUpload:          false,
-	NeedMs:            false,
 	DefaultRoot:       "cloudreve://my",
 	CheckStatus:       true,
-	Alert:             "",
 	NoOverwriteUpload: true,
 }