multipart.go

package ks3

import (
	"bytes"
	"encoding/xml"
	"io"
	"net/http"
	"os"
	"sort"
	"strconv"
)

// InitiateMultipartUpload initializes multipart upload
//
// objectKey    object name
// options    the object constricts for upload. The valid options are CacheControl, ContentDisposition, ContentEncoding, Expires,
//            ServerSideEncryption, Meta, check out the following link:
//            https://help.ksyun.com/document_detail/ks3/api-reference/multipart-upload/InitiateMultipartUpload.html
//
// InitiateMultipartUploadResult    the return value of the InitiateMultipartUpload, which is used for calls later on such as UploadPartFromFile,UploadPartCopy.
// error    it's nil if the operation succeeds, otherwise it's an error object.
//
func (bucket Bucket) InitiateMultipartUpload(objectKey string, options ...Option) (InitiateMultipartUploadResult, error) {
	var imur InitiateMultipartUploadResult
	opts := AddContentType(options, objectKey)
	params, _ := GetRawParams(options)
	paramKeys := []string{"sequential", "withHashContext", "x-ks3-enable-md5", "x-ks3-enable-sha1", "x-ks3-enable-sha256"}
	ConvertEmptyValueToNil(params, paramKeys)
	params["uploads"] = nil

	resp, err := bucket.do("POST", objectKey, params, opts, nil, nil)
	if err != nil {
		return imur, err
	}
	defer resp.Body.Close()

	err = xmlUnmarshal(resp.Body, &imur)
	return imur, err
}

// UploadPart uploads parts
//
// After initializing a Multipart Upload, the upload Id and object key could be used for uploading the parts.
// Each part has its part number (ranges from 1 to 10,000). And for each upload Id, the part number identifies the position of the part in the whole file.
// And thus with the same part number and upload Id, another part upload will overwrite the data.
// Except the last one, minimal part size is 100KB. There's no limit on the last part size.
//
// imur    the returned value of InitiateMultipartUpload.
// reader    io.Reader the reader for the part's data.
// size    the part size.
// partNumber    the part number (ranges from 1 to 10,000). Invalid part number will lead to InvalidArgument error.
//
// UploadPart    the return value of the upload part. It consists of PartNumber and ETag. It's valid when error is nil.
// error    it's nil if the operation succeeds, otherwise it's an error object.
//
func (bucket Bucket) UploadPart(imur InitiateMultipartUploadResult, reader io.Reader,
	partSize int64, partNumber int, options ...Option) (UploadPart, error) {
	request := &UploadPartRequest{
		InitResult: &imur,
		Reader:     reader,
		PartSize:   partSize,
		PartNumber: partNumber,
	}

	result, err := bucket.DoUploadPart(request, options)

	return result.Part, err
}

// UploadPartFromFile uploads part from the file.
//
// imur    the return value of a successful InitiateMultipartUpload.
// filePath    the local file path to upload.
// startPosition    the start position in the local file.
// partSize    the part size.
// partNumber    the part number (from 1 to 10,000)
//
// UploadPart    the return value consists of PartNumber and ETag.
// error    it's nil if the operation succeeds, otherwise it's an error object.
//
func (bucket Bucket) UploadPartFromFile(imur InitiateMultipartUploadResult, filePath string,
	startPosition, partSize int64, partNumber int, options ...Option) (UploadPart, error) {
	var part = UploadPart{}
	fd, err := os.Open(filePath)
	if err != nil {
		return part, err
	}
	defer fd.Close()
	fd.Seek(startPosition, os.SEEK_SET)

	request := &UploadPartRequest{
		InitResult: &imur,
		Reader:     fd,
		PartSize:   partSize,
		PartNumber: partNumber,
	}

	result, err := bucket.DoUploadPart(request, options)

	return result.Part, err
}

// DoUploadPart does the actual part upload.
//
// request    part upload request
//
// UploadPartResult    the result of uploading part.
// error    it's nil if the operation succeeds, otherwise it's an error object.
//
func (bucket Bucket) DoUploadPart(request *UploadPartRequest, options []Option) (*UploadPartResult, error) {
	listener := GetProgressListener(options)
	options = append(options, ContentLength(request.PartSize))
	params := map[string]interface{}{}
	params["partNumber"] = strconv.Itoa(request.PartNumber)
	params["uploadId"] = request.InitResult.UploadID
	resp, err := bucket.do("PUT", request.InitResult.Key, params, options,
		&io.LimitedReader{R: request.Reader, N: request.PartSize}, listener)
	if err != nil {
		return &UploadPartResult{}, err
	}
	defer resp.Body.Close()

	part := UploadPart{
		ETag:       resp.Headers.Get(HTTPHeaderEtag),
		PartNumber: request.PartNumber,
		Crc64:      resp.Headers.Get(HTTPHeaderKs3CRC64),
	}

	if bucket.GetConfig().IsEnableCRC {
		err = CheckCRC(resp, "DoUploadPart")
		bucket.Client.Config.WriteLog(Info, "check file crc64, client crc:%d, server crc:%d", resp.ClientCRC, resp.ServerCRC)
		if err != nil {
			return &UploadPartResult{part}, err
		}
	}

	return &UploadPartResult{part}, nil
}

