Feat: new changes in pkg: chunk, backoff, local, onedrive

pkg/filesystem/driver/onedrive/api.go

@@ -221,16 +221,8 @@ func (client *Client) GetUploadSessionStatus(ctx context.Context, uploadURL stri
 	return &uploadSession, nil
 }
 
-var index = 0
-
 // UploadChunk 上传分片
 func (client *Client) UploadChunk(ctx context.Context, uploadURL string, content io.Reader, current *chunk.ChunkGroup) (*UploadSessionResponse, error) {
-	index++
-	if index == 1 || index == 2 {
-		request.BlackHole(content)
-		return nil, errors.New("error")
-	}
-
 	res, err := client.request(
 		ctx, "PUT", uploadURL, content,
 		request.WithContentLength(current.Length()),
@@ -331,16 +323,6 @@ func (client *Client) SimpleUpload(ctx context.Context, dst string, body io.Read
 		request.WithTimeout(time.Duration(150)*time.Second),
 	)
 	if err != nil {
-		retried := 0
-		if v, ok := ctx.Value(fsctx.RetryCtx).(int); ok {
-			retried = v
-		}
-		if retried < model.GetIntSetting("chunk_retries", 5) {
-			retried++
-			util.Log().Debug("文件[%s]上传失败[%s]，5秒钟后重试", dst, err)
-			time.Sleep(time.Duration(5) * time.Second)
-			return client.SimpleUpload(context.WithValue(ctx, fsctx.RetryCtx, retried), dst, body, size, opts...)
-		}
 		return nil, err
 	}
 

pkg/filesystem/driver/onedrive/api_test.go

@@ -4,6 +4,11 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"github.com/cloudreve/Cloudreve/v3/pkg/filesystem/chunk"
+	"github.com/cloudreve/Cloudreve/v3/pkg/filesystem/chunk/backoff"
+	"github.com/cloudreve/Cloudreve/v3/pkg/filesystem/fsctx"
+	"github.com/cloudreve/Cloudreve/v3/pkg/mq"
+	"io"
 	"io/ioutil"
 	"net/http"
 	"strings"
@@ -12,7 +17,6 @@ import (
 
 	model "github.com/cloudreve/Cloudreve/v3/models"
 	"github.com/cloudreve/Cloudreve/v3/pkg/cache"
-	"github.com/cloudreve/Cloudreve/v3/pkg/filesystem/fsctx"
 	"github.com/cloudreve/Cloudreve/v3/pkg/request"
 	"github.com/stretchr/testify/assert"
 	testMock "github.com/stretchr/testify/mock"
@@ -307,6 +311,31 @@ func TestClient_Meta(t *testing.T) {
 		asserts.NotNil(res)
 		asserts.Equal("123321", res.Name)
 	}
+
+	// 返回正常, 使用资源id
+	{
+		client.Credential.ExpiresIn = time.Now().Add(time.Duration(100) * time.Hour).Unix()
+		clientMock := ClientMock{}
+		clientMock.On(
+			"Request",
+			"GET",
+			testMock.Anything,
+			testMock.Anything,
+			testMock.Anything,
+		).Return(&request.Response{
+			Err: nil,
+			Response: &http.Response{
+				StatusCode: 200,
+				Body:       ioutil.NopCloser(strings.NewReader(`{"name":"123321"}`)),
+			},
+		})
+		client.Request = clientMock
+		res, err := client.Meta(context.Background(), "123321", "123")
+		clientMock.AssertExpectations(t)
+		asserts.NoError(err)
+		asserts.NotNil(res)
+		asserts.Equal("123321", res.Name)
+	}
 }
 
 func TestClient_CreateUploadSession(t *testing.T) {
@@ -442,9 +471,11 @@ func TestClient_UploadChunk(t *testing.T) {
 	client, _ := NewClient(&model.Policy{})
 	client.Credential.AccessToken = "AccessToken"
 	client.Credential.ExpiresIn = time.Now().Add(time.Duration(100) * time.Hour).Unix()
+	cg := chunk.NewChunkGroup(&fsctx.FileStream{Size: 15}, 10, &backoff.ConstantBackoff{}, false)
 
 	// 非最后分片，正常
 	{
+		cg.Next()
 		client.Credential.ExpiresIn = time.Now().Add(time.Duration(100) * time.Hour).Unix()
 		clientMock := ClientMock{}
 		clientMock.On(
@@ -453,6 +484,10 @@ func TestClient_UploadChunk(t *testing.T) {
 			"http://dev.com",
 			testMock.Anything,
 			testMock.Anything,
+			testMock.Anything,
+			testMock.Anything,
+			testMock.Anything,
+			testMock.Anything,
 		).Return(&request.Response{
 			Err: nil,
 			Response: &http.Response{
@@ -461,13 +496,7 @@ func TestClient_UploadChunk(t *testing.T) {
 			},
 		})
 		client.Request = clientMock
-		res, err := client.UploadChunk(context.Background(), "http://dev.com", &Chunk{
-			Offset:    0,
-			ChunkSize: 10,
-			Total:     100,
-			Retried:   0,
-			Data:      []byte("12313121231312"),
-		})
+		res, err := client.UploadChunk(context.Background(), "http://dev.com", strings.NewReader("1234567890"), cg)
 		clientMock.AssertExpectations(t)
 		asserts.NoError(err)
 		asserts.Equal("http://dev.com/2", res.UploadURL)
@@ -491,13 +520,7 @@ func TestClient_UploadChunk(t *testing.T) {
 			},
 		})
 		client.Request = clientMock
-		res, err := client.UploadChunk(context.Background(), "http://dev.com", &Chunk{
-			Offset:    0,
-			ChunkSize: 10,
-			Total:     100,
-			Retried:   0,
-			Data:      []byte("12313112313122"),
-		})
+		res, err := client.UploadChunk(context.Background(), "http://dev.com", strings.NewReader("1234567890"), cg)
 		clientMock.AssertExpectations(t)
 		asserts.Error(err)
 		asserts.Nil(res)
@@ -505,6 +528,7 @@ func TestClient_UploadChunk(t *testing.T) {
 
 	// 最后分片，正常
 	{
+		cg.Next()
 		client.Credential.ExpiresIn = time.Now().Add(time.Duration(100) * time.Hour).Unix()
 		clientMock := ClientMock{}
 		clientMock.On(
@@ -521,19 +545,13 @@ func TestClient_UploadChunk(t *testing.T) {
 			},
 		})
 		client.Request = clientMock
-		res, err := client.UploadChunk(context.Background(), "http://dev.com", &Chunk{
-			Offset:    95,
-			ChunkSize: 5,
-			Total:     100,
-			Retried:   0,
-			Data:      []byte("1231312"),
-		})
+		res, err := client.UploadChunk(context.Background(), "http://dev.com", strings.NewReader("12345"), cg)
 		clientMock.AssertExpectations(t)
 		asserts.NoError(err)
 		asserts.Nil(res)
 	}
 
-	// 最后分片，第一次失败，重试后成功
+	// 最后分片，失败
 	{
 		cache.Set("setting_chunk_retries", "1", 0)
 		client.Credential.ExpiresIn = 0
@@ -542,32 +560,11 @@ func TestClient_UploadChunk(t *testing.T) {
 			client.Credential.ExpiresIn = time.Now().Add(time.Duration(100) * time.Hour).Unix()
 		}()
 		clientMock := ClientMock{}
-		clientMock.On(
-			"Request",
-			"PUT",
-			"http://dev.com",
-			testMock.Anything,
-			testMock.Anything,
-		).Return(&request.Response{
-			Err: nil,
-			Response: &http.Response{
-				StatusCode: 200,
-				Body:       ioutil.NopCloser(strings.NewReader(`???`)),
-			},
-		})
 		client.Request = clientMock
-		chunk := &Chunk{
-			Offset:    95,
-			ChunkSize: 5,
-			Total:     100,
-			Retried:   0,
-			Data:      []byte("1231312"),
-		}
-		res, err := client.UploadChunk(context.Background(), "http://dev.com", chunk)
+		res, err := client.UploadChunk(context.Background(), "http://dev.com", strings.NewReader("12345"), cg)
 		clientMock.AssertExpectations(t)
-		asserts.NoError(err)
+		asserts.Error(err)
 		asserts.Nil(res)
-		asserts.EqualValues(1, chunk.Retried)
 	}
 }
 
@@ -576,39 +573,18 @@ func TestClient_Upload(t *testing.T) {
 	client, _ := NewClient(&model.Policy{})
 	client.Credential.AccessToken = "AccessToken"
 	client.Credential.ExpiresIn = time.Now().Add(time.Duration(100) * time.Hour).Unix()
-	ctx := context.WithValue(context.Background(), fsctx.DisableOverwrite, true)
+	ctx := context.Background()
+	cache.Set("setting_chunk_retries", "1", 0)
+	cache.Set("setting_use_temp_chunk_buffer", "false", 0)
 
 	// 小文件，简单上传，失败
 	{
 		client.Credential.ExpiresIn = 0
-		err := client.Upload(ctx, "123.jpg", 3, strings.NewReader("123"))
-		asserts.Error(err)
-	}
-
-	// 上下文取消
-	{
-		client.Credential.ExpiresIn = time.Now().Add(time.Duration(100) * time.Hour).Unix()
-		clientMock := ClientMock{}
-		clientMock.On(
-			"Request",
-			"POST",
-			testMock.Anything,
-			testMock.Anything,
-			testMock.Anything,
-		).Return(&request.Response{
-			Err: nil,
-			Response: &http.Response{
-				StatusCode: 200,
-				Body:       ioutil.NopCloser(strings.NewReader(`{"uploadUrl":"123321"}`)),
-			},
+		err := client.Upload(ctx, &fsctx.FileStream{
+			Size: 5,
+			File: io.NopCloser(strings.NewReader("12345")),
 		})
-		client.Request = clientMock
-		ctx, cancel := context.WithCancel(context.Background())
-		cancel()
-		err := client.Upload(ctx, "123.jpg", 15*1024*1024, strings.NewReader("123"))
-		clientMock.AssertExpectations(t)
 		asserts.Error(err)
-		asserts.Equal(ErrClientCanceled, err)
 	}
 
 	// 无法创建分片会话
@@ -629,11 +605,54 @@ func TestClient_Upload(t *testing.T) {
 			},
 		})
 		client.Request = clientMock
-		err := client.Upload(context.Background(), "123.jpg", 15*1024*1024, strings.NewReader("123"))
+		err := client.Upload(context.Background(), &fsctx.FileStream{
+			Size: SmallFileSize + 1,
+			File: io.NopCloser(strings.NewReader("12345")),
+		})
 		clientMock.AssertExpectations(t)
 		asserts.Error(err)
 	}
 
+	// 分片上传失败
+	{
+		client.Credential.ExpiresIn = time.Now().Add(time.Duration(100) * time.Hour).Unix()
+		clientMock := ClientMock{}
+		clientMock.On(
+			"Request",
+			"POST",
+			testMock.Anything,
+			testMock.Anything,
+			testMock.Anything,
+		).Return(&request.Response{
+			Err: nil,
+			Response: &http.Response{
+				StatusCode: 200,
+				Body:       ioutil.NopCloser(strings.NewReader(`{"uploadUrl":"123321"}`)),
+			},
+		})
+		clientMock.On(
+			"Request",
+			"PUT",
+			testMock.Anything,
+			testMock.Anything,
+			testMock.Anything,
+		).Return(&request.Response{
+			Err: nil,
+			Response: &http.Response{
+				StatusCode: 400,
+				Body:       ioutil.NopCloser(strings.NewReader(`{"uploadUrl":"123321"}`)),
+			},
+		})
+		client.Request = clientMock
+		err := client.Upload(context.Background(), &fsctx.FileStream{
+			Size: SmallFileSize + 1,
+			File: io.NopCloser(strings.NewReader("12345")),
+		})
+		clientMock.AssertExpectations(t)
+		asserts.Error(err)
+		asserts.Contains(err.Error(), "failed to upload chunk")
+	}
+
 }
 
 func TestClient_SimpleUpload(t *testing.T) {
@@ -643,7 +662,7 @@ func TestClient_SimpleUpload(t *testing.T) {
 	client.Credential.ExpiresIn = time.Now().Add(time.Duration(100) * time.Hour).Unix()
 	cache.Set("setting_chunk_retries", "1", 0)
 
-	// 请求失败，并重试
+	// 请求失败
 	{
 		client.Credential.ExpiresIn = 0
 		res, err := client.SimpleUpload(context.Background(), "123.jpg", strings.NewReader("123"), 3)
@@ -651,7 +670,6 @@ func TestClient_SimpleUpload(t *testing.T) {
 		asserts.Nil(res)
 	}
 
-	cache.Set("setting_chunk_retries", "0", 0)
 	// 返回未知响应
 	{
 		client.Credential.ExpiresIn = time.Now().Add(time.Duration(100) * time.Hour).Unix()
@@ -988,7 +1006,7 @@ func TestClient_MonitorUpload(t *testing.T) {
 		asserts.NotPanics(func() {
 			go func() {
 				time.Sleep(time.Duration(1) * time.Second)
-				FinishCallback("key")
+				mq.GlobalMQ.Publish("key", mq.Message{})
 			}()
 			client.MonitorUpload("url", "key", "path", 10, 10)
 		})

pkg/filesystem/driver/onedrive/handler_test.go

@@ -4,6 +4,9 @@ import (
 	"context"
 	"fmt"
 	"github.com/cloudreve/Cloudreve/v3/pkg/auth"
+	"github.com/cloudreve/Cloudreve/v3/pkg/mq"
+	"github.com/cloudreve/Cloudreve/v3/pkg/serializer"
+	"github.com/jinzhu/gorm"
 	"io"
 	"io/ioutil"
 	"net/http"
@@ -12,51 +15,23 @@ import (
 	"testing"
 	"time"
 
-	"github.com/DATA-DOG/go-sqlmock"
 	model "github.com/cloudreve/Cloudreve/v3/models"
 	"github.com/cloudreve/Cloudreve/v3/pkg/cache"
 	"github.com/cloudreve/Cloudreve/v3/pkg/filesystem/fsctx"
 	"github.com/cloudreve/Cloudreve/v3/pkg/request"
-	"github.com/cloudreve/Cloudreve/v3/pkg/serializer"
 	"github.com/stretchr/testify/assert"
 	testMock "github.com/stretchr/testify/mock"
 )
 
 func TestDriver_Token(t *testing.T) {
 	asserts := assert.New(t)
-	handler := Driver{
-		Policy: &model.Policy{
-			AccessKey:  "ak",
-			SecretKey:  "sk",
-			BucketName: "test",
-			Server:     "test.com",
-		},
-	}
-
-	// 无法获取文件路径
-	{
-		ctx := context.WithValue(context.Background(), fsctx.FileSizeCtx, uint64(10))
-		res, err := handler.Token(ctx, 10, "key", nil)
-		asserts.Error(err)
-		asserts.Equal(serializer.UploadCredential{}, res)
-	}
-
-	// 无法获取文件大小
-	{
-		ctx := context.WithValue(context.Background(), fsctx.SavePathCtx, "/123")
-		res, err := handler.Token(ctx, 10, "key", nil)
-		asserts.Error(err)
-		asserts.Equal(serializer.UploadCredential{}, res)
-	}
-
-	// 小文件成功
-	{
-		ctx := context.WithValue(context.Background(), fsctx.SavePathCtx, "/123")
-		ctx = context.WithValue(ctx, fsctx.FileSizeCtx, uint64(10))
-		res, err := handler.Token(ctx, 10, "key", nil)
-		asserts.NoError(err)
-		asserts.Equal(serializer.UploadCredential{}, res)
-	}
+	h, _ := NewDriver(&model.Policy{
+		AccessKey:  "ak",
+		SecretKey:  "sk",
+		BucketName: "test",
+		Server:     "test.com",
+	})
+	handler := h.(Driver)
 
 	// 分片上传 失败
 	{
@@ -78,11 +53,9 @@ func TestDriver_Token(t *testing.T) {
 			},
 		})
 		handler.Client.Request = clientMock
-		ctx := context.WithValue(context.Background(), fsctx.SavePathCtx, "/123")
-		ctx = context.WithValue(ctx, fsctx.FileSizeCtx, uint64(20*1024*1024))
-		res, err := handler.Token(ctx, 10, "key", nil)
+		res, err := handler.Token(context.Background(), 10, &serializer.UploadSession{}, &fsctx.FileStream{})
 		asserts.Error(err)
-		asserts.Equal(serializer.UploadCredential{}, res)
+		asserts.Nil(res)
 	}
 
 	// 分片上传 成功
@@ -108,15 +81,13 @@ func TestDriver_Token(t *testing.T) {
 			},
 		})
 		handler.Client.Request = clientMock
-		ctx := context.WithValue(context.Background(), fsctx.SavePathCtx, "/123")
-		ctx = context.WithValue(ctx, fsctx.FileSizeCtx, uint64(20*1024*1024))
 		go func() {
 			time.Sleep(time.Duration(1) * time.Second)
-			FinishCallback("key")
+			mq.GlobalMQ.Publish("TestDriver_Token", mq.Message{})
 		}()
-		res, err := handler.Token(ctx, 10, "key", nil)
+		res, err := handler.Token(context.Background(), 10, &serializer.UploadSession{Key: "TestDriver_Token"}, &fsctx.FileStream{})
 		asserts.NoError(err)
-		asserts.Equal("123321", res.Policy)
+		asserts.Equal("123321", res.UploadURLs[0])
 	}
 }
 
@@ -295,12 +266,8 @@ func TestDriver_Thumb(t *testing.T) {
 	// 失败
 	{
 		ctx := context.WithValue(context.Background(), fsctx.ThumbSizeCtx, [2]uint{10, 20})
-		ctx = context.WithValue(ctx, fsctx.FileModelCtx, model.File{})
-		mock.ExpectBegin()
-		mock.ExpectExec("UPDATE(.+)").WillReturnResult(sqlmock.NewResult(1, 1))
-		mock.ExpectCommit()
+		ctx = context.WithValue(ctx, fsctx.FileModelCtx, model.File{PicInfo: "1,1", Model: gorm.Model{ID: 1}})
 		res, err := handler.Thumb(ctx, "123.jpg")
-		asserts.NoError(mock.ExpectationsWereMet())
 		asserts.Error(err)
 		asserts.Empty(res.URL)
 	}
@@ -308,7 +275,6 @@ func TestDriver_Thumb(t *testing.T) {
 	// 上下文错误
 	{
 		_, err := handler.Thumb(context.Background(), "123.jpg")
-		asserts.NoError(mock.ExpectationsWereMet())
 		asserts.Error(err)
 	}
 }
@@ -329,7 +295,6 @@ func TestDriver_Delete(t *testing.T) {
 	// 失败
 	{
 		_, err := handler.Delete(context.Background(), []string{"1"})
-		asserts.NoError(mock.ExpectationsWereMet())
 		asserts.Error(err)
 	}
 
@@ -350,7 +315,7 @@ func TestDriver_Put(t *testing.T) {
 
 	// 失败
 	{
-		err := handler.Put(context.Background(), ioutil.NopCloser(strings.NewReader("")), "dst", 0)
+		err := handler.Put(context.Background(), &fsctx.FileStream{})
 		asserts.Error(err)
 	}
 }
@@ -418,3 +383,55 @@ func TestDriver_Get(t *testing.T) {
 	asserts.NoError(err)
 	asserts.Equal("123", string(content))
 }
+
+func TestDriver_replaceSourceHost(t *testing.T) {
+	tests := []struct {
+		name    string
+		origin  string
+		cdn     string
+		want    string
+		wantErr bool
+	}{
+		{"TestNoReplace", "http://1dr.ms/download.aspx?123456", "", "http://1dr.ms/download.aspx?123456", false},
+		{"TestReplaceCorrect", "http://1dr.ms/download.aspx?123456", "https://test.com:8080", "https://test.com:8080/download.aspx?123456", false},
+		{"TestCdnFormatError", "http://1dr.ms/download.aspx?123456", string([]byte{0x7f}), "", true},
+		{"TestSrcFormatError", string([]byte{0x7f}), "https://test.com:8080", "", true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			policy := &model.Policy{}
+			policy.OptionsSerialized.OdProxy = tt.cdn
+			handler := Driver{
+				Policy: policy,
+			}
+			got, err := handler.replaceSourceHost(tt.origin)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("replaceSourceHost() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if got != tt.want {
+				t.Errorf("replaceSourceHost() got = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestDriver_CancelToken(t *testing.T) {
+	asserts := assert.New(t)
+	handler := Driver{
+		Policy: &model.Policy{
+			AccessKey:  "ak",
+			SecretKey:  "sk",
+			BucketName: "test",
+			Server:     "test.com",
+		},
+	}
+	handler.Client, _ = NewClient(&model.Policy{})
+	handler.Client.Credential.ExpiresIn = time.Now().Add(time.Duration(100) * time.Hour).Unix()
+
+	// 失败
+	{
+		err := handler.CancelToken(context.Background(), &serializer.UploadSession{})
+		asserts.Error(err)
+	}
+}

pkg/filesystem/driver/onedrive/handller_test.go

@@ -1,38 +0,0 @@
-package onedrive
-
-import (
-	model "github.com/cloudreve/Cloudreve/v3/models"
-	"testing"
-)
-
-func TestDriver_replaceSourceHost(t *testing.T) {
-	tests := []struct {
-		name    string
-		origin  string
-		cdn     string
-		want    string
-		wantErr bool
-	}{
-		{"TestNoReplace", "http://1dr.ms/download.aspx?123456", "", "http://1dr.ms/download.aspx?123456", false},
-		{"TestReplaceCorrect", "http://1dr.ms/download.aspx?123456", "https://test.com:8080", "https://test.com:8080/download.aspx?123456", false},
-		{"TestCdnFormatError", "http://1dr.ms/download.aspx?123456", string([]byte{0x7f}), "", true},
-		{"TestSrcFormatError", string([]byte{0x7f}), "https://test.com:8080", "", true},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			policy := &model.Policy{}
-			policy.OptionsSerialized.OdProxy = tt.cdn
-			handler := Driver{
-				Policy: policy,
-			}
-			got, err := handler.replaceSourceHost(tt.origin)
-			if (err != nil) != tt.wantErr {
-				t.Errorf("replaceSourceHost() error = %v, wantErr %v", err, tt.wantErr)
-				return
-			}
-			if got != tt.want {
-				t.Errorf("replaceSourceHost() got = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}

pkg/filesystem/driver/onedrive/types.go

@@ -98,15 +98,6 @@ type ListResponse struct {
 	Context string     `json:"@odata.context"`
 }
 
-// Chunk 文件分片
-type Chunk struct {
-	Offset    int
-	ChunkSize int
-	Total     int
-	Retried   int
-	Data      []byte
-}
-
 // oauthEndpoint OAuth接口地址
 type oauthEndpoint struct {
 	token     url.URL
@@ -142,8 +133,3 @@ type Site struct {
 func init() {
 	gob.Register(Credential{})
 }
-
-// IsLast 返回是否为最后一个分片
-func (chunk *Chunk) IsLast() bool {
-	return chunk.Total-chunk.Offset == chunk.ChunkSize
-}