Merge pull request #306 from tencentyun/feature_jojoliang_bbd92f3d

Feature jojoliang bbd92f3d

Author: agin719

example/object/put_from_url.go

@@ -0,0 +1,69 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"net/url"
+	"os"
+
+	"net/http"
+
+	"github.com/tencentyun/cos-go-sdk-v5"
+	"github.com/tencentyun/cos-go-sdk-v5/debug"
+)
+
+func logStatus(err error) {
+	if err == nil {
+		return
+	}
+	if cos.IsNotFoundError(err) {
+		// WARN
+		fmt.Println("WARN: Resource is not existed")
+	} else if e, ok := cos.IsCOSError(err); ok {
+		fmt.Printf("ERROR: Code: %v\n", e.Code)
+		fmt.Printf("ERROR: Message: %v\n", e.Message)
+		fmt.Printf("ERROR: Resource: %v\n", e.Resource)
+		fmt.Printf("ERROR: RequestId: %v\n", e.RequestID)
+		// ERROR
+	} else {
+		fmt.Printf("ERROR: %v\n", err)
+		// ERROR
+	}
+}
+
+func main() {
+	// 存储桶名称，由bucketname-appid 组成，appid必须填入，可以在COS控制台查看存储桶名称。 https://console.cloud.tencent.com/cos5/bucket
+	// 替换为用户的 region，存储桶region可以在COS控制台“存储桶概览”查看 https://console.cloud.tencent.com/ ，关于地域的详情见 https://cloud.tencent.com/document/product/436/6224 。
+	u, _ := url.Parse("http://test-1259654469.cos.ap-guangzhou.myqcloud.com")
+	b := &cos.BaseURL{BucketURL: u}
+	c := cos.NewClient(b, &http.Client{
+		Transport: &cos.AuthorizationTransport{
+			// 通过环境变量获取密钥
+			// 环境变量 SECRETID 表示用户的 SecretId，登录访问管理控制台查看密钥，https://console.cloud.tencent.com/cam/capi
+			SecretID: os.Getenv("SECRETID"),
+			// 环境变量 SECRETKEY 表示用户的 SecretKey，登录访问管理控制台查看密钥，https://console.cloud.tencent.com/cam/capi
+			SecretKey: os.Getenv("SECRETKEY"),
+			// Debug 模式，把对应 请求头部、请求内容、响应头部、响应内容 输出到标准输出
+			Transport: &debug.DebugRequestTransport{
+				RequestHeader: true,
+				// Notice when put a large file and set need the request body, might happend out of memory error.
+				RequestBody:    false,
+				ResponseHeader: true,
+				ResponseBody:   true,
+			},
+		},
+	})
+
+	// Case1 上传对象
+	name := "example"
+	opt := &cos.ObjectPutFromURLOptions{
+		PartSize: 1,
+		InitOptions: &cos.InitiateMultipartUploadOptions{
+			ACLHeaderOptions: &cos.ACLHeaderOptions{
+				XCosACL: "private",
+			},
+		},
+	}
+	_, _, err := c.Object.PutFromURL(context.Background(), name, "https://test-1259654469.cos.ap-guangzhou.myqcloud.com/pub", opt)
+	logStatus(err)
+}

object.go

@@ -1991,3 +1991,192 @@ func (s *ObjectService) GetSymlink(ctx context.Context, name string, opt *Object
 	}
 	return resp.Header.Get("x-cos-symlink-target"), resp, err
 }
