DownloadFile support range

oss/download.go

@@ -31,13 +31,27 @@ func (bucket Bucket) DownloadFile(objectKey, filePath string, partSize int64, op
 		return err
 	}
 
+	uRange, err := getRangeConfig(options)
+	if err != nil {
+		return err
+	}
+
 	routines := getRoutines(options)
 
 	if cpConf.IsEnable {
-		return bucket.downloadFileWithCp(objectKey, filePath, partSize, options, cpConf.FilePath, routines)
+		return bucket.downloadFileWithCp(objectKey, filePath, partSize, options, cpConf.FilePath, routines, uRange)
 	}
 
-	return bucket.downloadFile(objectKey, filePath, partSize, options, routines)
+	return bucket.downloadFile(objectKey, filePath, partSize, options, routines, uRange)
+}
+
+// 获取下载范围
+func getRangeConfig(options []Option) (*unpackedRange, error) {
+	rangeOpt, err := findOption(options, HTTPHeaderRange, nil)
+	if err != nil || rangeOpt == nil {
+		return nil, err
+	}
+	return parseRange(rangeOpt.(string))
 }
 
 // ----- 并发无断点的下载  -----
@@ -104,7 +118,7 @@ func downloadWorker(id int, arg downloadWorkerArg, jobs <-chan downloadPart, res
 		}
 		defer fd.Close()
 