// UploadPartCopy uploads part copy
//
// imur    the return value of InitiateMultipartUpload
// copySrc    source Object name
// startPosition    the part's start index in the source file
// partSize    the part size
// partNumber    the part number, ranges from 1 to 10,000. If it exceeds the range KS3 returns InvalidArgument error.
// options    the constraints of source object for the copy. The copy happens only when these contraints are met. Otherwise it returns error.
//            CopySourceIfNoneMatch, CopySourceIfModifiedSince  CopySourceIfUnmodifiedSince, check out the following link for the detail
//            https://help.ksyun.com/document_detail/ks3/api-reference/multipart-upload/UploadPartCopy.html
//
// UploadPart    the return value consists of PartNumber and ETag.
// error    it's nil if the operation succeeds, otherwise it's an error object.
//
func (bucket Bucket) UploadPartCopy(imur InitiateMultipartUploadResult, srcBucketName, srcObjectKey string,
	startPosition, partSize int64, partNumber int, options ...Option) (UploadPart, error) {
	var out UploadPartCopyResult
	var part UploadPart
	var opts []Option

	//first find version id
	versionIdKey := "versionId"
	versionId, _ := FindOption(options, versionIdKey, nil)
	if versionId == nil {
		opts = []Option{CopySource(srcBucketName, srcObjectKey),
			CopySourceRange(startPosition, partSize)}
	} else {
		opts = []Option{CopySourceVersion(srcBucketName, srcObjectKey, versionId.(string)),
			CopySourceRange(startPosition, partSize)}
		options = DeleteOption(options, versionIdKey)
	}

	opts = append(opts, options...)

	params := map[string]interface{}{}
	params["partNumber"] = strconv.Itoa(partNumber)
	params["uploadId"] = imur.UploadID
	resp, err := bucket.do("PUT", imur.Key, params, opts, nil, nil)
	if err != nil {
		return part, err
	}
	defer resp.Body.Close()

	err = xmlUnmarshal(resp.Body, &out)
	if err != nil {
		return part, err
	}
	part.ETag = out.ETag
	part.PartNumber = partNumber
	part.Crc64 = out.Crc64

	return part, nil
}

// CompleteMultipartUpload completes the multipart upload.
//
// imur    the return value of InitiateMultipartUpload.
// parts    the array of return value of UploadPart/UploadPartFromFile/UploadPartCopy.
//
// CompleteMultipartUploadResponse    the return value when the call succeeds. Only valid when the error is nil.
// error    it's nil if the operation succeeds, otherwise it's an error object.
//
func (bucket Bucket) CompleteMultipartUpload(imur InitiateMultipartUploadResult,
	parts []UploadPart, options ...Option) (CompleteMultipartUploadResult, error) {
	var out CompleteMultipartUploadResult

	sort.Sort(UploadParts(parts))
	cxml := completeMultipartUploadXML{}
	cxml.Part = parts
	bs, err := xml.Marshal(cxml)
	if err != nil {
		return out, err
	}
	buffer := new(bytes.Buffer)
	buffer.Write(bs)

	params := map[string]interface{}{}
	params["uploadId"] = imur.UploadID
	resp, err := bucket.do("POST", imur.Key, params, options, buffer, nil)
	if err != nil {
		return out, err
	}
	defer resp.Body.Close()

	err = xmlUnmarshal(resp.Body, &out)
	return out, err
}

// AbortMultipartUpload aborts the multipart upload.
//
// imur    the return value of InitiateMultipartUpload.
//
// error    it's nil if the operation succeeds, otherwise it's an error object.
//
func (bucket Bucket) AbortMultipartUpload(imur InitiateMultipartUploadResult, options ...Option) error {
	params := map[string]interface{}{}
	params["uploadId"] = imur.UploadID
	resp, err := bucket.do("DELETE", imur.Key, params, options, nil, nil)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	return CheckRespCode(resp.StatusCode, []int{http.StatusNoContent})
}

// ListUploadedParts lists the uploaded parts.
//
// imur    the return value of InitiateMultipartUpload.
//
// ListUploadedPartsResponse    the return value if it succeeds, only valid when error is nil.
// error    it's nil if the operation succeeds, otherwise it's an error object.
//
func (bucket Bucket) ListUploadedParts(imur InitiateMultipartUploadResult, options ...Option) (ListUploadedPartsResult, error) {
	var out ListUploadedPartsResult

	params := map[string]interface{}{}
	params, err := GetRawParams(options)
	if err != nil {
		return out, err
	}

	params["uploadId"] = imur.UploadID
	resp, err := bucket.do("GET", imur.Key, params, options, nil, nil)
	if err != nil {
		return out, err
	}
	defer resp.Body.Close()

	err = xmlUnmarshal(resp.Body, &out)
	if err != nil {
		return out, err
	}
	err = decodeListUploadedPartsResult(&out)
	return out, err
}

// ListMultipartUploads lists all ongoing multipart upload tasks
//
// options    listObject's filter. Prefix specifies the returned object's prefix; KeyMarker specifies the returned object's start point in lexicographic order;
//            MaxKeys specifies the max entries to return; Delimiter is the character for grouping object keys.
//
// ListMultipartUploadResponse    the return value if it succeeds, only valid when error is nil.
// error    it's nil if the operation succeeds, otherwise it's an error object.
//
func (bucket Bucket) ListMultipartUploads(options ...Option) (ListMultipartUploadResult, error) {
	var out ListMultipartUploadResult

	params, err := GetRawParams(options)
	if err != nil {
		return out, err
	}
	params["uploads"] = nil

	resp, err := bucket.do("GET", "", params, options, nil, nil)
	if err != nil {
		return out, err
	}
	defer resp.Body.Close()

	err = xmlUnmarshal(resp.Body, &out)
	if err != nil {
		return out, err
	}
	err = decodeListMultipartUploadResult(&out)
	return out, err
}