perf(ftp): improve concurrent Link response; fix alias/local driver issues (#974)

Author: j2rong4cn

drivers/189pc/utils.go

@@ -550,9 +550,9 @@ func (y *Cloud189PC) StreamUpload(ctx context.Context, dstDir model.Obj, file mo
 						return err
 					}
 					silceMd5.Reset()
-					w, _ := utils.CopyWithBuffer(writers, reader)
+					w, err := utils.CopyWithBuffer(writers, reader)
 					if w != size {
-						return fmt.Errorf("can't read data, expected=%d, got=%d", size, w)
+						return fmt.Errorf("failed to read all data: (expect =%d, actual =%d) %w", size, w, err)
 					}
 					// 计算块md5并进行hex和base64编码
 					md5Bytes := silceMd5.Sum(nil)

drivers/alias/driver.go

@@ -78,10 +78,18 @@ func (d *Alias) Get(ctx context.Context, path string) (model.Obj, error) {
 		return nil, errs.ObjectNotFound
 	}
 	for _, dst := range dsts {
-		obj, err := d.get(ctx, path, dst, sub)
-		if err == nil {
-			return obj, nil
+		obj, err := fs.Get(ctx, stdpath.Join(dst, sub), &fs.GetArgs{NoLog: true})
+		if err != nil {
+			continue
 		}
+		return &model.Object{
+			Path:     path,
+			Name:     obj.GetName(),
+			Size:     obj.GetSize(),
+			Modified: obj.ModTime(),
+			IsFolder: obj.IsDir(),
+			HashInfo: obj.GetHash(),
+		}, nil
 	}
 	return nil, errs.ObjectNotFound
 }
@@ -99,7 +107,27 @@ func (d *Alias) List(ctx context.Context, dir model.Obj, args model.ListArgs) ([
 	var objs []model.Obj
 	fsArgs := &fs.ListArgs{NoLog: true, Refresh: args.Refresh}
 	for _, dst := range dsts {
-		tmp, err := d.list(ctx, dst, sub, fsArgs)
+		tmp, err := fs.List(ctx, stdpath.Join(dst, sub), fsArgs)
+		if err == nil {
+			tmp, err = utils.SliceConvert(tmp, func(obj model.Obj) (model.Obj, error) {
+				thumb, ok := model.GetThumb(obj)
+				objRes := model.Object{
+					Name:     obj.GetName(),
+					Size:     obj.GetSize(),
+					Modified: obj.ModTime(),
+					IsFolder: obj.IsDir(),
+				}
+				if !ok {
+					return &objRes, nil
+				}
+				return &model.ObjThumb{
+					Object: objRes,
+					Thumbnail: model.Thumbnail{
+						Thumbnail: thumb,
+					},
+				}, nil
+			})
+		}
 		if err == nil {
 			objs = append(objs, tmp...)
 		}
@@ -113,43 +141,50 @@ func (d *Alias) Link(ctx context.Context, file model.Obj, args model.LinkArgs) (
 	if !ok {
 		return nil, errs.ObjectNotFound
 	}
+	// proxy || ftp,s3
+	if common.GetApiUrl(ctx) == "" {
+		args.Redirect = false
+	}
 	for _, dst := range dsts {
 		reqPath := stdpath.Join(dst, sub)
-		link, file, err := d.link(ctx, reqPath, args)
+		link, fi, 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),
-				ContentLength: link.ContentLength,
-			}
-			if link.MFile != nil {
-				resultLink.RangeReader = &model.FileRangeReader{
-					RangeReaderIF: stream.GetRangeReaderFromMFile(file.GetSize(), link.MFile),
-				}
-			}
-
-		} else {
-			resultLink = &model.Link{
+		if link == nil {
+			// 重定向且需要通过代理
+			return &model.Link{
 				URL: fmt.Sprintf("%s/p%s?sign=%s",
 					common.GetApiUrl(ctx),
 					utils.EncodePath(reqPath, true),
 					sign.Sign(reqPath)),
-			}
+			}, nil
+		}
+		if args.Redirect {
+			return link, 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 !args.Redirect {
-			if d.DownloadConcurrency > 0 {
-				resultLink.Concurrency = d.DownloadConcurrency
-			}
-			if d.DownloadPartSize > 0 {
-				resultLink.PartSize = d.DownloadPartSize * utils.KB
-			}
+		if resultLink.ContentLength == 0 {
+			resultLink.ContentLength = fi.GetSize()
+		}
+		if resultLink.MFile != 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
 	}

drivers/alias/util.go

@@ -54,55 +54,12 @@ func (d *Alias) getRootAndPath(path string) (string, string) {
 	return parts[0], parts[1]
 }
 
-func (d *Alias) get(ctx context.Context, path string, dst, sub string) (model.Obj, error) {
-	obj, err := fs.Get(ctx, stdpath.Join(dst, sub), &fs.GetArgs{NoLog: true})
-	if err != nil {
-		return nil, err
-	}
-	return &model.Object{
-		Path:     path,
-		Name:     obj.GetName(),
-		Size:     obj.GetSize(),
-		Modified: obj.ModTime(),
-		IsFolder: obj.IsDir(),
-		HashInfo: obj.GetHash(),
-	}, nil
-}
-
-func (d *Alias) list(ctx context.Context, dst, sub string, args *fs.ListArgs) ([]model.Obj, error) {
-	objs, err := fs.List(ctx, stdpath.Join(dst, sub), args)
-	// the obj must implement the model.SetPath interface
-	// return objs, err
-	if err != nil {
-		return nil, err
-	}
-	return utils.SliceConvert(objs, func(obj model.Obj) (model.Obj, error) {
-		thumb, ok := model.GetThumb(obj)
-		objRes := model.Object{
-			Name:     obj.GetName(),
-			Size:     obj.GetSize(),
-			Modified: obj.ModTime(),
-			IsFolder: obj.IsDir(),
-		}
-		if !ok {
-			return &objRes, nil
-		}
-		return &model.ObjThumb{
-			Object: objRes,
-			Thumbnail: model.Thumbnail{
-				Thumbnail: thumb,
-			},
-		}, nil
-	})
-}
-
 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, nil, err
 	}
-	// proxy || ftp,s3
-	if !args.Redirect || len(common.GetApiUrl(ctx)) == 0 {
+	if !args.Redirect {
 		return op.Link(ctx, storage, reqActualPath, args)
 	}
 	obj, err := fs.Get(ctx, reqPath, &fs.GetArgs{NoLog: true})

drivers/aliyundrive_open/upload.go

@@ -137,11 +137,8 @@ func (d *AliyundriveOpen) calProofCode(stream model.FileStreamer) (string, error
 	}
 	buf := make([]byte, length)
 	n, err := io.ReadFull(reader, buf)
-	if err == io.ErrUnexpectedEOF {
-		return "", fmt.Errorf("can't read data, expected=%d, got=%d", len(buf), n)
-	}
-	if err != nil {
-		return "", err
+	if n != int(length) {
+		return "", fmt.Errorf("failed to read all data: (expect =%d, actual =%d) %w", length, n, err)
 	}
 	return base64.StdEncoding.EncodeToString(buf), nil
 }

drivers/crypt/driver.go

@@ -292,10 +292,10 @@ func (d *Crypt) Link(ctx context.Context, file model.Obj, args model.LinkArgs) (
 
 		if offset == 0 && limit > 0 {
 			fileHeader = make([]byte, fileHeaderSize)
-			n, _ := io.ReadFull(remoteReader, fileHeader)
+			n, err := io.ReadFull(remoteReader, fileHeader)
 			if n != fileHeaderSize {
 				fileHeader = nil
-				return nil, fmt.Errorf("can't read data, expected=%d, got=%d", fileHeaderSize, n)
+				return nil, fmt.Errorf("failed to read all data: (expect =%d, actual =%d) %w", fileHeaderSize, n, err)
 			}
 			if limit <= fileHeaderSize {
 				remoteReader.Close()

drivers/doubao/util.go

@@ -460,9 +460,9 @@ func (d *Doubao) Upload(ctx context.Context, config *UploadConfig, dstDir model.
 
 	// 计算CRC32
 	crc32Hash := crc32.NewIEEE()
-	w, _ := utils.CopyWithBuffer(crc32Hash, reader)
+	w, err := utils.CopyWithBuffer(crc32Hash, reader)
 	if w != file.GetSize() {
-		return nil, fmt.Errorf("can't read data, expected=%d, got=%d", file.GetSize(), w)
+		return nil, fmt.Errorf("failed to read all data: (expect =%d, actual =%d) %w", file.GetSize(), w, err)
 	}
 	crc32Value := hex.EncodeToString(crc32Hash.Sum(nil))
 
@@ -588,9 +588,9 @@ func (d *Doubao) UploadByMultipart(ctx context.Context, config *UploadConfig, fi
 						return err
 					}
 					hash.Reset()
-					w, _ := utils.CopyWithBuffer(hash, reader)
+					w, err := utils.CopyWithBuffer(hash, reader)
 					if w != size {
-						return fmt.Errorf("can't read data, expected=%d, got=%d", size, w)
+						return fmt.Errorf("failed to read all data: (expect =%d, actual =%d) %w", size, w, err)
 					}
 					crc32Value = hex.EncodeToString(hash.Sum(nil))
 					rateLimitedRd = driver.NewLimitedUploadStream(ctx, reader)

drivers/ftp/driver.go

@@ -2,12 +2,16 @@ package ftp
 
 import (
 	"context"
+	"io"
 	stdpath "path"
+	"sync"
+	"time"
 
 	"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/http_range"
 	"github.com/OpenListTeam/OpenList/v4/pkg/utils"
 	"github.com/jlaffaye/ftp"
 )
@@ -16,6 +20,9 @@ type FTP struct {
 	model.Storage
 	Addition
 	conn *ftp.ServerConn
+
+	ctx    context.Context
+	cancel context.CancelFunc
 }
 
 func (d *FTP) Config() driver.Config {
@@ -27,12 +34,16 @@ func (d *FTP) GetAddition() driver.Additional {
 }
 
 func (d *FTP) Init(ctx context.Context) error {
-	return d._login()
+	d.ctx, d.cancel = context.WithCancel(context.Background())
+	var err error
+	d.conn, err = d._login(ctx)
+	return err
 }
 
 func (d *FTP) Drop(ctx context.Context) error {
 	if d.conn != nil {
-		_ = d.conn.Logout()
+		_ = d.conn.Quit()
+		d.cancel()
 	}
 	return nil
 }
@@ -61,26 +72,53 @@ func (d *FTP) List(ctx context.Context, dir model.Obj, args model.ListArgs) ([]m
 	return res, nil
 }
 
-func (d *FTP) Link(ctx context.Context, file model.Obj, args model.LinkArgs) (*model.Link, error) {
-	if err := d.login(); err != nil {
+func (d *FTP) Link(_ context.Context, file model.Obj, args model.LinkArgs) (*model.Link, error) {
+	ctx, cancel := context.WithCancel(context.Background())
+	conn, err := d._login(ctx)
+	if err != nil {
+		cancel()
 		return nil, err
 	}
+	close := func() error {
+		_ = conn.Quit()
+		cancel()
+		return nil
+	}
 
-	remoteFile := NewFileReader(d.conn, encode(file.GetPath(), d.Encoding), file.GetSize())
-	if remoteFile != nil && !d.Config().OnlyLinkMFile {
-		return &model.Link{
-			RangeReader: &model.FileRangeReader{
-				RangeReaderIF: stream.RateLimitRangeReaderFunc(stream.GetRangeReaderFromMFile(file.GetSize(), remoteFile)),
-			},
-			SyncClosers: utils.NewSyncClosers(remoteFile),
+	path := encode(file.GetPath(), d.Encoding)
+	size := file.GetSize()
+	mu := &sync.Mutex{}
+	resultRangeReader := func(context context.Context, httpRange http_range.Range) (io.ReadCloser, error) {
+		length := httpRange.Length
+		if length < 0 || httpRange.Start+length > size {
+			length = size - httpRange.Start
+		}
+		mu.Lock()
+		defer mu.Unlock()
+		r, err := conn.RetrFrom(path, uint64(httpRange.Start))
+		if err != nil {
+			_ = conn.Quit()
+			conn, err = d._login(ctx)
+			if err == nil {
+				r, err = conn.RetrFrom(path, uint64(httpRange.Start))
+			}
+			if err != nil {
+				return nil, err
+			}
+		}
+		r.SetDeadline(time.Now().Add(time.Second))
+		return &FileReader{
+			Response: r,
+			Reader:   io.LimitReader(r, length),
+			ctx:      context,
 		}, nil
 	}
+
 	return &model.Link{
-		MFile: &stream.RateLimitFile{
-			File:    remoteFile,
-			Limiter: stream.ServerDownloadLimit,
-			Ctx:     ctx,
+		RangeReader: &model.FileRangeReader{
+			RangeReaderIF: stream.RateLimitRangeReaderFunc(resultRangeReader),
 		},
+		SyncClosers: utils.NewSyncClosers(utils.CloseFunc(close)),
 	}, nil
 }
 

drivers/ftp/meta.go

@@ -33,7 +33,7 @@ type Addition struct {
 var config = driver.Config{
 	Name:          "FTP",
 	LocalSort:     true,
-	OnlyLinkMFile: true,
+	OnlyLinkMFile: false,
 	DefaultRoot:   "/",
 	NoLinkURL:     true,
 }