-		_, err = fd.Seek(part.Start, os.SEEK_SET)
+		_, err = fd.Seek(part.Start-part.Offset, os.SEEK_SET)
 		if err != nil {
 			failed <- err
 			break
@@ -130,13 +144,14 @@ func downloadScheduler(jobs chan downloadPart, parts []downloadPart) {
 
 // 下载片
 type downloadPart struct {
-	Index int   // 片序号，从0开始编号
-	Start int64 // 片起始位置
-	End   int64 // 片结束位置
+	Index  int   // 片序号，从0开始编号
+	Start  int64 // 片起始位置
+	End    int64 // 片结束位置
+	Offset int64 // 偏移位置
 }
 
 // 文件分片
-func getDownloadParts(bucket *Bucket, objectKey string, partSize int64) ([]downloadPart, error) {
+func getDownloadParts(bucket *Bucket, objectKey string, partSize int64, uRange *unpackedRange) ([]downloadPart, error) {
 	meta, err := bucket.GetObjectDetailedMeta(objectKey)
 	if err != nil {
 		return nil, err
@@ -150,10 +165,12 @@ func getDownloadParts(bucket *Bucket, objectKey string, partSize int64) ([]downl
 
 	part := downloadPart{}
 	i := 0
-	for offset := int64(0); offset < objectSize; offset += partSize {
+	start, end := adjustRange(uRange, objectSize)
+	for offset := start; offset < end; offset += partSize {
 		part.Index = i
 		part.Start = offset
-		part.End = GetPartEnd(offset, objectSize, partSize)
+		part.End = GetPartEnd(offset, end, partSize)
+		part.Offset = start
 		parts = append(parts, part)
 		i++
 	}
@@ -170,7 +187,7 @@ func getObjectBytes(parts []downloadPart) int64 {
 }
 
 // 并发无断点续传的下载
-func (bucket Bucket) downloadFile(objectKey, filePath string, partSize int64, options []Option, routines int) error {
+func (bucket Bucket) downloadFile(objectKey, filePath string, partSize int64, options []Option, routines int, uRange *unpackedRange) error {
 	tempFilePath := filePath + TempFileSuffix
 	listener := getProgressListener(options)
 
@@ -182,7 +199,7 @@ func (bucket Bucket) downloadFile(objectKey, filePath string, partSize int64, op
 	fd.Close()
 
 	// 分割文件
-	parts, err := getDownloadParts(&bucket, objectKey, partSize)
+	parts, err := getDownloadParts(&bucket, objectKey, partSize, uRange)
 	if err != nil {
 		return err
 	}
@@ -247,6 +264,8 @@ type downloadCheckpoint struct {
 	ObjStat  objectStat     // 文件状态
 	Parts    []downloadPart // 全部分片
 	PartStat []bool         // 分片下载是否完成
+	Start    int64          // 起点
+	End      int64          // 终点
 }
 
 type objectStat struct {
@@ -256,7 +275,7 @@ type objectStat struct {
 }
 
 // CP数据是否有效，CP有效且Object没有更新时有效
-func (cp downloadCheckpoint) isValid(bucket *Bucket, objectKey string) (bool, error) {
+func (cp downloadCheckpoint) isValid(bucket *Bucket, objectKey string, uRange *unpackedRange) (bool, error) {
 	// 比较CP的Magic及MD5
 	cpb := cp
 	cpb.MD5 = ""
@@ -286,6 +305,14 @@ func (cp downloadCheckpoint) isValid(bucket *Bucket, objectKey string) (bool, er
 		return false, nil
 	}
 
+	// 确认下载范围是否变化
+	if uRange != nil {
+		start, end := adjustRange(uRange, objectSize)
+		if start != cp.Start || end != cp.End {
+			return false, nil
+		}
+	}
+
 	return true, nil
 }
 
@@ -347,7 +374,7 @@ func (cp downloadCheckpoint) getCompletedBytes() int64 {
 }
 
 // 初始化下载任务
-func (cp *downloadCheckpoint) prepare(bucket *Bucket, objectKey, filePath string, partSize int64) error {
+func (cp *downloadCheckpoint) prepare(bucket *Bucket, objectKey, filePath string, partSize int64, uRange *unpackedRange) error {
 	// cp
 	cp.Magic = downloadCpMagic
 	cp.FilePath = filePath
@@ -369,7 +396,7 @@ func (cp *downloadCheckpoint) prepare(bucket *Bucket, objectKey, filePath string
 	cp.ObjStat.Etag = meta.Get(HTTPHeaderEtag)
 
 	// parts
-	cp.Parts, err = getDownloadParts(bucket, objectKey, partSize)
+	cp.Parts, err = getDownloadParts(bucket, objectKey, partSize, uRange)
 	if err != nil {
 		return err
 	}
@@ -387,7 +414,7 @@ func (cp *downloadCheckpoint) complete(cpFilePath, downFilepath string) error {
 }
 
 // 并发带断点的下载
-func (bucket Bucket) downloadFileWithCp(objectKey, filePath string, partSize int64, options []Option, cpFilePath string, routines int) error {
+func (bucket Bucket) downloadFileWithCp(objectKey, filePath string, partSize int64, options []Option, cpFilePath string, routines int, uRange *unpackedRange) error {
 	tempFilePath := filePath + TempFileSuffix
 	listener := getProgressListener(options)
 
@@ -399,9 +426,9 @@ func (bucket Bucket) downloadFileWithCp(objectKey, filePath string, partSize int
 	}
 
 	// LOAD出错或数据无效重新初始化下载
-	valid, err := dcp.isValid(&bucket, objectKey)
+	valid, err := dcp.isValid(&bucket, objectKey, uRange)
 	if err != nil || !valid {
-		if err = dcp.prepare(&bucket, objectKey, filePath, partSize); err != nil {
+		if err = dcp.prepare(&bucket, objectKey, filePath, partSize, uRange); err != nil {
 			return err
 		}
 		os.Remove(cpFilePath)

oss/download_test.go

@@ -1,6 +1,7 @@
 package oss
 
 import (
+	"bytes"
 	"fmt"
 	"os"
 	"time"
@@ -353,3 +354,194 @@ func (s *OssDownloadSuite) TestDownloadNegative(c *C) {
 	err = s.bucket.DownloadFile(objectName, newFile, 1024*1024*1024*100, Routines(2), Checkpoint(true, ""))
 	c.Assert(err, NotNil)
 }
+
+// TestDownloadWithRange 带范围的并发下载、断点下载测试
+func (s *OssDownloadSuite) TestDownloadWithRange(c *C) {
+	objectName := objectNamePrefix + "tdwr"
+	fileName := "../sample/BingWallpaper-2015-11-07.jpg"
+	newFile := "down-new-file-tdwr.jpg"
+
+	// 上传文件
+	err := s.bucket.UploadFile(objectName, fileName, 100*1024, Routines(3))
+	c.Assert(err, IsNil)
+
+	fileSize, err := getFileSize(fileName)
+	c.Assert(err, IsNil)
+
+	// 范围下载，从1024到4096
+	os.Remove(newFile)
+	err = s.bucket.DownloadFile(objectName, newFile, 100*1024, Routines(3), Range(1024, 4096))
+	c.Assert(err, IsNil)
+
+	// check
+	eq, err := compareFilesWithRange(fileName, 1024, newFile, 0, 3072)
+	c.Assert(err, IsNil)
+	c.Assert(eq, Equals, true)
+
+	// 范围下载，从1024到4096
+	os.Remove(newFile)
+	err = s.bucket.DownloadFile(objectName, newFile, 1024, Routines(3), NormalizedRange("1024-4096"))
+	c.Assert(err, IsNil)
+
+	// check
+	eq, err = compareFilesWithRange(fileName, 1024, newFile, 0, 3072)
+	c.Assert(err, IsNil)
+	c.Assert(eq, Equals, true)
+
+	// 范围下载，从2048到结束
+	os.Remove(newFile)
+	err = s.bucket.DownloadFile(objectName, newFile, 1024*1024, Routines(3), NormalizedRange("2048-"))
+	c.Assert(err, IsNil)
+
+	// check
+	eq, err = compareFilesWithRange(fileName, 2048, newFile, 0, fileSize-2048)
+	c.Assert(err, IsNil)
+	c.Assert(eq, Equals, true)
+
+	// 范围下载，最后4096个字节
+	os.Remove(newFile)
+	err = s.bucket.DownloadFile(objectName, newFile, 1024, Routines(3), NormalizedRange("-4096"))
+	c.Assert(err, IsNil)
+
+	// check
+	eq, err = compareFilesWithRange(fileName, fileSize-4096, newFile, 0, 4096)
+	c.Assert(err, IsNil)
+	c.Assert(eq, Equals, true)
+
+	err = s.bucket.DeleteObject(objectName)
+	c.Assert(err, IsNil)
+}
+
+// TestDownloadWithCheckoutAndRange 带范围的并发下载、断点下载测试
+func (s *OssDownloadSuite) TestDownloadWithCheckoutAndRange(c *C) {
+	objectName := objectNamePrefix + "tdwcr"
+	fileName := "../sample/BingWallpaper-2015-11-07.jpg"
+	newFile := "down-new-file-tdwcr.jpg"
+
+	// 上传文件
+	err := s.bucket.UploadFile(objectName, fileName, 100*1024, Routines(3))
+	c.Assert(err, IsNil)
+
+	fileSize, err := getFileSize(fileName)
+	c.Assert(err, IsNil)
+
+	// 范围下载，从1024到4096
+	os.Remove(newFile)
+	err = s.bucket.DownloadFile(objectName, newFile, 100*1024, Routines(3), Checkpoint(true, ""), Range(1024, 4096))
+	c.Assert(err, IsNil)
+
+	// check
+	eq, err := compareFilesWithRange(fileName, 1024, newFile, 0, 3072)
+	c.Assert(err, IsNil)
+	c.Assert(eq, Equals, true)
+
+	// 范围下载，从1024到4096
+	os.Remove(newFile)
+	err = s.bucket.DownloadFile(objectName, newFile, 1024, Routines(3), Checkpoint(true, ""), NormalizedRange("1024-4096"))
+	c.Assert(err, IsNil)
+
+	// check
+	eq, err = compareFilesWithRange(fileName, 1024, newFile, 0, 3072)
+	c.Assert(err, IsNil)
+	c.Assert(eq, Equals, true)
+
+	// 范围下载，从2048到结束
+	os.Remove(newFile)
+	err = s.bucket.DownloadFile(objectName, newFile, 1024*1024, Routines(3), Checkpoint(true, ""), NormalizedRange("2048-"))
+	c.Assert(err, IsNil)
+
+	// check
+	eq, err = compareFilesWithRange(fileName, 2048, newFile, 0, fileSize-2048)
+	c.Assert(err, IsNil)
+	c.Assert(eq, Equals, true)
+
+	// 范围下载，最后4096个字节
+	os.Remove(newFile)
+	err = s.bucket.DownloadFile(objectName, newFile, 1024, Routines(3), Checkpoint(true, ""), NormalizedRange("-4096"))
+	c.Assert(err, IsNil)
+
+	// check
+	eq, err = compareFilesWithRange(fileName, fileSize-4096, newFile, 0, 4096)
+	c.Assert(err, IsNil)
+	c.Assert(eq, Equals, true)
+
+	err = s.bucket.DeleteObject(objectName)
+	c.Assert(err, IsNil)
+}
+
+func getFileSize(fileName string) (int64, error) {
+	file, err := os.Open(fileName)
+	if err != nil {
+		return 0, err
+	}
+	defer file.Close()
+
+	stat, err := file.Stat()
+	if err != nil {
+		return 0, err
+	}
+
+	return stat.Size(), nil
+}
+
+// compare the content between fileL and fileR with specified range
+func compareFilesWithRange(fileL string, offsetL int64, fileR string, offsetR int64, size int64) (bool, error) {
+	finL, err := os.Open(fileL)
+	if err != nil {
+		return false, err
+	}
+	defer finL.Close()
+	finL.Seek(offsetL, os.SEEK_SET)
+
+	finR, err := os.Open(fileR)
+	if err != nil {
+		return false, err
+	}
+	defer finR.Close()
+	finR.Seek(offsetR, os.SEEK_SET)
+
+	statL, err := finL.Stat()
+	if err != nil {
+		return false, err
+	}
+
+	statR, err := finR.Stat()
+	if err != nil {
+		return false, err
+	}
+
+	if (offsetL+size > statL.Size()) || (offsetR+size > statR.Size()) {
+		return false, nil
+	}
+
+	part := statL.Size() - offsetL
+	if part > 16*1024 {
+		part = 16 * 1024
+	}
+
+	bufL := make([]byte, part)
+	bufR := make([]byte, part)
+	for readN := int64(0); readN < size; {
+		n, _ := finL.Read(bufL)
+		if 0 == n {
+			break
+		}
+
+		n, _ = finR.Read(bufR)
+		if 0 == n {
+			break
+		}
+
+		tailer := part
+		if tailer > size-readN {
+			tailer = size - readN
+		}
+		readN += tailer
+
+		if !bytes.Equal(bufL[0:tailer], bufR[0:tailer]) {
+			return false, nil
+		}
+	}
+
+	return true, nil
+}

oss/utils.go

@@ -9,6 +9,8 @@ import (
 	"os"
 	"os/exec"
 	"runtime"
+	"strconv"
+	"strings"
 	"time"
 )
 
@@ -44,6 +46,104 @@ func getSysInfo() sysInfo {
 	return sysInfo{name: name, release: release, machine: machine}
 }
 
+// unpackedRange
+type unpackedRange struct {
+	hasStart bool  // 是否指定了起点
+	hasEnd   bool  // 是否指定了终点
+	start    int64 // 起点
+	end      int64 // 终点
+}
+
+// invalid Range Error
+func invalidRangeError(r string) error {
+	return fmt.Errorf("InvalidRange %s", r)
+}
+
+// parseRange parse various styles of range such as bytes=M-N
+func parseRange(normalizedRange string) (*unpackedRange, error) {
+	var err error
+	hasStart := false
+	hasEnd := false
+	var start int64
+	var end int64
+
+	// bytes==M-N or ranges=M-N
+	nrSlice := strings.Split(normalizedRange, "=")
+	if len(nrSlice) != 2 || nrSlice[0] != "bytes" {
+		return nil, invalidRangeError(normalizedRange)
+	}
+
+	// bytes=M-N,X-Y
+	rSlice := strings.Split(nrSlice[1], ",")
+	rStr := rSlice[0]
+
+	if strings.HasSuffix(rStr, "-") { // M-
+		startStr := rStr[:len(rStr)-1]
+		start, err = strconv.ParseInt(startStr, 10, 64)
+		if err != nil {
+			return nil, invalidRangeError(normalizedRange)
+		}
+		hasStart = true
+	} else if strings.HasPrefix(rStr, "-") { // -N
+		len := rStr[1:]
+		end, err = strconv.ParseInt(len, 10, 64)
+		if err != nil {
+			return nil, invalidRangeError(normalizedRange)
+		}
+		if end == 0 { // -0
+			return nil, invalidRangeError(normalizedRange)
+		}
+		hasEnd = true
+	} else { // M-N
+		valSlice := strings.Split(rStr, "-")
+		if len(valSlice) != 2 {
+			return nil, invalidRangeError(normalizedRange)
+		}
+		start, err = strconv.ParseInt(valSlice[0], 10, 64)
+		if err != nil {
+			return nil, invalidRangeError(normalizedRange)
+		}
+		hasStart = true
+		end, err = strconv.ParseInt(valSlice[1], 10, 64)
+		if err != nil {
+			return nil, invalidRangeError(normalizedRange)
+		}
+		hasEnd = true
+	}
+
+	return &unpackedRange{hasStart, hasEnd, start, end}, nil
+}
+
+// adjustRange return adjusted range, adjust the range according to the length of the file
+func adjustRange(ur *unpackedRange, size int64) (start, end int64) {
+	if ur == nil {
+		return 0, size
+	}
+
+	if ur.hasStart && ur.hasEnd {
+		start = ur.start
+		end = ur.end
+		if ur.start < 0 || ur.end > size || ur.start > ur.end {
+			start = 0
+			end = size
+		}
+	} else if ur.hasStart {
+		start = ur.start
+		end = size
+		if ur.start < 0 {
+			start = 0
+		}
+	} else if ur.hasEnd {
+		start = size - ur.end
+		end = size
+		if ur.end < 0 || ur.end > size {
+			start = 0
+			end = size
+		}
+	}
+	return
+}
+
 // GetNowSec returns Unix time, the number of seconds elapsed since January 1, 1970 UTC.
 // 获取当前时间，从UTC开始的秒数。
 func GetNowSec() int64 {

oss/utils_test.go

@@ -1,8 +1,6 @@
 package oss
 
-import (
-	. "gopkg.in/check.v1"
-)
+import . "gopkg.in/check.v1"
 
 type OssUtilsSuite struct{}
 
@@ -103,3 +101,110 @@ func (s *OssUtilsSuite) TestGetPartEnd(c *C) {
 	end = GetPartEnd(7, 10, 3)
 	c.Assert(end, Equals, int64(9))
 }
+
+func (s *OssUtilsSuite) TestParseRange(c *C) {
+	// InvalidRange bytes==M-N
+	_, err := parseRange("bytes==M-N")
+	c.Assert(err, NotNil)
+	c.Assert(err.Error(), Equals, "InvalidRange bytes==M-N")
+
+	// InvalidRange ranges=M-N
+	_, err = parseRange("ranges=M-N")
+	c.Assert(err, NotNil)
+	c.Assert(err.Error(), Equals, "InvalidRange ranges=M-N")
+
+	// InvalidRange ranges=M-N
+	_, err = parseRange("bytes=M-N")
+	c.Assert(err, NotNil)
+	c.Assert(err.Error(), Equals, "InvalidRange bytes=M-N")
+
+	// InvalidRange ranges=M-
+	_, err = parseRange("bytes=M-")
+	c.Assert(err, NotNil)
+	c.Assert(err.Error(), Equals, "InvalidRange bytes=M-")
+
+	// InvalidRange ranges=-N
+	_, err = parseRange("bytes=-N")
+	c.Assert(err, NotNil)
+	c.Assert(err.Error(), Equals, "InvalidRange bytes=-N")
+
+	// InvalidRange ranges=-0
+	_, err = parseRange("bytes=-0")
+	c.Assert(err, NotNil)
+	c.Assert(err.Error(), Equals, "InvalidRange bytes=-0")
+
+	// ranges=M-N
+	ur, err := parseRange("bytes=1024-4096")
+	c.Assert(err, IsNil)
+	c.Assert(ur.start, Equals, (int64)(1024))
+	c.Assert(ur.end, Equals, (int64)(4096))
+	c.Assert(ur.hasStart, Equals, true)
+	c.Assert(ur.hasEnd, Equals, true)
+
+	// ranges=M-N,X-Y
+	ur, err = parseRange("bytes=1024-4096,2048-4096")
+	c.Assert(err, IsNil)
+	c.Assert(ur.start, Equals, (int64)(1024))
+	c.Assert(ur.end, Equals, (int64)(4096))
+	c.Assert(ur.hasStart, Equals, true)
+	c.Assert(ur.hasEnd, Equals, true)
+
+	// ranges=M-
+	ur, err = parseRange("bytes=1024-")
+	c.Assert(err, IsNil)
+	c.Assert(ur.start, Equals, (int64)(1024))
+	c.Assert(ur.end, Equals, (int64)(0))
+	c.Assert(ur.hasStart, Equals, true)
+	c.Assert(ur.hasEnd, Equals, false)
+
+	// ranges=-N
+	ur, err = parseRange("bytes=-4096")
+	c.Assert(err, IsNil)
+	c.Assert(ur.start, Equals, (int64)(0))
+	c.Assert(ur.end, Equals, (int64)(4096))
+	c.Assert(ur.hasStart, Equals, false)
+	c.Assert(ur.hasEnd, Equals, true)
+}
+
+func (s *OssUtilsSuite) TestAdjustRange(c *C) {
+	// nil
+	start, end := adjustRange(nil, 8192)
+	c.Assert(start, Equals, (int64)(0))
+	c.Assert(end, Equals, (int64)(8192))
+
+	// 1024-4096
+	ur := &unpackedRange{true, true, 1024, 4096}
+	start, end = adjustRange(ur, 8192)
+	c.Assert(start, Equals, (int64)(1024))
+	c.Assert(end, Equals, (int64)(4096))
+
+	// 1024-
+	ur = &unpackedRange{true, false, 1024, 4096}
+	start, end = adjustRange(ur, 8192)
+	c.Assert(start, Equals, (int64)(1024))
+	c.Assert(end, Equals, (int64)(8192))
+
+	// -4096
+	ur = &unpackedRange{false, true, 1024, 4096}
+	start, end = adjustRange(ur, 8192)
+	c.Assert(start, Equals, (int64)(4096))
+	c.Assert(end, Equals, (int64)(8192))
+
+	// Invalid range 4096-1024
+	ur = &unpackedRange{true, true, 4096, 1024}
+	start, end = adjustRange(ur, 8192)
+	c.Assert(start, Equals, (int64)(0))
+	c.Assert(end, Equals, (int64)(8192))
+
+	// Invalid range -1-
+	ur = &unpackedRange{true, false, -1, 0}
+	start, end = adjustRange(ur, 8192)
+	c.Assert(start, Equals, (int64)(0))
+	c.Assert(end, Equals, (int64)(8192))
+
+	// Invalid range -9999
+	ur = &unpackedRange{false, true, 0, 9999}
+	start, end = adjustRange(ur, 8192)
+	c.Assert(start, Equals, (int64)(0))
+	c.Assert(end, Equals, (int64)(8192))
+}