+
+type ObjectPutFromURLOptions struct {
+	PartSize    int
+	QueueSize   int
+	InitOptions *InitiateMultipartUploadOptions
+}
+
+func (s *ObjectService) PutFromURL(ctx context.Context, name string, downloadURL string, opt *ObjectPutFromURLOptions) (*CompleteMultipartUploadResult, *Response, error) {
+	if opt == nil {
+		opt = &ObjectPutFromURLOptions{}
+	}
+	// init
+	v, resp, err := s.InitiateMultipartUpload(ctx, name, opt.InitOptions)
+	if err != nil {
+		return nil, resp, err
+	}
+	uploadId := v.UploadID
+	var isErr bool
+	defer func() {
+		if isErr {
+			s.AbortMultipartUpload(ctx, name, uploadId, nil)
+		}
+	}()
+	// request from url
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, downloadURL, nil)
+	if err != nil {
+		isErr = true
+		return nil, nil, err
+	}
+	rsp, err := http.DefaultClient.Do(req)
+	if err != nil || rsp == nil {
+		isErr = true
+		return nil, nil, err
+	}
+	defer rsp.Body.Close()
+	if rsp.StatusCode > 299 {
+		isErr = true
+		return nil, &Response{rsp}, fmt.Errorf("the status code of downloadURL response is failed: %d", rsp.StatusCode)
+	}
+	factory := newPartFactory(opt.PartSize, opt.QueueSize)
+	partChannel, errChannel := factory.Produce(rsp.Body)
+	defer factory.Close()
+
+	comOpt := &CompleteMultipartUploadOptions{}
+	var partNumber int
+	for {
+		select {
+		case part, ok := <-partChannel:
+			if !ok {
+				partChannel = nil
+				break
+			}
+			partNumber++
+			resp, err := s.UploadPart(ctx, name, uploadId, partNumber, part, nil)
+			if err != nil {
+				isErr = true
+				return nil, resp, err
+			}
+			comOpt.Parts = append(comOpt.Parts, Object{
+				PartNumber: partNumber,
+				ETag:       resp.Header.Get("ETag"),
+			})
+		case err, ok := <-errChannel:
+			if !ok {
+				errChannel = nil
+				break
+			}
+			if err != nil {
+				isErr = true
+				return nil, nil, err
+			}
+		}
+		if partChannel == nil && errChannel == nil {
+			break
+		}
+	}
+	res, resp, err := s.CompleteMultipartUpload(ctx, name, uploadId, comOpt)
+	if err != nil {
+		isErr = true
+	}
+	return res, resp, err
+}
+
+type partFactory struct {
+	partSize      int
+	queueSize     int
+	current       *bytes.Buffer
+	partChannel   chan *bytes.Buffer
+	errChannel    chan error
+	cancelChannel chan struct{}
+}
+
+const CHUNK_SIZE = 1024 * 1024
+
+func newPartFactory(partSize int, queueSize int) *partFactory {
+	if partSize <= 0 {
+		partSize = 8
+	}
+	if queueSize <= 0 {
+		queueSize = 10
+	}
+	return &partFactory{
+		partSize:  partSize * 1024 * 1024,
+		queueSize: queueSize,
+		current:   bytes.NewBuffer(nil),
+	}
+}
+
+func (pf *partFactory) Produce(reader io.ReadCloser) (<-chan *bytes.Buffer, <-chan error) {
+	pf.cancelChannel = make(chan struct{}, 1)
+	pf.partChannel = make(chan *bytes.Buffer, pf.queueSize)
+	pf.errChannel = make(chan error, 1)
+
+	go pf.Run(reader)
+	return pf.partChannel, pf.errChannel
+}
+
+func (pf *partFactory) Close() {
+	pf.cancelChannel <- struct{}{}
+}
+
+func (pf *partFactory) Run(reader io.ReadCloser) {
+	var total, parts int
+	defer func() {
+		close(pf.errChannel)
+		close(pf.partChannel)
+	}()
+	buf := make([]byte, CHUNK_SIZE)
+	for {
+		select {
+		case <-pf.cancelChannel:
+			return
+		default:
+			n, err := reader.Read(buf)
+			total += n
+			if n > 0 {
+				part, e := pf.Write(buf[:n])
+				if e != nil {
+					pf.errChannel <- e
+					return
+				}
+				if part != nil {
+					parts++
+					select {
+					case pf.partChannel <- part:
+					case <-pf.cancelChannel:
+						return
+					}
+				}
+			}
+			if err != nil && err != io.EOF {
+				pf.errChannel <- err
+				return
+			}
+			if err == io.EOF || n == 0 {
+				if pf.current.Len() > 0 {
+					parts++
+					select {
+					case pf.partChannel <- pf.current:
+					case <-pf.cancelChannel:
+						return
+					}
+				}
+				return
+			}
+		}
+	}
+}
+
+func (pf *partFactory) Write(p []byte) (*bytes.Buffer, error) {
+	var res *bytes.Buffer
+	for nwrite := 0; nwrite < len(p); {
+		if pf.current.Len() == pf.partSize {
+			res = pf.current
+			pf.current = bytes.NewBuffer(nil)
+		}
+		end := len(p)
+		// 大于缓存区大小
+		if pf.current.Len()+end-nwrite > pf.partSize {
+			end = nwrite + pf.partSize - pf.current.Len()
+		}
+		nr, err := pf.current.Write(p[nwrite:end])
+		if err != nil {
+			return res, err
+		}
+		nwrite += nr
+	}
+	return res, nil
+}