Skip to content

Commit

Permalink
feat(115): enhance cache (#7479)
Browse files Browse the repository at this point in the history
  • Loading branch information
@foxxorcat
foxxorcat authored Nov 8, 2024
1 parent 67c93ee commit 0a46979
Show file tree
Hide file tree
Showing 3 changed files with 146 additions and 40 deletions.
100 changes: 74 additions & 26 deletions drivers/115/driver.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -79,28 +79,60 @@ func (d *Pan115) Link(ctx context.Context, file model.Obj, args model.LinkArgs)
return link, nil
}

func (d *Pan115) MakeDir(ctx context.Context, parentDir model.Obj, dirName string) error {
func (d *Pan115) MakeDir(ctx context.Context, parentDir model.Obj, dirName string) (model.Obj, error) {
if err := d.WaitLimit(ctx); err != nil {
return err
return nil, err
}
if _, err := d.client.Mkdir(parentDir.GetID(), dirName); err != nil {
return err

result := driver115.MkdirResp{}
form := map[string]string{
"pid": parentDir.GetID(),
"cname": dirName,
}
return nil
req := d.client.NewRequest().
SetFormData(form).
SetResult(&result).
ForceContentType("application/json;charset=UTF-8")

resp, err := req.Post(driver115.ApiDirAdd)

err = driver115.CheckErr(err, &result, resp)
if err != nil {
return nil, err
}
f, err := d.getNewFile(result.FileID)
if err != nil {
return nil, nil
}
return f, nil
}

func (d *Pan115) Move(ctx context.Context, srcObj, dstDir model.Obj) error {
func (d *Pan115) Move(ctx context.Context, srcObj, dstDir model.Obj) (model.Obj, error) {
if err := d.WaitLimit(ctx); err != nil {
return err
return nil, err
}
return d.client.Move(dstDir.GetID(), srcObj.GetID())
if err := d.client.Move(dstDir.GetID(), srcObj.GetID()); err != nil {
return nil, err
}
f, err := d.getNewFile(srcObj.GetID())
if err != nil {
return nil, nil
}
return f, nil
}

func (d *Pan115) Rename(ctx context.Context, srcObj model.Obj, newName string) error {
func (d *Pan115) Rename(ctx context.Context, srcObj model.Obj, newName string) (model.Obj, error) {
if err := d.WaitLimit(ctx); err != nil {
return err
return nil, err
}
if err := d.client.Rename(srcObj.GetID(), newName); err != nil {
return nil, err
}
return d.client.Rename(srcObj.GetID(), newName)
f, err := d.getNewFile((srcObj.GetID()))
if err != nil {
return nil, nil
}
return f, nil
}

func (d *Pan115) Copy(ctx context.Context, srcObj, dstDir model.Obj) error {
Expand All @@ -117,9 +149,9 @@ func (d *Pan115) Remove(ctx context.Context, obj model.Obj) error {
return d.client.Delete(obj.GetID())
}

func (d *Pan115) Put(ctx context.Context, dstDir model.Obj, stream model.FileStreamer, up driver.UpdateProgress) error {
func (d *Pan115) Put(ctx context.Context, dstDir model.Obj, stream model.FileStreamer, up driver.UpdateProgress) (model.Obj, error) {
if err := d.WaitLimit(ctx); err != nil {
return err
return nil, err
}

var (
Expand All @@ -128,10 +160,10 @@ func (d *Pan115) Put(ctx context.Context, dstDir model.Obj, stream model.FileStr
)

if ok, err := d.client.UploadAvailable(); err != nil || !ok {
return err
return nil, err
}
if stream.GetSize() > d.client.UploadMetaInfo.SizeLimit {
return driver115.ErrUploadTooLarge
return nil, driver115.ErrUploadTooLarge
}
//if digest, err = d.client.GetDigestResult(stream); err != nil {
// return err
Expand All @@ -144,22 +176,22 @@ func (d *Pan115) Put(ctx context.Context, dstDir model.Obj, stream model.FileStr
}
reader, err := stream.RangeRead(http_range.Range{Start: 0, Length: hashSize})
if err != nil {
return err
return nil, err
}
preHash, err := utils.HashReader(utils.SHA1, reader)
if err != nil {
return err
return nil, err
}
preHash = strings.ToUpper(preHash)
fullHash := stream.GetHash().GetHash(utils.SHA1)
if len(fullHash) <= 0 {
tmpF, err := stream.CacheFullInTempFile()
if err != nil {
return err
return nil, err
}
fullHash, err = utils.HashFile(utils.SHA1, tmpF)
if err != nil {
return err
return nil, err
}
}
fullHash = strings.ToUpper(fullHash)
Expand All @@ -168,20 +200,36 @@ func (d *Pan115) Put(ctx context.Context, dstDir model.Obj, stream model.FileStr
// note that 115 add timeout for rapid-upload,
// and "sig invalid" err is thrown even when the hash is correct after timeout.
if fastInfo, err = d.rapidUpload(stream.GetSize(), stream.GetName(), dirID, preHash, fullHash, stream); err != nil {
return err
return nil, err
}
if matched, err := fastInfo.Ok(); err != nil {
return err
return nil, err
} else if matched {
return nil
f, err := d.getNewFileByPickCode(fastInfo.PickCode)
if err != nil {
return nil, nil
}
return f, nil
}

var uploadResult *UploadResult
// 闪传失败，上传
if stream.GetSize() <= utils.KB { // 文件大小小于1KB，改用普通模式上传
return d.client.UploadByOSS(&fastInfo.UploadOSSParams, stream, dirID)
if stream.GetSize() <= 10*utils.MB { // 文件大小小于10MB，改用普通模式上传
if uploadResult, err = d.UploadByOSS(&fastInfo.UploadOSSParams, stream, dirID); err != nil {
return nil, err
}
} else {
// 分片上传
if uploadResult, err = d.UploadByMultipart(&fastInfo.UploadOSSParams, stream.GetSize(), stream, dirID); err != nil {
return nil, err
}
}

file, err := d.getNewFile(uploadResult.Data.FileID)
if err != nil {
return nil, nil
}
// 分片上传
return d.UploadByMultipart(&fastInfo.UploadOSSParams, stream.GetSize(), stream, dirID)
return file, nil
}

func (d *Pan115) OfflineList(ctx context.Context) ([]*driver115.OfflineTask, error) {
Expand Down
2 changes: 1 addition & 1 deletion drivers/115/meta.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ type Addition struct {
QRCodeToken string `json:"qrcode_token" type:"text" help:"one of QR code token and cookie required"`
QRCodeSource string `json:"qrcode_source" type:"select" options:"web,android,ios,tv,alipaymini,wechatmini,qandroid" default:"linux" help:"select the QR code device, default linux"`
PageSize int64 `json:"page_size" type:"number" default:"1000" help:"list api per page size of 115 driver"`
LimitRate float64 `json:"limit_rate" type:"number" default:"2" help:"limit all api request rate (1r/[limit_rate]s)"`
LimitRate float64 `json:"limit_rate" type:"number" default:"2" help:"limit all api request rate ([limit]r/1s)"`
driver.RootID
}

Expand Down
84 changes: 71 additions & 13 deletions drivers/115/util.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -74,6 +74,34 @@ func (d *Pan115) getFiles(fileId string) ([]FileObj, error) {
return res, nil
}

func (d *Pan115) getNewFile(fileId string) (*FileObj, error) {
file, err := d.client.GetFile(fileId)
if err != nil {
return nil, err
}
return &FileObj{*file}, nil
}

func (d *Pan115) getNewFileByPickCode(pickCode string) (*FileObj, error) {
result := driver115.GetFileInfoResponse{}
req := d.client.NewRequest().
SetQueryParam("pick_code", pickCode).
ForceContentType("application/json;charset=UTF-8").
SetResult(&result)
resp, err := req.Get(driver115.ApiFileInfo)
if err := driver115.CheckErr(err, &result, resp); err != nil {
return nil, err
}
if len(result.Files) == 0 {
return nil, errors.New("not get file info")
}
fileInfo := result.Files[0]

f := &FileObj{}
f.From(fileInfo)
return f, nil
}

func (d *Pan115) getUA() string {
return fmt.Sprintf("Mozilla/5.0 115Browser/%s", appVer)
}
Expand Down Expand Up @@ -244,8 +272,38 @@ func UploadDigestRange(stream model.FileStreamer, rangeSpec string) (result stri
return
}

// UploadByOSS use aliyun sdk to upload
func (c *Pan115) UploadByOSS(params *driver115.UploadOSSParams, r io.Reader, dirID string) (*UploadResult, error) {
ossToken, err := c.client.GetOSSToken()
if err != nil {
return nil, err
}
ossClient, err := oss.New(driver115.OSSEndpoint, ossToken.AccessKeyID, ossToken.AccessKeySecret)
if err != nil {
return nil, err
}
bucket, err := ossClient.Bucket(params.Bucket)
if err != nil {
return nil, err
}

var bodyBytes []byte
if err = bucket.PutObject(params.Object, r, append(
driver115.OssOption(params, ossToken),
oss.CallbackResult(&bodyBytes),
)...); err != nil {
return nil, err
}

var uploadResult UploadResult
if err = json.Unmarshal(bodyBytes, &uploadResult); err != nil {
return nil, err
}
return &uploadResult, uploadResult.Err(string(bodyBytes))
}

// UploadByMultipart upload by mutipart blocks
func (d *Pan115) UploadByMultipart(params *driver115.UploadOSSParams, fileSize int64, stream model.FileStreamer, dirID string, opts ...driver115.UploadMultipartOption) error {
func (d *Pan115) UploadByMultipart(params *driver115.UploadOSSParams, fileSize int64, stream model.FileStreamer, dirID string, opts ...driver115.UploadMultipartOption) (*UploadResult, error) {
var (
chunks []oss.FileChunk
parts []oss.UploadPart
Expand All @@ -259,7 +317,7 @@ func (d *Pan115) UploadByMultipart(params *driver115.UploadOSSParams, fileSize i

tmpF, err := stream.CacheFullInTempFile()
if err != nil {
return err
return nil, err
}

options := driver115.DefalutUploadMultipartOptions()
Expand All @@ -272,15 +330,15 @@ func (d *Pan115) UploadByMultipart(params *driver115.UploadOSSParams, fileSize i
options.ThreadsNum = 1

if ossToken, err = d.client.GetOSSToken(); err != nil {
return err
return nil, err
}

if ossClient, err = oss.New(driver115.OSSEndpoint, ossToken.AccessKeyID, ossToken.AccessKeySecret, oss.EnableMD5(true), oss.EnableCRC(true)); err != nil {
return err
return nil, err
}

if bucket, err = ossClient.Bucket(params.Bucket); err != nil {
return err
return nil, err
}

// ossToken一小时后就会失效，所以每50分钟重新获取一次
Expand All @@ -290,15 +348,15 @@ func (d *Pan115) UploadByMultipart(params *driver115.UploadOSSParams, fileSize i
timeout := time.NewTimer(options.Timeout)

if chunks, err = SplitFile(fileSize); err != nil {
return err
return nil, err
}

if imur, err = bucket.InitiateMultipartUpload(params.Object,
oss.SetHeader(driver115.OssSecurityTokenHeaderName, ossToken.SecurityToken),
oss.UserAgentHeader(driver115.OSSUserAgent),
oss.EnableSha1(), oss.Sequential(),
); err != nil {
return err
return nil, err
}

wg := sync.WaitGroup{}
Expand Down Expand Up @@ -364,14 +422,14 @@ LOOP:
case <-ticker.C:
// 到时重新获取ossToken
if ossToken, err = d.client.GetOSSToken(); err != nil {
return err
return nil, err
}
case <-quit:
break LOOP
case <-errCh:
return err
return nil, err
case <-timeout.C:
return fmt.Errorf("time out")
return nil, fmt.Errorf("time out")
}
}

Expand All @@ -381,14 +439,14 @@ LOOP:
driver115.OssOption(params, ossToken),
oss.CallbackResult(&bodyBytes),
)...); err != nil {
return err
return nil, err
}

var uploadResult UploadResult
if err = json.Unmarshal(bodyBytes, &uploadResult); err != nil {
return err
return nil, err
}
return uploadResult.Err(string(bodyBytes))
return &uploadResult, uploadResult.Err(string(bodyBytes))
}

func chunksProducer(ch chan oss.FileChunk, chunks []oss.FileChunk) {
Expand Down

0 comments on commit 0a46979

Please sign in to comment.