feat: Upload large file from aria2

Author: HFO4

debug

@@ -1,6 +1,7 @@
 package onedrive
 
 import (
+	"bufio"
 	"bytes"
 	"encoding/json"
 	"fmt"
@@ -81,24 +82,42 @@ func (client *Client) CreateUploadSession(path string) (string, string) {
 }
 
 //UploadChunk 上传文件片
-func (client *Client) UploadChunk(url string, from int, to int, total int, stream *os.File) (string, string) {
+func (client *Client) UploadChunk(url string, from int, to int, total int, stream *os.File, reader *bufio.Reader) (string, string) {
 	headers := make(map[string]string)
 	headers["Content-Length"] = strconv.Itoa(to - from + 1)
 	headers["Content-Range"] = fmt.Sprintf("bytes %d-%d/%d", from, to, total)
-	res := client.apiChunkPut(url, stream, headers)
+	var res Response
+	if reader == nil {
+		res = client.apiChunkPut(url, stream, headers, nil)
+	} else {
+		buf := make([]byte, to-from+1)
+		_, err := reader.Read(buf)
+		if err != nil {
+			return "", err.Error()
+		}
+		res = client.apiChunkPut(url, stream, headers, &buf)
+	}
 	if res.Success {
 		return res.ResString, ""
 	}
 	return "", res.Error.Error.Message
 }
 
 //apiChunkPut 发送分片PUT请求
-func (client *Client) apiChunkPut(url string, stream *os.File, headers map[string]string) Response {
+func (client *Client) apiChunkPut(url string, stream *os.File, headers map[string]string, data *[]byte) Response {
 	if client.Tried > maxTry {
 		return buildResponseResult("PUT failed, reached the maximum number of attempts.", 0)
 	}
+	var (
+		req *http.Request
+		err error
+	)
+	if data == nil {
+		req, err = http.NewRequest("PUT", url, stream)
+	} else {
+		req, err = http.NewRequest("PUT", url, bytes.NewReader(*data))
+	}
 
-	req, err := http.NewRequest("PUT", url, stream)
 	if err != nil {
 		return buildResponseResult(err.Error(), 0)
 	}
@@ -116,7 +135,7 @@ func (client *Client) apiChunkPut(url string, stream *os.File, headers map[strin
 	res, err := client.HTTPClient.Do(req)
 	if err != nil {
 		client.Tried++
-		return client.apiChunkPut(url, stream, headers)
+		return client.apiChunkPut(url, stream, headers, data)
 	}
 	defer res.Body.Close()
 

onedrive/client.go

@@ -1,6 +1,7 @@
 package task
 
 import (
+	"bufio"
 	"encoding/json"
 	"fmt"
 	"log"
@@ -11,6 +12,7 @@ import (
 )
 
 const uploadThreadNum int = 4
+const aria2ChunkSize int = 10 * 1024 * 1024
 
 type chunkFile struct {
 	ID      int    `json:"id"`
@@ -23,14 +25,15 @@ type chunkFile struct {
 }
 
 type oneDriveUploadAttr struct {
-	Fname    string      `json:"fname"`
-	Path     string      `json:"path"`
-	Objname  string      `json:"objname"`
-	SavePath string      `json:"savePath"`
-	Fsize    uint64      `json:"fsize"`
-	PicInfo  string      `json:"picInfo"`
-	PolicyID int         `json:"policyId"`
-	Chunks   []chunkFile `json:"chunks"`
+	Fname      string      `json:"fname"`
+	Path       string      `json:"path"`
+	Objname    string      `json:"objname"`
+	SavePath   string      `json:"savePath"`
+	Fsize      uint64      `json:"fsize"`
+	PicInfo    string      `json:"picInfo"`
+	PolicyID   int         `json:"policyId"`
+	Chunks     []chunkFile `json:"chunks"`
+	OriginPath string      `json:"originPath"`
 }
 
 //OneDriveUpload OneDrive上传类型Task
@@ -78,9 +81,9 @@ func (task *OneDriveUpload) Excute() {
 	}
 	Client.Init()
 
-	if task.Type == "uploadSingleToOnedrive" {
+	if task.Type == "uploadSingleToOnedrive" || task.Type == "UploadRegularRemoteDownloadFileToOnedrive" {
 		task.uploadRegularFile(&Client)
-	} else if task.Type == "uploadChunksToOnedrive" {
+	} else if task.Type == "uploadChunksToOnedrive" || task.Type == "UploadLargeRemoteDownloadFileToOnedrive" {
 		task.uploadChunks(&Client)
 	}
 
@@ -124,19 +127,28 @@ func (task *OneDriveUpload) uploadSingleChunk(chunk Chunk, Client *onedrive.Clie
 
 	var r *os.File
 	var err error
+	var bfRd *bufio.Reader
 	if chunk.Type == 0 {
 		r, err = os.Open(chunk.ChunkPath)
+		bfRd = nil
 	} else {
 		r, err = os.Open(chunk.ChunkPath)
+		r.Seek(int64(chunk.From), 0)
+		bfRd = bufio.NewReader(r)
 	}
 
 	if err != nil {
 		task.Log("[Error] Failed to open file," + err.Error())
 		task.Error()
 		return false
 	}
+	var uploadErr string
+	if bfRd == nil {
+		_, uploadErr = Client.UploadChunk(url, chunk.From, chunk.To, int(task.Attr.Fsize), r, nil)
+	} else {
+		_, uploadErr = Client.UploadChunk(url, chunk.From, chunk.To, int(task.Attr.Fsize), r, bfRd)
+	}
 
-	_, uploadErr := Client.UploadChunk(url, chunk.From, chunk.To, int(task.Attr.Fsize), r)
 	if uploadErr != "" {
 		task.Log("[Error] Failed to upload chunk," + uploadErr)
 		task.Error()
@@ -158,23 +170,62 @@ func (task *OneDriveUpload) buildChunks() ([]Chunk, string) {
 	var chunkList []Chunk
 	var offset int
 
-	for _, v := range task.Attr.Chunks {
+	if chunkType == 0 {
+
+		for _, v := range task.Attr.Chunks {
+
+			var (
+				chunkPath string
+				chunkSize int
+			)
+			chunkPath = task.BasePath + "public/uploads/chunks/" + v.ObjName + ".chunk"
+
+			fileInfo, err := os.Stat(chunkPath)
+			if os.IsNotExist(err) {
+				return chunkList, "Chunk file " + chunkPath + " not exist"
+			}
+			chunkSize = int(fileInfo.Size())
 
-		chunkPath := task.BasePath + "public/uploads/chunks/" + v.ObjName + ".chunk"
+			chunkList = append(chunkList, Chunk{
+				Type:      chunkType,
+				From:      offset,
+				To:        offset + chunkSize - 1,
+				ChunkPath: chunkPath,
+			})
+
+			offset += chunkSize
+		}
+
+	} else {
+		for {
+
+			var (
+				chunkPath string
+				chunkSize int
+			)
+
+			chunkPath = task.Attr.OriginPath
+			if uint64(offset+aria2ChunkSize) > task.Attr.Fsize {
+				chunkSize = int(task.Attr.Fsize) - offset
+			} else {
+				chunkSize = aria2ChunkSize
+			}
+
+			chunkList = append(chunkList, Chunk{
+				Type:      chunkType,
+				From:      offset,
+				To:        offset + chunkSize - 1,
+				ChunkPath: chunkPath,
+			})
+
+			offset += chunkSize
+
+			if offset >= int(task.Attr.Fsize) {
+				break
+			}
 
-		fileInfo, err := os.Stat(chunkPath)
-		if os.IsNotExist(err) {
-			return chunkList, "Chunk file " + chunkPath + " not exist"
 		}
-		chunkSize := int(fileInfo.Size())
 
-		chunkList = append(chunkList, Chunk{
-			Type:      chunkType,
-			From:      offset,
-			To:        offset + chunkSize - 1,
-			ChunkPath: chunkPath,
-		})
-		offset += chunkSize
 	}
 
 	return chunkList, ""
@@ -183,7 +234,12 @@ func (task *OneDriveUpload) buildChunks() ([]Chunk, string) {
 
 func (task *OneDriveUpload) uploadRegularFile(Client *onedrive.Client) {
 	var filePath string
-	filePath = task.BasePath + "public/uploads/" + task.Attr.SavePath + "/" + task.Attr.Objname
+	if task.Type == "UploadRegularRemoteDownloadFileToOnedrive" {
+		filePath = task.Attr.OriginPath
+	} else {
+		filePath = task.BasePath + "public/uploads/" + task.Attr.SavePath + "/" + task.Attr.Objname
+	}
+
 	r, err := os.Open(filePath)
 	defer r.Close()
 	if err != nil {

task/TaskAdapter.go

@@ -58,6 +58,22 @@ func (task *taskInfo) Run() {
 			BasePath: task.siteInfo["basePath"],
 		}
 		newTask.Init()
+	case task.sqlInfo.TaskType == "UploadRegularRemoteDownloadFileToOnedrive":
+		newTask = &OneDriveUpload{
+			Info:     task,
+			Tried:    0,
+			Type:     task.sqlInfo.TaskType,
+			BasePath: task.siteInfo["basePath"],
+		}
+		newTask.Init()
+	case task.sqlInfo.TaskType == "UploadLargeRemoteDownloadFileToOnedrive":
+		newTask = &OneDriveUpload{
+			Info:     task,
+			Tried:    0,
+			Type:     task.sqlInfo.TaskType,
+			BasePath: task.siteInfo["basePath"],
+		}
+		newTask.Init()
 	}
 	newTask.Excute()
 }