update multidownload
This commit is contained in:
@@ -100,7 +100,7 @@ func (s *CosTestSuite) SetupSuite() {
|
|||||||
XCosACL: "public-read",
|
XCosACL: "public-read",
|
||||||
}
|
}
|
||||||
r, err := s.Client.Bucket.Put(context.Background(), opt)
|
r, err := s.Client.Bucket.Put(context.Background(), opt)
|
||||||
if err != nil && r.StatusCode == 409 {
|
if err != nil && r != nil && r.StatusCode == 409 {
|
||||||
fmt.Println("BucketAlreadyOwnedByYou")
|
fmt.Println("BucketAlreadyOwnedByYou")
|
||||||
} else if err != nil {
|
} else if err != nil {
|
||||||
assert.Nil(s.T(), err, "PutBucket Failed")
|
assert.Nil(s.T(), err, "PutBucket Failed")
|
||||||
@@ -142,7 +142,7 @@ func (s *CosTestSuite) TestPutHeadDeleteBucket() {
|
|||||||
},
|
},
|
||||||
})
|
})
|
||||||
r, err := client.Bucket.Put(context.Background(), nil)
|
r, err := client.Bucket.Put(context.Background(), nil)
|
||||||
if err != nil && r.StatusCode == 409 {
|
if err != nil && r != nil && r.StatusCode == 409 {
|
||||||
fmt.Println("BucketAlreadyOwnedByYou")
|
fmt.Println("BucketAlreadyOwnedByYou")
|
||||||
} else if err != nil {
|
} else if err != nil {
|
||||||
assert.Nil(s.T(), err, "PutBucket Failed")
|
assert.Nil(s.T(), err, "PutBucket Failed")
|
||||||
@@ -507,6 +507,37 @@ func (s *CosTestSuite) TestPutGetDeleteObjectByUpload_10MB() {
|
|||||||
assert.Nil(s.T(), err, "remove local file Failed")
|
assert.Nil(s.T(), err, "remove local file Failed")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (s *CosTestSuite) TestPutGetDeleteObjectByUploadAndDownload_10MB() {
|
||||||
|
// Create tmp file
|
||||||
|
filePath := "tmpfile" + time.Now().Format(time.RFC3339)
|
||||||
|
newfile, err := os.Create(filePath)
|
||||||
|
assert.Nil(s.T(), err, "create tmp file Failed")
|
||||||
|
defer newfile.Close()
|
||||||
|
|
||||||
|
name := "test/objectUpload" + time.Now().Format(time.RFC3339)
|
||||||
|
b := make([]byte, 1024*1024*10)
|
||||||
|
_, err = rand.Read(b)
|
||||||
|
|
||||||
|
newfile.Write(b)
|
||||||
|
opt := &cos.MultiUploadOptions{
|
||||||
|
PartSize: 1,
|
||||||
|
ThreadPoolSize: 3,
|
||||||
|
}
|
||||||
|
_, _, err = s.Client.Object.Upload(context.Background(), name, filePath, opt)
|
||||||
|
assert.Nil(s.T(), err, "PutObject Failed")
|
||||||
|
|
||||||
|
// Over write tmp file
|
||||||
|
_, err = s.Client.Object.Download(context.Background(), name, filePath, nil)
|
||||||
|
assert.Nil(s.T(), err, "DownloadObject Failed")
|
||||||
|
|
||||||
|
_, err = s.Client.Object.Delete(context.Background(), name)
|
||||||
|
assert.Nil(s.T(), err, "DeleteObject Failed")
|
||||||
|
|
||||||
|
// remove the local tmp file
|
||||||
|
err = os.Remove(filePath)
|
||||||
|
assert.Nil(s.T(), err, "remove local file Failed")
|
||||||
|
}
|
||||||
|
|
||||||
func (s *CosTestSuite) TestPutGetDeleteObjectSpecialName() {
|
func (s *CosTestSuite) TestPutGetDeleteObjectSpecialName() {
|
||||||
f := strings.NewReader("test")
|
f := strings.NewReader("test")
|
||||||
name := s.SepFileName + time.Now().Format(time.RFC3339)
|
name := s.SepFileName + time.Now().Format(time.RFC3339)
|
||||||
@@ -606,7 +637,7 @@ func (s *CosTestSuite) TestCopyObject() {
|
|||||||
|
|
||||||
// Notice in intranet the bucket host sometimes has i/o timeout problem
|
// Notice in intranet the bucket host sometimes has i/o timeout problem
|
||||||
r, err := c.Bucket.Put(context.Background(), opt)
|
r, err := c.Bucket.Put(context.Background(), opt)
|
||||||
if err != nil && r.StatusCode == 409 {
|
if err != nil && r != nil && r.StatusCode == 409 {
|
||||||
fmt.Println("BucketAlreadyOwnedByYou")
|
fmt.Println("BucketAlreadyOwnedByYou")
|
||||||
} else if err != nil {
|
} else if err != nil {
|
||||||
assert.Nil(s.T(), err, "PutBucket Failed")
|
assert.Nil(s.T(), err, "PutBucket Failed")
|
||||||
@@ -971,7 +1002,7 @@ func (s *CosTestSuite) TestMultiCopy() {
|
|||||||
|
|
||||||
// Notice in intranet the bucket host sometimes has i/o timeout problem
|
// Notice in intranet the bucket host sometimes has i/o timeout problem
|
||||||
r, err := c.Bucket.Put(context.Background(), opt)
|
r, err := c.Bucket.Put(context.Background(), opt)
|
||||||
if err != nil && r.StatusCode == 409 {
|
if err != nil && r != nil && r.StatusCode == 409 {
|
||||||
fmt.Println("BucketAlreadyOwnedByYou")
|
fmt.Println("BucketAlreadyOwnedByYou")
|
||||||
} else if err != nil {
|
} else if err != nil {
|
||||||
assert.Nil(s.T(), err, "PutBucket Failed")
|
assert.Nil(s.T(), err, "PutBucket Failed")
|
||||||
|
|||||||
11
object.go
11
object.go
@@ -673,8 +673,9 @@ func downloadWorker(s *ObjectService, jobs <-chan *Jobs, results chan<- *Results
|
|||||||
fd.Seek(j.Chunk.OffSet, os.SEEK_SET)
|
fd.Seek(j.Chunk.OffSet, os.SEEK_SET)
|
||||||
n, err := io.Copy(fd, LimitReadCloser(resp.Body, j.Chunk.Size))
|
n, err := io.Copy(fd, LimitReadCloser(resp.Body, j.Chunk.Size))
|
||||||
if n != j.Chunk.Size || err != nil {
|
if n != j.Chunk.Size || err != nil {
|
||||||
res.err = fmt.Errorf("io.Copy Failed, read:%v, size:%v, err:%v", n, j.Chunk.Size, err)
|
res.err = fmt.Errorf("io.Copy Failed, nread:%v, want:%v, err:%v", n, j.Chunk.Size, err)
|
||||||
}
|
}
|
||||||
|
fd.Close()
|
||||||
results <- &res
|
results <- &res
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
@@ -1040,7 +1041,7 @@ func (s *ObjectService) Download(ctx context.Context, name string, filepath stri
|
|||||||
opt = &MultiDownloadOptions{}
|
opt = &MultiDownloadOptions{}
|
||||||
}
|
}
|
||||||
if opt.Opt != nil && opt.Opt.Range != "" {
|
if opt.Opt != nil && opt.Opt.Range != "" {
|
||||||
return nil, fmt.Errorf("does not supported Range Get")
|
return nil, fmt.Errorf("Download doesn't support Range Options")
|
||||||
}
|
}
|
||||||
// 获取文件长度和CRC
|
// 获取文件长度和CRC
|
||||||
var coscrc string
|
var coscrc string
|
||||||
@@ -1048,6 +1049,7 @@ func (s *ObjectService) Download(ctx context.Context, name string, filepath stri
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return resp, err
|
return resp, err
|
||||||
}
|
}
|
||||||
|
// 如果对象不存在x-cos-hash-crc64ecma,则跳过不做校验
|
||||||
coscrc = resp.Header.Get("x-cos-hash-crc64ecma")
|
coscrc = resp.Header.Get("x-cos-hash-crc64ecma")
|
||||||
strTotalBytes := resp.Header.Get("Content-Length")
|
strTotalBytes := resp.Header.Get("Content-Length")
|
||||||
totalBytes, err := strconv.ParseInt(strTotalBytes, 10, 64)
|
totalBytes, err := strconv.ParseInt(strTotalBytes, 10, 64)
|
||||||
@@ -1072,6 +1074,7 @@ func (s *ObjectService) Download(ctx context.Context, name string, filepath stri
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return rsp, err
|
return rsp, err
|
||||||
}
|
}
|
||||||
|
defer fd.Close()
|
||||||
localcrc, err := calCRC64(fd)
|
localcrc, err := calCRC64(fd)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return rsp, err
|
return rsp, err
|
||||||
@@ -1082,11 +1085,13 @@ func (s *ObjectService) Download(ctx context.Context, name string, filepath stri
|
|||||||
}
|
}
|
||||||
return rsp, err
|
return rsp, err
|
||||||
}
|
}
|
||||||
|
// 创建文件
|
||||||
nfile, err := os.OpenFile(filepath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0660)
|
nfile, err := os.OpenFile(filepath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0660)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return resp, err
|
return resp, err
|
||||||
}
|
}
|
||||||
nfile.Close()
|
nfile.Close()
|
||||||
|
|
||||||
var poolSize int
|
var poolSize int
|
||||||
if opt.ThreadPoolSize > 0 {
|
if opt.ThreadPoolSize > 0 {
|
||||||
poolSize = opt.ThreadPoolSize
|
poolSize = opt.ThreadPoolSize
|
||||||
@@ -1104,6 +1109,7 @@ func (s *ObjectService) Download(ctx context.Context, name string, filepath stri
|
|||||||
var downOpt ObjectGetOptions
|
var downOpt ObjectGetOptions
|
||||||
if opt.Opt != nil {
|
if opt.Opt != nil {
|
||||||
downOpt = *opt.Opt
|
downOpt = *opt.Opt
|
||||||
|
downOpt.Listener = nil // listener need to set nil
|
||||||
}
|
}
|
||||||
job := &Jobs{
|
job := &Jobs{
|
||||||
Name: name,
|
Name: name,
|
||||||
@@ -1135,6 +1141,7 @@ func (s *ObjectService) Download(ctx context.Context, name string, filepath stri
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return resp, err
|
return resp, err
|
||||||
}
|
}
|
||||||
|
defer fd.Close()
|
||||||
localcrc, err := calCRC64(fd)
|
localcrc, err := calCRC64(fd)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return resp, err
|
return resp, err
|
||||||
|
|||||||
@@ -10,6 +10,7 @@ import (
|
|||||||
"hash/crc64"
|
"hash/crc64"
|
||||||
"io"
|
"io"
|
||||||
"io/ioutil"
|
"io/ioutil"
|
||||||
|
math_rand "math/rand"
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/url"
|
"net/url"
|
||||||
"os"
|
"os"
|
||||||
@@ -528,19 +529,19 @@ func TestObjectService_Download(t *testing.T) {
|
|||||||
}
|
}
|
||||||
defer os.Remove(filePath)
|
defer os.Remove(filePath)
|
||||||
// 源文件内容
|
// 源文件内容
|
||||||
totalBytes := int64(1024 * 1024 * 10)
|
totalBytes := int64(1024*1024*9 + 123)
|
||||||
b := make([]byte, totalBytes)
|
b := make([]byte, totalBytes)
|
||||||
_, err = rand.Read(b)
|
_, err = rand.Read(b)
|
||||||
newfile.Write(b)
|
newfile.Write(b)
|
||||||
newfile.Close()
|
newfile.Close()
|
||||||
tb := crc64.MakeTable(crc64.ECMA)
|
tb := crc64.MakeTable(crc64.ECMA)
|
||||||
localcrc := crc64.Update(0, tb, b)
|
localcrc := strconv.FormatUint(crc64.Update(0, tb, b), 10)
|
||||||
|
|
||||||
retryMap := make(map[int64]int)
|
retryMap := make(map[int64]int)
|
||||||
mux.HandleFunc("/test.go.download", func(w http.ResponseWriter, r *http.Request) {
|
mux.HandleFunc("/test.go.download", func(w http.ResponseWriter, r *http.Request) {
|
||||||
if r.Method == http.MethodHead {
|
if r.Method == http.MethodHead {
|
||||||
w.Header().Add("Content-Length", strconv.FormatInt(totalBytes, 10))
|
w.Header().Add("Content-Length", strconv.FormatInt(totalBytes, 10))
|
||||||
w.Header().Add("x-cos-hash-crc64ecma", strconv.FormatUint(localcrc, 10))
|
w.Header().Add("x-cos-hash-crc64ecma", localcrc)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
strRange := r.Header.Get("Range")
|
strRange := r.Header.Get("Range")
|
||||||
@@ -549,36 +550,21 @@ func TestObjectService_Download(t *testing.T) {
|
|||||||
start, _ := strconv.ParseInt(slice2[0], 10, 64)
|
start, _ := strconv.ParseInt(slice2[0], 10, 64)
|
||||||
end, _ := strconv.ParseInt(slice2[1], 10, 64)
|
end, _ := strconv.ParseInt(slice2[1], 10, 64)
|
||||||
if retryMap[start] == 0 {
|
if retryMap[start] == 0 {
|
||||||
|
// 重试校验1
|
||||||
retryMap[start]++
|
retryMap[start]++
|
||||||
w.WriteHeader(http.StatusGatewayTimeout)
|
w.WriteHeader(http.StatusGatewayTimeout)
|
||||||
} else if retryMap[start] == 1 {
|
} else if retryMap[start] == 1 {
|
||||||
|
// 重试检验2
|
||||||
retryMap[start]++
|
retryMap[start]++
|
||||||
w.WriteHeader(http.StatusGatewayTimeout)
|
io.Copy(w, bytes.NewBuffer(b[start:end]))
|
||||||
return
|
} else if retryMap[start] == 2 {
|
||||||
fd, err := os.Open(filePath)
|
// 重试检验3
|
||||||
if err != nil {
|
retryMap[start]++
|
||||||
t.Fatalf("open file failed: %v", err)
|
st := math_rand.Int63n(totalBytes - 1024*1024)
|
||||||
}
|
et := st + end - start
|
||||||
defer fd.Close()
|
io.Copy(w, bytes.NewBuffer(b[st:et+1]))
|
||||||
w.Header().Add("x-cos-hash-crc64ecma", strconv.FormatUint(localcrc, 10))
|
|
||||||
fd.Seek(start, os.SEEK_SET)
|
|
||||||
n, err := io.Copy(w, LimitReadCloser(fd, (end-start)/2))
|
|
||||||
if err != nil || int64(n) != (end-start)/2 {
|
|
||||||
t.Fatalf("write file failed:%v, n:%v", err, n)
|
|
||||||
}
|
|
||||||
|
|
||||||
} else {
|
} else {
|
||||||
fd, err := os.Open(filePath)
|
io.Copy(w, bytes.NewBuffer(b[start:end+1]))
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("open file failed: %v", err)
|
|
||||||
}
|
|
||||||
defer fd.Close()
|
|
||||||
w.Header().Add("x-cos-hash-crc64ecma", strconv.FormatUint(localcrc, 10))
|
|
||||||
fd.Seek(start, os.SEEK_SET)
|
|
||||||
n, err := io.Copy(w, LimitReadCloser(fd, end-start+1))
|
|
||||||
if err != nil || int64(n) != end-start+1 {
|
|
||||||
t.Fatalf("write file failed:%v, n:%v", err, n)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
|
||||||
@@ -587,9 +573,20 @@ func TestObjectService_Download(t *testing.T) {
|
|||||||
PartSize: 1,
|
PartSize: 1,
|
||||||
}
|
}
|
||||||
downPath := "down.file" + time.Now().Format(time.RFC3339)
|
downPath := "down.file" + time.Now().Format(time.RFC3339)
|
||||||
|
defer os.Remove(downPath)
|
||||||
_, err = client.Object.Download(context.Background(), "test.go.download", downPath, opt)
|
_, err = client.Object.Download(context.Background(), "test.go.download", downPath, opt)
|
||||||
if err != nil {
|
if err == nil {
|
||||||
|
// 长度不一致 Failed
|
||||||
|
t.Fatalf("Object.Upload returned error: %v", err)
|
||||||
|
}
|
||||||
|
_, err = client.Object.Download(context.Background(), "test.go.download", downPath, opt)
|
||||||
|
if err == nil {
|
||||||
|
// CRC不一致
|
||||||
|
t.Fatalf("Object.Upload returned error: %v", err)
|
||||||
|
}
|
||||||
|
_, err = client.Object.Download(context.Background(), "test.go.download", downPath, opt)
|
||||||
|
if err != nil {
|
||||||
|
// 正确
|
||||||
t.Fatalf("Object.Upload returned error: %v", err)
|
t.Fatalf("Object.Upload returned error: %v", err)
|
||||||
}
|
}
|
||||||
os.Remove(downPath)
|
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user